Christian Werner
0d6034702a
Covers the plain property, per-record-type output (printable chars, CR/LF newlines, LISTEN/UNLISTEN dividers, silenced events and other commands), and a full-session integration check. Generated by Clanker
195 lines
6.2 KiB
Python
195 lines
6.2 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, debug=True).format_stream(records))
|
|
|
|
|
|
def _plain(records, my_address=4):
|
|
return list(TraceFormatter(my_address=my_address).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'
|
|
|
|
|
|
# --- plain mode -------------------------------------------------------------
|
|
def _make_record(kind, value, flags=0):
|
|
from iecpoc.device import IecRecord
|
|
return IecRecord(kind, value, flags, 0)
|
|
|
|
|
|
def test_plain_property():
|
|
assert TraceFormatter(debug=False).plain is True
|
|
assert TraceFormatter(debug=True).plain is False
|
|
assert TraceFormatter(raw=True).plain is False
|
|
|
|
|
|
def test_plain_listen_divider():
|
|
rec = _make_record(device.KIND_COMMAND, 0x24, device.FLAG_ADDRESSED) # LISTEN 4
|
|
out = _plain([rec])
|
|
assert len(out) == 1
|
|
assert "LISTEN 4" in out[0]
|
|
|
|
|
|
def test_plain_unlisten_divider():
|
|
rec = _make_record(device.KIND_COMMAND, 0x3F) # UNLISTEN
|
|
out = _plain([rec])
|
|
assert len(out) == 1
|
|
assert "UNLISTEN" in out[0]
|
|
|
|
|
|
def test_plain_data_printable():
|
|
rec = _make_record(device.KIND_DATA, 0x48, device.FLAG_ADDRESSED) # 'H'
|
|
assert _plain([rec]) == ["H"]
|
|
|
|
|
|
def test_plain_data_cr_yields_newline():
|
|
rec = _make_record(device.KIND_DATA, 0x0D, device.FLAG_ADDRESSED)
|
|
assert _plain([rec]) == ["\n"]
|
|
|
|
|
|
def test_plain_data_lf_yields_newline():
|
|
rec = _make_record(device.KIND_DATA, 0x0A, device.FLAG_ADDRESSED)
|
|
assert _plain([rec]) == ["\n"]
|
|
|
|
|
|
def test_plain_data_shift_cr_yields_newline():
|
|
rec = _make_record(device.KIND_DATA, 0x8D, device.FLAG_ADDRESSED)
|
|
assert _plain([rec]) == ["\n"]
|
|
|
|
|
|
def test_plain_events_are_silent():
|
|
recs = [
|
|
_make_record(device.KIND_EVENT, device.EV_IDLE),
|
|
_make_record(device.KIND_EVENT, device.EV_ATN_ASSERTED),
|
|
_make_record(device.KIND_EVENT, device.EV_ATN_RELEASED),
|
|
]
|
|
assert _plain(recs) == []
|
|
|
|
|
|
def test_plain_non_listen_commands_silent():
|
|
recs = [
|
|
_make_record(device.KIND_COMMAND, 0xF0), # OPEN SA=0
|
|
_make_record(device.KIND_COMMAND, 0xE0), # CLOSE SA=0
|
|
_make_record(device.KIND_COMMAND, 0x60), # SECOND/DATA SA=0
|
|
_make_record(device.KIND_COMMAND, 0x48), # TALK 8
|
|
_make_record(device.KIND_COMMAND, 0x5F), # UNTALK
|
|
]
|
|
assert _plain(recs) == []
|
|
|
|
|
|
def test_plain_graphic_bytes_silent():
|
|
rec = _make_record(device.KIND_DATA, 0x70, device.FLAG_ADDRESSED) # graphic range
|
|
assert _plain([rec]) == []
|
|
|
|
|
|
def test_plain_control_bytes_silent():
|
|
rec = _make_record(device.KIND_DATA, 0x12, device.FLAG_ADDRESSED) # RVS-ON
|
|
assert _plain([rec]) == []
|
|
|
|
|
|
def test_plain_full_session():
|
|
out = _plain(fixtures.build_session())
|
|
joined = "".join(out)
|
|
assert "HELLO WORLD" in joined
|
|
assert "LINE TWO" in joined
|
|
assert "LISTEN 4" in joined
|
|
assert "UNLISTEN" in joined
|
|
# each PRINT# record ends with CR → two newlines from data, plus divider newlines
|
|
assert joined.count("\n") >= 2
|