forked from arti/vt1200
153 lines
6.2 KiB
Python
153 lines
6.2 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] *and* the payload, and
|
|
the exact byte offsets the C side will mirror.
|
|
"""
|
|
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_crc16_spans_the_payload():
|
|
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
|
|
# The same header checked without its payload must not validate: that is the
|
|
# difference between "header covers the payload" and "header only".
|
|
assert unpack_header(h)["crc_ok"] is False
|
|
# And a payload bit flip is caught.
|
|
bad = bytearray(payload)
|
|
bad[100] ^= 0x80
|
|
assert unpack_header(h, bytes(bad))["crc_ok"] is False
|
|
|
|
|
|
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():
|
|
"""A CRC sealed over N bytes must not be sent under a different payload_len."""
|
|
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
|
|
|
|
|
|
# --- 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)
|