forked from arti/vt1200
opencode.energia.arti.ee
16148d0d24
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.
195 lines
8.0 KiB
Python
195 lines
8.0 KiB
Python
"""Tests for the frame header and CRC shared by the host sender and the payload.
|
|
|
|
The plan's two tests come first, verbatim. The rest pin down the parts of the
|
|
format the plan text is ambiguous about, because both ends must agree on them:
|
|
the CCITT-FALSE polynomial, the CRC covering header[0:12] **only** (ruling R14),
|
|
and the exact byte offsets the C side will mirror.
|
|
|
|
Ruling R14 changed the CRC's coverage, and the consequence is stated here on
|
|
purpose: a corrupted **payload** byte leaves a frame ACCEPTED (the Ethernet FCS
|
|
of each datagram is the check for those bytes), while a corrupted **header**
|
|
byte is REJECTED, and a truncated frame (``len < 16 + payload_len``) is rejected
|
|
by the length bound that ``fp_parse`` applies before the CRC.
|
|
"""
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from harness.udpstream.frame import (CRC_OFFSET, HEADER_SIZE, KIND_STATUS,
|
|
KIND_STRIP, MAGIC, VERSION, crc16,
|
|
pack_header, seal_header, unpack_header)
|
|
|
|
|
|
def test_header_is_16_bytes_and_round_trips():
|
|
h = pack_header(KIND_STRIP, frame_no=7, payload_len=12800, rows=80, stretch=12)
|
|
assert len(h) == HEADER_SIZE
|
|
d = unpack_header(h)
|
|
assert d["kind"] == KIND_STRIP and d["frame_no"] == 7
|
|
assert d["payload_len"] == 12800 and d["rows"] == 80 and d["stretch"] == 12
|
|
assert d["crc_ok"] is True
|
|
|
|
|
|
def test_crc_detects_a_single_bit_flip():
|
|
b = bytearray(pack_header(KIND_STRIP, 1, 10, 8, 128))
|
|
b[5] ^= 0x01
|
|
assert unpack_header(bytes(b))["crc_ok"] is False
|
|
|
|
|
|
# --- the CRC itself ------------------------------------------------------
|
|
|
|
def test_crc16_is_ccitt_false():
|
|
"""The canonical CCITT-FALSE check value; catches a wrong poly/init/reflect."""
|
|
assert crc16(b"123456789") == 0x29B1
|
|
assert crc16(b"") == 0xFFFF
|
|
|
|
|
|
def test_crc16_matches_an_independent_implementation():
|
|
"""Cross-check the table against the stdlib, which is CCITT with init 0xFFFF.
|
|
|
|
One byte per call exercises every table entry; the longer inputs exercise
|
|
the running state, including the sizes the stream actually uses.
|
|
"""
|
|
import binascii
|
|
for b in range(256):
|
|
assert crc16(bytes([b])) == binascii.crc_hqx(bytes([b]), 0xFFFF), b
|
|
for n in (0, 1, 12, 160, 12816, 20496):
|
|
data = bytes((i * 37 + 11) & 0xFF for i in range(n))
|
|
assert crc16(data) == binascii.crc_hqx(data, 0xFFFF), n
|
|
|
|
|
|
def test_crc_covers_the_header_not_the_payload():
|
|
"""Ruling R14: payload bytes are the chip's job, header bytes are ours.
|
|
|
|
A corrupted payload byte must leave ``crc_ok`` True (the LANCE has already
|
|
validated each datagram's Ethernet FCS) and a corrupted header byte must
|
|
make it False. The payload argument is advisory: it must match
|
|
``payload_len`` in length, and its bytes are not hashed.
|
|
"""
|
|
payload = bytes(range(256)) * 2
|
|
h = pack_header(KIND_STRIP, frame_no=1, payload_len=len(payload), rows=8,
|
|
stretch=128, payload=payload)
|
|
assert unpack_header(h, payload)["crc_ok"] is True
|
|
# Same result without the body at all: the body is not part of the CRC.
|
|
assert unpack_header(h)["crc_ok"] is True
|
|
|
|
# A payload bit flip is ACCEPTED here: the Ethernet FCS is the chip's job.
|
|
bad = bytearray(payload)
|
|
bad[100] ^= 0x80
|
|
assert unpack_header(h, bytes(bad))["crc_ok"] is True
|
|
|
|
# A header bit flip (inside bytes 0..11) is REJECTED.
|
|
hh = bytearray(h)
|
|
hh[8] ^= 0x01
|
|
assert unpack_header(bytes(hh))["crc_ok"] is False
|
|
|
|
# The payload argument still has to be the length the header claims: a
|
|
# truncated body is a length error, not a CRC error (fp_parse's job).
|
|
with pytest.raises(ValueError):
|
|
unpack_header(h, payload[:-1])
|
|
with pytest.raises(ValueError):
|
|
unpack_header(h, payload + b"\x00")
|
|
|
|
|
|
def test_payload_bytes_do_not_change_the_sealed_header():
|
|
"""One header, two different bodies: the header bytes are identical."""
|
|
a = b"\x00" * 160
|
|
b = b"\xFF" * 160
|
|
ha = pack_header(KIND_STRIP, 5, len(a), 1, 8, payload=a)
|
|
hb = pack_header(KIND_STRIP, 5, len(b), 1, 8, payload=b)
|
|
assert ha == hb
|
|
assert unpack_header(ha, a)["crc_ok"] is True
|
|
assert unpack_header(hb, b)["crc_ok"] is True
|
|
|
|
|
|
def test_seal_header_after_the_payload_is_known():
|
|
"""The chunked sender builds the header first, then seals it (Task 2/6 path)."""
|
|
payload = b"\x01\x02\x03\x04"
|
|
h = pack_header(KIND_STRIP, 3, len(payload), 8, 128)
|
|
sealed = seal_header(h, payload)
|
|
assert sealed[:CRC_OFFSET] == h[:CRC_OFFSET]
|
|
assert unpack_header(sealed, payload)["crc_ok"] is True
|
|
# Sealing twice is idempotent.
|
|
assert seal_header(sealed, payload) == sealed
|
|
# And it produces the same bytes as the one-shot call.
|
|
assert sealed == pack_header(KIND_STRIP, 3, len(payload), 8, 128,
|
|
payload=payload)
|
|
|
|
|
|
def test_seal_header_rejects_over_length_input():
|
|
"""A buffer longer than a header must not be silently truncated."""
|
|
h = pack_header(KIND_STRIP, 3, 4, 8, 128)
|
|
with pytest.raises(ValueError):
|
|
seal_header(h + b"\x01\x02\x03\x04") # header + payload, not header
|
|
with pytest.raises(ValueError):
|
|
seal_header(h + b"\x00" * HEADER_SIZE)
|
|
with pytest.raises(ValueError):
|
|
seal_header(h[:HEADER_SIZE - 1])
|
|
# Exactly a header is still accepted, and matches the one-shot call.
|
|
assert seal_header(h, b"\x01\x02\x03\x04") == pack_header(
|
|
KIND_STRIP, 3, 4, 8, 128, payload=b"\x01\x02\x03\x04")
|
|
|
|
|
|
def test_pack_header_rejects_payload_length_mismatch():
|
|
"""The payload argument is advisory: it must match payload_len when given."""
|
|
payload = b"\xAA" * 10
|
|
pack_header(KIND_STRIP, 1, 10, 8, 128, payload=payload) # exact: fine
|
|
with pytest.raises(ValueError):
|
|
pack_header(KIND_STRIP, 1, 9, 8, 128, payload=payload)
|
|
with pytest.raises(ValueError):
|
|
pack_header(KIND_STRIP, 1, 11, 8, 128, payload=payload)
|
|
with pytest.raises(ValueError):
|
|
pack_header(KIND_STRIP, 1, 10, 8, 128, payload=b"") # explicit empty
|
|
# Omitting the payload is the header-only path: no length to disagree with.
|
|
assert unpack_header(pack_header(KIND_STRIP, 1, 10, 8, 128))["crc_ok"] is True
|
|
# seal_header applies the same advisory rule.
|
|
h = pack_header(KIND_STRIP, 1, 10, 8, 128)
|
|
assert seal_header(h, b"\x00" * 10) == h
|
|
assert seal_header(h) == h # not given: no check
|
|
with pytest.raises(ValueError):
|
|
seal_header(h, b"\x00" * 9)
|
|
|
|
|
|
# --- the byte layout the C guest will mirror -----------------------------
|
|
|
|
def test_byte_offsets_and_endianness():
|
|
h = pack_header(KIND_STATUS, frame_no=0x1234, payload_len=0x5678, rows=0x9A,
|
|
stretch=0xBC)
|
|
assert h[0:2] == MAGIC == b"VS"
|
|
assert h[2] == VERSION == 0x01
|
|
assert h[3] == KIND_STATUS == 5
|
|
assert h[4:6] == b"\x34\x12" # frame_no, little-endian
|
|
assert h[6:8] == b"\x78\x56" # payload_len, little-endian
|
|
assert h[8] == 0x9A # rows
|
|
assert h[9] == 0xBC # stretch
|
|
assert h[10:12] == b"\x00\x00" # reserved
|
|
assert h[12:14] == struct.pack("<H", crc16(h[0:12]))
|
|
assert h[14:16] == b"\x00\x00" # padding to HEADER_SIZE
|
|
|
|
|
|
def test_unpack_reports_magic_and_version():
|
|
d = unpack_header(pack_header(KIND_STRIP, 0, 0, 0, 1))
|
|
assert d["magic"] == b"VS" and d["version"] == VERSION
|
|
d = unpack_header(b"XX" + pack_header(KIND_STRIP, 0, 0, 0, 1)[2:])
|
|
assert d["magic"] == b"XX"
|
|
|
|
|
|
# --- malformed input -----------------------------------------------------
|
|
|
|
def test_short_buffer_is_rejected():
|
|
with pytest.raises(ValueError):
|
|
unpack_header(b"V" * (HEADER_SIZE - 1))
|
|
|
|
|
|
@pytest.mark.parametrize("kwargs", [
|
|
dict(kind=256), dict(kind=-1),
|
|
dict(frame_no=0x10000), dict(frame_no=-1),
|
|
dict(payload_len=0x10000), dict(payload_len=-1),
|
|
dict(rows=256), dict(rows=-1),
|
|
dict(stretch=256), dict(stretch=-1),
|
|
])
|
|
def test_out_of_range_fields_raise(kwargs):
|
|
base = dict(kind=KIND_STRIP, frame_no=1, payload_len=10, rows=8, stretch=128)
|
|
base.update(kwargs)
|
|
with pytest.raises(ValueError):
|
|
pack_header(**base)
|