watlowlib.testing¶

def FakeTransportFromArrowFixture(  # noqa: N802 — public factory, title-case matches the class it returns
    path: str | Path,
    *,
    label: str | None = None,
) -> FakeTransport:
    """Load a plaintext-arrow fixture into a built :class:`FakeTransport`.

    Convenience wrapper around :func:`parse_arrow_fixture` plus
    :class:`FakeTransport` construction. The returned transport is
    not opened — the caller awaits ``.open()`` as usual.

    Args:
        path: Path to the ``.txt`` fixture.
        label: Optional override for :attr:`FakeTransport.label`;
            defaults to ``"fixture://<basename>"`` so error contexts
            point at the actual fixture during failures.
    """
    script = parse_arrow_fixture(path)
    resolved_label = label if label is not None else f"fixture://{Path(path).name}"
    return FakeTransport(script, label=resolved_label)

async def controller_from_fixture(
    path: str | Path,
    *,
    profile: DeviceProfile = EZZONE_PROFILE,
) -> 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).

    Args:
        path: JSONL fixture path.
        profile: Device profile to decode against. Defaults to
            :data:`~watlowlib.devices.profile.EZZONE_PROFILE`; pass
            :data:`~watlowlib.devices.profile.SERIES_SD_PROFILE` so an
            SD fixture decodes against ``SD_PARAMETERS``.
    """
    fixture = load_fixture(path)
    if fixture.protocol is ProtocolKind.STDBUS:
        transport: Transport = fixture.fake_transport()
        return await _open_fixture_controller(
            transport,
            protocol=ProtocolKind.STDBUS,
            address=fixture.address,
            serial_settings=fixture.serial_settings,
            profile=profile,
        )

    # 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,
        profile=profile,
        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,
    )

async def open_test_controller(
    transport: Transport,
    *,
    protocol: ProtocolKind = ProtocolKind.STDBUS,
    address: int = 1,
    serial_settings: SerialSettings | None = None,
    profile: DeviceProfile = EZZONE_PROFILE,
    wire_temperature_unit: Unit | None = None,
) -> Controller:
    """Build an opened :class:`Controller` over an existing :class:`Transport`.

    Test-surface companion to :func:`watlowlib.open_device`. Production
    code always goes through :func:`open_device` (which opens a real
    serial port and runs auto-detect when asked); tests drive the
    facade through a :class:`FakeTransport` and use this helper to
    skip the serial-port plumbing.

    Args:
        transport: The transport to wire the controller to. Typically
            a :class:`FakeTransport` scripted with captured round-
            trips.
        protocol: Wire protocol. ``AUTO`` is rejected — auto-detection
            is a real-port-only path on :func:`open_device`.
        address: Bus address.
        serial_settings: Optional override; defaults to a
            ``fake://test`` placeholder so call sites that don't care
            about framing stay terse.
        profile: Device profile to decode against. Defaults to
            :data:`~watlowlib.devices.profile.EZZONE_PROFILE`; pass
            :data:`~watlowlib.devices.profile.SERIES_SD_PROFILE` to
            decode an SD fixture against ``SD_PARAMETERS``.
        wire_temperature_unit: Already-coerced :class:`Unit` to drive
            :class:`Reading.unit` / :class:`Sample.unit` for
            temperature parameters. Tests usually leave this ``None``.

    Returns:
        An opened :class:`Controller`. The caller is responsible for
        closing it (typically via ``async with``).
    """
    settings = serial_settings or SerialSettings(port=transport.label)
    return await _open_fixture_controller(
        transport,
        protocol=protocol,
        address=address,
        serial_settings=settings,
        profile=profile,
        wire_temperature_unit=wire_temperature_unit,
    )

def parse_arrow_fixture(path: str | Path) -> dict[bytes, bytes]:
    r"""Parse a plaintext-arrow fixture into a :class:`FakeTransport` script.

    The fixture format is intentionally human-skimmable so captured
    Std Bus traffic round-trips through code review (cross-package
    convention shared with :mod:`alicatlib.testing` and
    :mod:`sartoriuslib.testing`)::

        # scenario: read_pv (PM3, parameter 4001)
        > 55 FF 05 10 00 00 06 E8 01 03 01 04 01 01 E3 99
        < 55 FF 06 00 10 00 0B 88 02 03 01 04 01 01 08 45 1E 3C D4 A7 28

    Parsing rules:

    - Lines starting with ``#`` are comments; ignored.
    - Blank lines are ignored.
    - ``>`` introduces a request — bytes after the marker are decoded
      as space-separated hex. Whitespace within a payload is ignored
      so callers can group bytes for readability.
    - ``<`` introduces one reply — same hex encoding. Multiple ``<``
      lines after a single ``>`` concatenate into one scripted reply
      (useful when a logical reply was captured across multiple
      reads).
    - Duplicate ``>`` entries are a fixture error rather than a
      silent overwrite.

    Returns:
        Mapping ``request_bytes → reply_bytes`` ready to feed
        :class:`FakeTransport`.

    Raises:
        ValueError: On malformed lines, a ``<`` before any ``>``, or
            a duplicate ``>`` entry. Every error message names the
            offending line number.
        FileNotFoundError: Via the underlying :meth:`Path.read_text`.
    """
    fixture_path = Path(path)
    script: dict[bytes, bytes] = {}
    current_send: bytes | None = None
    current_reply_chunks: list[bytes] = []

    def _flush() -> None:
        nonlocal current_send, current_reply_chunks
        if current_send is None:
            return
        if current_send in script:
            raise ValueError(
                f"{fixture_path}: duplicate send entry {current_send!r}",
            )
        script[current_send] = b"".join(current_reply_chunks)
        current_send = None
        current_reply_chunks = []

    text = fixture_path.read_text(encoding="utf-8")
    for line_number, line in _iter_semantic_lines(text.splitlines()):
        lean = line.lstrip()
        if lean.startswith(">"):
            _flush()
            current_send = _hex_payload(
                line,
                ">",
                source=fixture_path,
                line_no=line_number,
            )
        elif lean.startswith("<"):
            if current_send is None:
                raise ValueError(
                    f"{fixture_path}:{line_number}: '<' line without preceding '>'",
                )
            current_reply_chunks.append(
                _hex_payload(line, "<", source=fixture_path, line_no=line_number),
            )
        else:
            raise ValueError(
                f"{fixture_path}:{line_number}: unrecognized line {line!r}; "
                f"lines must start with '>', '<', or '#'",
            )
    _flush()
    return script