Instrument Control

Unit-Testing Protocol Abstraction Layers with Mock Instruments

Instrument-control code that can only be exercised with the physical rig plugged in is code that is never tested until it fails a run. A mock instrument breaks that dependency: a fake transport that satisfies the same write/read/query contract as the real bus, driven by a scripted request-to-response table so your Protocol Abstraction Layers run against a deterministic SCPI state machine in CI. This guide builds that mock, wires it into pytest fixtures, injects faults the hardware only produces once a year, and runs one contract suite against both the mock and — when available — the live instrument.

Scope: Substituting the Transport, Not the Adapter

The unit under test is the abstraction layer itself — the adapter that serializes a Command, drives a transport, and hands back verified frames. The scope here is narrow: replace only the lowest tier, the byte-level transport, and leave every layer above it running exactly as it does in production. A mock that stubs out the adapter’s own query() method tests nothing; a mock that implements the transport’s write(data, timeout) / read(max_bytes, timeout) / close() interface tests the entire serialization, framing, timeout, and error-classification path with only the silicon removed. That boundary is the same TransportLayer protocol a serial bridge, raw socket, or VISA Resource Manager session satisfies, so the adapter cannot tell it is talking to a fake.

Two assumptions hold throughout. First, the instrument is request-response and deterministic: a given command in a given device state yields one defined reply, which is what makes a scripted table faithful. Second, the mock is not a second implementation of the protocol — it is a lookup table plus a small state machine for the handful of commands that mutate state (*RST, an output-enable toggle, the SYST:ERR? queue). Everything runs on Python 3.11+ with pytest >= 7.4; the mock itself depends only on the standard library. The goal is a suite fast enough to run on every commit and honest enough that a green suite means the adapter will behave on the bench.

A Scripted Command Table and a State-Transition Coverage Metric

The mock’s behaviour is a function from (device_state, command) → (reply, next_state). Most SCPI commands are stateless queries — *IDN?, MEAS:VOLT? — and collapse to a flat table. The few that mutate state form the interesting part, and the risk in a mock is that you script the happy path and never exercise the transitions the adapter’s recovery logic depends on. Quantify that with a state-transition coverage ratio: over the set of reachable (state, command) pairs your device model admits, count how many the test suite actually drives.

A suite that queries MEAS:VOLT? a hundred times but never issues it while the mock is in its ERROR state has a high assertion count and a low C_trans. Tracking the ratio makes the untested transitions visible — for a model with states {READY, MEASURING, ERROR} and a dozen commands, full pairwise coverage is a small, enumerable target, and the gap between covered and reachable pairs is exactly the list of tests still to write. Determinism is the other design rule: the mock must resolve every request to the same reply on every run, with no wall-clock dependency and no hidden ordering, so a failure reproduces from the command log alone rather than only on the machine where it first appeared.

Contract suite driving one adapter against mock and real transports On the left, a single contract test suite calls the abstraction-layer adapter through the canonical write/read/query transport contract. A pytest fixture parameter selects which transport the adapter binds to. The upper branch is a MockInstrument: a scripted request-to-response command table plus a small SCPI state machine and fault-injection hooks that can force timeouts, malformed frames, and error-queue replies. The lower branch is the real instrument reached over VISA or serial. Because both transports satisfy the same contract, the identical suite validates both, and any divergence between mock and firmware surfaces as a failing contract test. ONE SUITE · TWO TRANSPORTS · SAME CONTRACT Contract test suite pytest, one file Adapter under test serialize · frame · classify fixture selects transport write/read/query MockInstrument scripted table · SCPI state machine + fault-injection hooks Real instrument VISA / serial transport optional · marked slow / hardware
One contract suite drives the adapter through the shared transport interface; a fixture parameter swaps the mock (with fault hooks) for the real bus. Identical assertions on both mean a mock divergence shows up as a red test, not a silent lab surprise.

Building a MockInstrument Transport with Fault Hooks

The mock below satisfies the same write/read/query contract the production adapter expects. It holds a command table, a tiny state machine, a SYST:ERR? error queue, and a list of injectable faults that fire in FIFO order so a test can script “the third read times out, then recovers”.

from __future__ import annotations

import re
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Optional


