watlowlib.testing¶

async def controller_from_fixture(
    path: str | Path,
    *,
    family: ControllerFamily = ControllerFamily.PM,
) -> Controller:
    """Build an unopened :class:`Controller` scripted by ``path``.

    Returned in unopened form so the caller drives lifecycle through
    ``async with``. Std Bus fixtures wire through :class:`FakeTransport`;
    Modbus fixtures wire through :class:`FakeSlave` (the
    :class:`Transport` shim only carries lifecycle, since the Modbus
    protocol client talks to its slave provider directly).
    """
    fixture = load_fixture(path)
    if fixture.protocol is ProtocolKind.STDBUS:
        transport: Transport = fixture.fake_transport()
        return await open_controller(
            transport,
            protocol=ProtocolKind.STDBUS,
            address=fixture.address,
            serial_settings=fixture.serial_settings,
            family=family,
        )

    # Modbus path: build the protocol client over a FakeSlave directly,
    # and hand the controller a lifecycle-only :class:`FakeTransport`.
    slave = fixture.fake_slave()
    transport = FakeTransport(label="fake://fixture-modbus")
    # FakeSlave is structurally compatible with the
    # :class:`SlaveLike` Protocol the Modbus client expects.
    client = ModbusProtocolClient(
        slave_provider=lambda _addr: slave,
        port=transport.label,
    )
    session = Session(
        client,
        registry=PARAMETERS,
        family=family,
        address=fixture.address,
        port=transport.label,
    )
    return Controller(session, transport, serial_settings=fixture.serial_settings)

def load_fixture(path: str | Path) -> Fixture:
    """Parse ``path`` as a JSONL fixture and return a :class:`Fixture`.

    The first line may be a header (``{"kind": "header", ...}``) that
    sets ``address`` and the serial framing for the capture. Subsequent
    lines are rounds; their ``protocol`` field must agree with the
    header (or with the first round, if the header is omitted).

    Raises:
        ValueError: malformed JSONL, missing fields, or a row whose
            ``protocol`` field disagrees with the rest of the file.
    """
    p = Path(path)
    rows: list[dict[str, Any]] = []
    for line_no, raw in enumerate(p.read_text(encoding="utf-8").splitlines(), start=1):
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        try:
            rows.append(json.loads(line))
        except json.JSONDecodeError as exc:
            raise ValueError(f"{p}:{line_no}: not valid JSON: {exc}") from exc
    if not rows:
        raise ValueError(f"{p}: fixture is empty")

    header: dict[str, Any] = {}
    body: list[dict[str, Any]] = list(rows)
    if rows[0].get("kind") == "header":
        header = rows[0]
        body = rows[1:]

    protocol = _resolve_protocol(header, body, source=p)

    address = int(header.get("address", 1) or 1)
    serial_settings = _settings_from_header(header)

    if protocol is ProtocolKind.STDBUS:
        stdbus_rounds = tuple(_parse_stdbus_row(row, p, idx) for idx, row in enumerate(body, 2))
        return Fixture(
            protocol=protocol,
            address=address,
            serial_settings=serial_settings,
            stdbus_rounds=stdbus_rounds,
        )

    modbus_rounds = tuple(_parse_modbus_row(row, p, idx) for idx, row in enumerate(body, 2))
    return Fixture(
        protocol=protocol,
        address=address,
        serial_settings=serial_settings,
        modbus_rounds=modbus_rounds,
    )