alicatlib.testing¶

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

    The convenience for test code is that one line replaces the
    boilerplate "parse fixture, construct transport". The returned
    transport is *not* open — the caller awaits ``.open()`` as usual —
    because :class:`FakeTransport`'s construction is synchronous but
    opening isn't.

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

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

    The returned dict maps ``send_bytes → reply_bytes``. Both payloads
    are ASCII-encoded with a trailing ``\r`` on every logical line — the
    exact shape :class:`FakeTransport` expects.

    Raises:
        ValueError: On malformed lines, a ``<`` before any ``>``, or a
            duplicate ``>`` entry. Every error message names the
            offending line number in the source file.
        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 = []

    # File is read as UTF-8 so comments can include non-ASCII characters.
    # Individual ``>`` / ``<`` payloads are encoded back to ASCII at the
    # byte-emit step below — any non-ASCII payload surfaces as UnicodeEncodeError
    # at that point, which is the right behavior (Alicat wire is ASCII-only).
    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()
            payload = _content_after_marker(line, ">")
            current_send = payload.encode("ascii") + b"\r"
        elif lean.startswith("<"):
            if current_send is None:
                raise ValueError(
                    f"{fixture_path}:{line_number}: '<' line without preceding '>'",
                )
            payload = _content_after_marker(line, "<")
            current_reply_chunks.append(payload.encode("ascii") + b"\r")
        else:
            raise ValueError(
                f"{fixture_path}:{line_number}: unrecognized line {line!r}; "
                f"lines must start with '>', '<', or '#'",
            )
    _flush()
    return script