class InstrumentTimeoutError(Exception):
    """Raised by the mock to emulate a transport-level read timeout."""


@dataclass
class Fault:
    """One scripted fault. `kind` selects the failure the next read emits."""
    kind: str            # "timeout" | "malformed" | "error_queue"
    payload: bytes = b""  # bytes to return for a "malformed" fault


@dataclass
class MockInstrument:
    """A fake transport driving a scripted SCPI state machine.

    Implements the same write/read/query surface as a serial or VISA
    transport, so an abstraction-layer adapter binds to it unchanged.
    """
    idn: str = "MockCorp,DMM-1000,SN0001,1.4.2"
    state: str = "READY"
    _table: dict[str, Callable[["MockInstrument"], bytes]] = field(default_factory=dict)
    _errors: Deque[str] = field(default_factory=deque)
    _faults: Deque[Fault] = field(default_factory=deque)
    _pending: bytes = b""

    def __post_init__(self) -> None:
        self._table = {
            r"\*IDN\?": lambda s: s.idn.encode() + b"\n",
            r"\*RST": self._reset,
            r"MEAS:VOLT\?": self._measure_voltage,
            r"SYST:ERR\?": self._pop_error,
        }

    # -- fault scripting -----------------------------------------------------
    def inject(self, fault: Fault) -> None:
        """Queue a fault to fire on subsequent reads (FIFO)."""
        self._faults.append(fault)

    def push_error(self, code: int, text: str) -> None:
        self._errors.append(f'{code},"{text}"')

    # -- transport contract --------------------------------------------------
    def write(self, data: bytes, timeout: float) -> None:
        cmd = data.decode("ascii", "replace").strip()
        if self._faults and self._faults[0].kind == "error_queue":
            self._faults.popleft()
            self.state = "ERROR"
            self.push_error(-113, "Undefined header")
            self._pending = b""
            return
        for pattern, handler in self._table.items():
            if re.fullmatch(pattern, cmd):
                self._pending = handler(self)
                return
        self.state = "ERROR"
        self.push_error(-113, "Undefined header")
        self._pending = b""

    def read(self, max_bytes: int, timeout: float) -> bytes:
        if self._faults:
            fault = self._faults[0]
            if fault.kind == "timeout":
                self._faults.popleft()
                raise InstrumentTimeoutError("mock: injected read timeout")
            if fault.kind == "malformed":
                self._faults.popleft()
                return fault.payload  # e.g. truncated frame, wrong terminator
        chunk, self._pending = self._pending[:max_bytes], self._pending[max_bytes:]
        return chunk

    def query(self, data: bytes, timeout: float) -> bytes:
        self.write(data, timeout)
        return self.read(4096, timeout)

    def close(self) -> None:
        self._pending = b""

    # -- state machine handlers ---------------------------------------------
    def _reset(self, _s: "MockInstrument") -> bytes:
        self.state = "READY"
        self._errors.clear()
        return b""

    def _measure_voltage(self, _s: "MockInstrument") -> bytes:
        if self.state != "READY":
            self.push_error(-221, "Settings conflict")
            return b"9.91E+37\n"   # SCPI NaN sentinel
        return b"5.001200E+00\n"

    def _pop_error(self, _s: "MockInstrument") -> bytes:
        if self._errors:
            return self._errors.popleft().encode() + b"\n"
        return b'0,"No error"\n'

The load-bearing detail is that faults are queued and consumed, not toggled by a flag. A test that needs “two timeouts then success” appends two Fault("timeout") entries and the adapter’s retry loop drains them naturally, exercising exactly the transient-versus-deterministic split that Timeout Handling & Retry Logic defines — without waiting on real backoff delays.

Pytest Fixtures and Contract Tests Across Mock and Hardware

A single parametrized fixture yields the transport, so the same test body runs against the mock always and the real rig when a hardware marker is enabled. The mock case runs in CI; the hardware case runs on a bench with --hardware.

import os
import pytest


@pytest.fixture(params=["mock", pytest.param("real", marks=pytest.mark.hardware)])
def transport(request):
    """Yield a transport satisfying the write/read/query contract.

    The identical contract suite runs against the mock in CI and, when
    invoked with --hardware on a bench, against the live instrument.
    """
    if request.param == "mock":
        yield MockInstrument()
    else:
        resource = os.environ["DMM_RESOURCE"]
        real = open_visa_transport(resource)  # your VISA-backed adapter transport
        try:
            yield real
        finally:
            real.close()


