mirror of
https://git.k-space.ee/arti/vt1200.git
synced 2026-11-02 12:23:17 +02:00
291 lines
12 KiB
Python
291 lines
12 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.
|
|
|
|
The last section pins the *other* direction: the terminal's status body
|
|
(``KIND_STATUS``) and the Ethernet envelope both ends put the frame in. The
|
|
offsets there are the contract with ``send_status()`` in vidstream.c.
|
|
"""
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from harness.udpstream.frame import (CRC_OFFSET, HEADER_SIZE, KIND_STATUS,
|
|
KIND_STRIP, MAGIC, VERSION, STATUS_SIZE,
|
|
BROADCAST, ETHERTYPE, TERMINAL_MAC,
|
|
crc16, ethernet_frame, pack_header,
|
|
pack_status, seal_header, unpack_header,
|
|
unpack_status)
|
|
|
|
|
|
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)
|
|
|
|
|
|
# --- the status body that travels back (Task 6) --------------------------
|
|
|
|
def test_status_body_is_twelve_bytes_little_endian():
|
|
"""The terminal's status body, field by field, at the offsets the C writes.
|
|
|
|
The C side (vidstream.c, send_status()) writes the same five fields with
|
|
byte stores, so the offsets below are the contract between the two ends.
|
|
"""
|
|
b = pack_status(frames_rx=0x12345678, crc_err=0x0102, drops=0x0304,
|
|
csr0=0x20F2, heap_free_kb=0x0506)
|
|
assert len(b) == STATUS_SIZE == 12
|
|
assert b[0:4] == b"\x78\x56\x34\x12" # frames_rx, u32 LE
|
|
assert b[4:6] == b"\x02\x01" # crc_err, u16 LE
|
|
assert b[6:8] == b"\x04\x03" # drops, u16 LE
|
|
assert b[8:10] == b"\xF2\x20" # csr0, u16 LE
|
|
assert b[10:12] == b"\x06\x05" # heap_free_kb, u16 LE
|
|
assert unpack_status(b) == {
|
|
"frames_rx": 0x12345678, "crc_err": 0x0102, "drops": 0x0304,
|
|
"csr0": 0x20F2, "heap_free_kb": 0x0506,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("kwargs", [
|
|
dict(frames_rx=0x100000000), dict(frames_rx=-1),
|
|
dict(crc_err=0x10000), dict(drops=-1),
|
|
dict(csr0=0x10000), dict(heap_free_kb=-1),
|
|
])
|
|
def test_status_fields_out_of_range_raise(kwargs):
|
|
base = dict(frames_rx=1, crc_err=0, drops=0, csr0=0, heap_free_kb=0)
|
|
base.update(kwargs)
|
|
with pytest.raises(ValueError):
|
|
pack_status(**base)
|
|
|
|
|
|
def test_status_body_must_be_exactly_twelve_bytes():
|
|
b = pack_status(1, 0, 0, 0, 0)
|
|
with pytest.raises(ValueError):
|
|
unpack_status(b[:-1])
|
|
with pytest.raises(ValueError):
|
|
unpack_status(b + b"\x00")
|
|
|
|
|
|
def test_status_uses_the_same_header_with_zero_geometry():
|
|
"""A status frame is a normal frame: same header, same header-only CRC."""
|
|
body = pack_status(200, 0, 0, 0x20F2, 2048)
|
|
h = pack_header(KIND_STATUS, frame_no=3, payload_len=len(body), rows=0,
|
|
stretch=0, payload=body)
|
|
d = unpack_header(h, body)
|
|
assert d["kind"] == KIND_STATUS
|
|
assert d["rows"] == 0 and d["stretch"] == 0
|
|
assert d["crc_ok"] is True
|
|
|
|
|
|
def test_ethernet_envelope_is_the_streams_own_type():
|
|
"""The status frame travels in an ordinary Ethernet frame on its own type.
|
|
|
|
0x88B7 is the video stream's: 0x88B5 is the probe frame sendraw.py sends
|
|
and 0x88B6 the command frame lance_cmd.c uses, so a captured status frame
|
|
cannot be mistaken for either. The terminal's LANCE does not filter on
|
|
ethertype at all (it accepts broadcast and its own physical address), so
|
|
this value is for the host's benefit.
|
|
"""
|
|
body = pack_status(1, 0, 0, 0, 0)
|
|
frame = pack_header(KIND_STATUS, 1, len(body), 0, 0) + body
|
|
data = ethernet_frame(frame, src=TERMINAL_MAC, dst=BROADCAST)
|
|
assert len(data) == 14 + HEADER_SIZE + STATUS_SIZE
|
|
assert data[0:6] == b"\xff" * 6
|
|
assert data[6:12] == TERMINAL_MAC
|
|
assert data[12:14] == struct.pack(">H", ETHERTYPE) == b"\x88\xB7"
|
|
assert data[14:] == frame
|
|
|
|
|
|
def test_ethernet_frame_pads_to_the_minimum_only_at_the_end():
|
|
"""A 60-byte Ethernet minimum must not be inserted between chunks.
|
|
|
|
Padding mid-frame would put bytes the sender never meant between two
|
|
datagrams of one frame, and the receiver would see a frame whose body is
|
|
longer than payload_len and shifted -- exactly the bug this pins.
|
|
"""
|
|
short = ethernet_frame(b"VS" + bytes(10), src=TERMINAL_MAC, pad=True)
|
|
assert len(short) == 60
|
|
assert short[-18:] == bytes(18)
|
|
long = b"x" * 1450
|
|
assert ethernet_frame(long, src=TERMINAL_MAC, pad=True) == (
|
|
BROADCAST + TERMINAL_MAC + b"\x88\xB7" + long)
|
|
# Off by default: the chunking layer decides, and only for the last chunk.
|
|
assert len(ethernet_frame(b"VS" + bytes(10), src=TERMINAL_MAC)) == 26
|