Files
vt1200/harness/udpstream/tests/test_vidsrc.py
T
opencode.energia.arti.ee 16148d0d24 fix(stream): header-only frame CRC; draw-mode evidence guards
Ruling R14: the frame CRC now covers header bytes 0..11 only.  A
payload-wide CRC cost the terminal ~139 instructions per payload byte
(the parser, not the wire or the blitter, was the frame-rate ceiling),
while every datagram already carries an Ethernet FCS the LANCE checks
and the length bound catches truncation.  frame.py, fp_parse and the
harness's own frame signer follow the new rule; a corrupted payload
byte is now ACCEPTED and a corrupted header byte rejected, stated
explicitly in both test suites, and the C golden vectors were
regenerated from frame.py.

Measured after: RX_READY_A write -> first RX_OK read is 1 emulated
frame for a 20,480-byte frame (84 before), which is the frame hook's
resolution.

Ruling R15: draw mode deletes its dump files before the run (wrapper
and Lua) so a failed run cannot be judged on the previous screen; a
wait that hits its cap reports "timed out waiting for the frame" and
the checker reports it as a timeout, not as pixel mismatches; the
checker asserts the VRAM canary word is 0xA5C3, so a payload that
stopped arming it fails instead of passing quietly.

rx mode now corrupts a header byte in slot B, because a payload flip is
accepted by design under R14.
2026-10-30 23:59:20 +00:00

141 lines
5.3 KiB
Python

"""Tests for the strip encoder: 1-bpp row packing and frame chunking.
The plan's three tests come first, verbatim. The extras pin down what the plan
text leaves implicit but the terminal depends on: the frame is split at 1450
bytes with only the last packet short, and a payload whose length disagrees with
``rows`` is rejected here instead of on the wire.
Note the third plan test checks ``crc_ok`` *with the payload*: since ruling R14
the header CRC covers header[0:12] only, so the same result must hold with or
without the body -- the payload argument is an advisory length check, not part
of the CRC.
"""
import pytest
from harness.udpstream.frame import HEADER_SIZE, KIND_STRIP, unpack_header
from harness.udpstream.vidsrc import Rows, StripSource
#: Largest UDP payload the transport promises per packet.
CHUNK = 1450
def test_pack_row_is_msb_first_and_160_bytes():
px = [0] * 1280
px[0] = 1
px[7] = 1
px[8] = 1
b = Rows.pack_row(px)
assert len(b) == 160
# Pixel 0 -> bit 7, pixel 7 -> bit 0, pixel 8 -> bit 7 of byte 1.
# NOTE: the plan/brief write this as 0b11000001, which is 0xC1 and encodes
# pixels 0, 1 and 7 -- three pixels from a two-pixel segment of the row, so
# no packing rule can produce it from px[0] and px[7]. MSB-first, as the
# same brief's prose states and as the terminal's bit-addressed row needs,
# is 0b10000001. See the task-2 report.
assert b[0] == 0b10000001 and b[1] == 0b10000000
def test_strip_source_rejects_stretch_that_does_not_divide_1024():
with pytest.raises(ValueError):
StripSource(rows=100, stretch=7)
def test_strip_frame_round_trips_through_the_header():
src = StripSource(rows=8, stretch=128)
rows = bytes([0xAA]) * (8 * 160)
pkts = src.encode(frame_no=3, rows_1bpp=rows)
head = unpack_header(pkts[0][:16])
assert head["kind"] == 1 and head["rows"] == 8 and head["stretch"] == 128
assert head["payload_len"] == 8 * 160
assert b"".join(pkts)[16:] == rows
assert unpack_header(pkts[0][:16], b"".join(pkts)[16:])["crc_ok"] is True
# --- packing -------------------------------------------------------------
def test_pack_row_puts_every_pixel_where_the_terminal_reads_it():
"""Pixel i is bit (7 - i % 8) of byte i // 8, i.e. MSB first, and no bit leaks."""
px = [1 if i % 3 == 0 else 0 for i in range(1280)]
b = Rows.pack_row(px)
assert len(b) == 160
for i, v in enumerate(px):
assert (b[i // 8] >> (7 - i % 8)) & 1 == v, i
def test_pack_row_rejects_a_row_that_is_not_1280_pixels():
with pytest.raises(ValueError):
Rows.pack_row([1] * 1279)
with pytest.raises(ValueError):
Rows.pack_row([1] * 1281)
# --- framing -------------------------------------------------------------
def test_encode_splits_the_frame_at_1450_bytes_and_only_the_last_is_short():
"""A full-screen frame is 20,496 B including the header: 15 packets."""
src = StripSource(rows=128, stretch=8)
rows = bytes([0x5A]) * (128 * 160)
pkts = src.encode(frame_no=1, rows_1bpp=rows)
assert all(len(p) <= CHUNK for p in pkts)
assert all(len(p) == CHUNK for p in pkts[:-1])
assert len(pkts) == 15
assert sum(len(p) for p in pkts) == HEADER_SIZE + 128 * 160
assert b"".join(pkts)[HEADER_SIZE:] == rows
def test_a_small_strip_is_a_single_packet():
src = StripSource(rows=8, stretch=128)
rows = bytes([0x11]) * (8 * 160)
pkts = src.encode(frame_no=2, rows_1bpp=rows)
assert len(pkts) == 1
assert len(pkts[0]) == HEADER_SIZE + 8 * 160
assert unpack_header(pkts[0][:HEADER_SIZE], pkts[0][HEADER_SIZE:])["crc_ok"] is True
def test_every_packet_carries_the_same_valid_header():
"""The header travels once, in the first packet, and the body follows it."""
src = StripSource(rows=8, stretch=128)
rows = bytes([0x0F]) * (8 * 160)
pkts = src.encode(frame_no=0x1234, rows_1bpp=rows)
head = unpack_header(pkts[0][:HEADER_SIZE])
assert head["magic"] == b"VS"
assert head["kind"] == KIND_STRIP
assert head["frame_no"] == 0x1234
assert head["payload_len"] == 8 * 160
assert head["rows"] == 8 and head["stretch"] == 128
assert unpack_header(pkts[0][:HEADER_SIZE], b"".join(pkts)[HEADER_SIZE:])["crc_ok"] is True
# The header-only check gives the same answer (ruling R14): the body is
# covered by the Ethernet FCS, not by this CRC.
assert head["crc_ok"] is True
def test_encode_rejects_a_payload_that_is_not_rows_times_160():
src = StripSource(rows=8, stretch=128)
with pytest.raises(ValueError):
src.encode(frame_no=0, rows_1bpp=bytes(8 * 160 - 1))
with pytest.raises(ValueError):
src.encode(frame_no=0, rows_1bpp=bytes(8 * 160 + 1))
def test_encode_rejects_a_frame_number_that_does_not_fit_u16():
src = StripSource(rows=8, stretch=128)
with pytest.raises(ValueError):
src.encode(frame_no=0x10000, rows_1bpp=bytes(8 * 160))
# --- geometry validation -------------------------------------------------
@pytest.mark.parametrize("stretch", [0, 7, 1023, 1025, 2048])
def test_strip_source_rejects_impossible_stretch(stretch):
with pytest.raises(ValueError):
StripSource(rows=8, stretch=stretch)
def test_strip_source_rejects_rows_that_do_not_fit_the_header():
with pytest.raises(ValueError):
StripSource(rows=0, stretch=128)
with pytest.raises(ValueError):
StripSource(rows=256, stretch=4)