def test_idn_roundtrips(transport):
    """The adapter parses a vendor,model,serial,firmware identity string."""
    reply = transport.query(b"*IDN?\n", timeout=2.0).decode().strip()
    assert reply.count(",") == 3


def test_measure_returns_float(transport):
    reply = transport.query(b"MEAS:VOLT?\n", timeout=2.0).decode().strip()
    assert float(reply) == pytest.approx(5.0, abs=0.5)


def test_injected_timeout_raises(transport):
    """A transient read timeout must surface as InstrumentTimeoutError."""
    if not isinstance(transport, MockInstrument):
        pytest.skip("fault injection is mock-only")
    transport.inject(Fault(kind="timeout"))
    with pytest.raises(InstrumentTimeoutError):
        transport.query(b"MEAS:VOLT?\n", timeout=0.5)


def test_error_queue_is_drained(transport):
    """An undefined header lands a -113 in SYST:ERR? and clears afterward."""
    transport.query(b"BOGUS:CMD\n", timeout=1.0)
    err = transport.query(b"SYST:ERR?\n", timeout=1.0).decode()
    assert err.startswith("-113")
    nxt = transport.query(b"SYST:ERR?\n", timeout=1.0).decode()
    assert nxt.startswith("0")

The verification is the suite itself: test_idn_roundtrips asserts the query round-trip, test_injected_timeout_raises asserts the timeout path raises rather than returning empty bytes, and test_error_queue_is_drained asserts the SYST:ERR? handshake that couples the mock to Error Code Categorization. Because every test takes transport and never references the concrete class, running with --hardware re-executes the same assertions against firmware — the strongest available evidence that the mock and the instrument agree. To keep the mock honest over time, capture a real session once (log every write and the bytes each read returned), store it as a fixture, and replay it: a record-and-replay test feeds the captured replies through the adapter and asserts the decoded results match, so a firmware change that alters a reply format fails the replay before it reaches production. Those format shifts are the concern of Versioning Protocol Adapters Across Firmware Revisions, and a captured session is the artifact that ties a mock to a specific revision.

Failure Modes Unique to Mock-Based Testing

These faults are specific to testing an abstraction against a fake instrument, distinct from the framing faults the parser itself must survive.

Mock diverging from real firmware. The mock encodes your belief about the instrument, and beliefs drift. A firmware update changes MEAS:VOLT? from 5.0012E+00 to a fixed-width +5.001200, the mock still returns the old form, and the suite stays green while the bench fails. Diagnose by scheduling the --hardware run — even weekly — and by driving the mock from recorded sessions rather than hand-typed strings. When the replay test and the mock disagree, the recording wins, because it came from the wire.

Over-mocking that hides integration bugs. Stubbing the adapter’s own query() instead of the transport tests the mock, not the code. The serialization, the terminator handling, the SCPI Command-Set Standardization mapping — all of it is bypassed, so a bug in Command.serialize() sails through a fully green suite. Keep the seam at the transport boundary: the mock receives bytes and returns bytes, and every layer that turns a method call into those bytes stays under test.

Non-deterministic fixture ordering. A module-scoped mock whose error queue or fault list carries state between tests makes the suite order-dependent — test_error_queue_is_drained passes alone and fails after a test that left an entry in _errors. Diagnose with pytest -p no:randomly versus pytest-randomly: a suite that only passes in file order is leaking state. Default to function-scoped fixtures so each test gets a pristine instrument, and reset the queues explicitly if you must share one for speed.

Forgetting the error-queue path. The happy-path table is easy; the SYST:ERR? drain after a rejected command is the branch that actually exercises recovery, and it is the one most often left uncovered. A mock that never pushes an error and a suite that never reads one leave the adapter’s error-classification code entirely untested, which is precisely the code you need working when the instrument faults mid-run. Track it through the state-transition coverage ratio above: an ERROR state with no inbound transition in the test set is the metric flagging the gap.

← Back to Protocol Abstraction Layers

References