104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
"""Replay a captured /dev/iec0 record stream and check the produced trace.
|
|
|
|
End-to-end Phase 0 test: build the canonical session, round-trip it through the
|
|
binary wire format, and assert the formatted trace matches the documented PoC
|
|
output (PLAN.md §8).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
|
|
from iecpoc import device
|
|
from iecpoc.log import TraceFormatter
|
|
|
|
from . import fixtures
|
|
|
|
|
|
def _trace(records, my_address=4, raw=False):
|
|
return list(TraceFormatter(my_address=my_address, raw=raw).format_stream(records))
|
|
|
|
|
|
# --- wire-format round trip -------------------------------------------------
|
|
def test_record_roundtrip():
|
|
rec = device.IecRecord(device.KIND_DATA, 0x48, device.FLAG_EOI, 123456789)
|
|
assert device.IecRecord.from_bytes(rec.to_bytes()) == rec
|
|
|
|
|
|
def test_iter_records_roundtrip():
|
|
records = fixtures.build_session()
|
|
buf = io.BytesIO()
|
|
device.write_records(buf, records)
|
|
buf.seek(0)
|
|
assert list(device.iter_records(buf)) == records
|
|
|
|
|
|
def test_iter_records_rejects_truncated():
|
|
buf = io.BytesIO(b"\x00\x01\x02") # < RECORD_SIZE
|
|
import pytest
|
|
|
|
with pytest.raises(ValueError):
|
|
list(device.iter_records(buf))
|
|
|
|
|
|
# --- captured binary fixture matches the in-memory builder ------------------
|
|
def test_capture_file_matches_builder():
|
|
with open(fixtures.CAPTURE_PATH, "rb") as fh:
|
|
from_file = list(device.iter_records(fh))
|
|
assert from_file == fixtures.build_session()
|
|
|
|
|
|
# --- trace formatting -------------------------------------------------------
|
|
def test_trace_contains_plan_section8_lines():
|
|
trace = _trace(fixtures.build_session())
|
|
|
|
# the §8 command-phase lines
|
|
assert "[IDLE] waiting for ATN" in trace
|
|
assert "[ATN] asserted -> DATA low (ack)" in trace
|
|
assert any(
|
|
line.startswith("[CMD] $24 LISTEN 4") and line.endswith("(addressed: ME)")
|
|
for line in trace
|
|
)
|
|
assert "[CMD] $F0 OPEN SA=0" in trace
|
|
assert "[ATN] released -> LISTENER" in trace
|
|
assert "[ATN] asserted -> command phase" in trace
|
|
assert "[CMD] $3F UNLISTEN" in trace
|
|
assert "[CMD] $E0 CLOSE SA=0" in trace
|
|
assert "[ATN] released -> not addressed -> IDLE" in trace
|
|
|
|
|
|
def test_trace_data_bytes_and_eoi():
|
|
trace = _trace(fixtures.build_session())
|
|
# HELLO WORLD bytes appear as glyphs
|
|
assert "[DATA] $48 'H'" in trace
|
|
assert "[DATA] $4F 'O'" in trace
|
|
assert "[DATA] $20 ' '" in trace
|
|
# each PRINT# record ends with a CR carrying EOI
|
|
assert "[DATA] $0D <CR> <EOI>" in trace
|
|
# exactly two EOI markers (two PRINT# statements)
|
|
assert sum("<EOI>" in line for line in trace) == 2
|
|
|
|
|
|
def test_trace_full_text_reconstructable():
|
|
"""The data bytes, in order, reconstruct the two printed lines."""
|
|
data_bytes = [r.value for r in fixtures.build_session() if r.is_data]
|
|
text = bytes(data_bytes).decode("ascii")
|
|
assert text == "HELLO WORLD\rLINE TWO\r"
|
|
|
|
|
|
def test_command_not_for_us_has_no_me_annotation():
|
|
# if we are device 8, LISTEN 4 is not for us
|
|
trace = _trace(fixtures.build_session(), my_address=8)
|
|
assert "[CMD] $24 LISTEN 4" in trace
|
|
assert not any("(addressed: ME)" in line for line in trace)
|
|
|
|
|
|
def test_raw_mode_is_hex_only():
|
|
trace = _trace(fixtures.build_session(), raw=True)
|
|
# no bracketed tags in raw mode
|
|
assert all(not line.startswith("[") for line in trace)
|
|
# first emitted byte is the LISTEN 4 command byte $24
|
|
assert trace[0] == "24"
|
|
# data bytes present as hex
|
|
assert "48" in trace # 'H'
|