`servomexlib.testing`¶

FakeTransport, fixture loaders, the MockSlave-backed Modbus fake, and canned frames.

servomexlib.testing ¶

Public testing seam.

Re-exports the in-process :class:FakeTransport plus fixture helpers so tests — in this package and in downstream code — can drive every protocol without hardware. Fixtures use the human-readable > hex / < hex arrow format shared with the sibling .testing modules.

For the Modbus path we do not hand-roll an ADU simulator: the :func:mock_modbus_pair helper preloads the byte-accurate anymodbus.testing.MockSlave (RTU and ASCII framing, FC01/02/04/05/08) with the 4100's register/coil banks and binds our ModbusClient to it over an in-process serial pair. The Modbus imports are lazy so servomexlib.testing stays importable without the optional [modbus] extra.

CoilOp `dataclass` ¶

CoilOp(function_code, coil, on)

One observed coil operation against a recording :class:MockSlave.

Used by the autocalibration coil-pulse tests to assert the ordered 0→1→0 transition: a start_calibration pulse should record a WRITE_SINGLE_COIL on=True (FC05), then a READ_COILS readback (FC01), then a WRITE_SINGLE_COIL on=False. coil is the PDU address.

FakeTransport ¶

FakeTransport(script=None, *, label='fake://test')

Bases: ByteStreamTransport

Scripted, in-process :class:ByteStreamTransport.

Source code in src/servomexlib/transport/fake.py

def __init__(
    self,
    script: Mapping[bytes, ScriptedReply] | None = None,
    *,
    label: str = "fake://test",
) -> None:
    super().__init__(label=label)
    self._script: dict[bytes, ScriptedReply] = dict(script or {})
    self._writes: list[bytes] = []
    self._inbound = bytearray()
    self._event = anyio.Event()
    self._closed = False

writes `property` ¶

writes

Every payload written through :meth:send, in order.

add_script ¶

add_script(request, reply)

Register or overwrite the canned reply for request.

Source code in src/servomexlib/transport/fake.py

def add_script(self, request: bytes, reply: ScriptedReply) -> None:
    """Register or overwrite the canned reply for ``request``."""
    self._script[bytes(request)] = reply

emit `async` ¶

emit(frames, *, interval=0.0)

Feed frames one at a time, optionally interval seconds apart.

Run as a background task to simulate a continuous broadcaster: task_group.start_soon(lambda: fake.emit(frames, interval=0.05)).

Source code in src/servomexlib/transport/fake.py

async def emit(self, frames: Sequence[bytes], *, interval: float = 0.0) -> None:
    """Feed ``frames`` one at a time, optionally ``interval`` seconds apart.

    Run as a background task to simulate a continuous broadcaster:
    ``task_group.start_soon(lambda: fake.emit(frames, interval=0.05))``.
    """
    for frame in frames:
        if interval:
            await anyio.sleep(interval)
        self.feed(frame)

feed ¶

feed(data)

Push unsolicited bytes into the inbound buffer.

Source code in src/servomexlib/transport/fake.py

def feed(self, data: bytes) -> None:
    """Push unsolicited bytes into the inbound buffer."""
    self._inbound += data
    self._wake()

MockChannel `dataclass` ¶

MockChannel(
    channel,
    value,
    name=b"\x00\x00\x00\x00\x00\x00",
    unit=b"\x00\x00\x00",
    status_bits=(False,) * 8,
)

One channel's contents to preload into a :class:MockSlave's banks.

name / unit are raw display-ROM bytes (not ASCII), so a test can reproduce e.g. CO₂'s name 43 4F 82 20 20 20 and exercise the charset decode. status_bits are the 8 FC02 discrete bits (Fault, Maintenance, Calibration, WarmingUp, Alarm1..4).

ReadOp `dataclass` ¶

ReadOp(function_code, address, count)

One observed read request against a recording :class:MockSlave.