vt1200/harness/udpstream/tests/test_frame.py

"""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, KINDS, 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


def test_payload_header_kinds_match_the_wire_table():
    """The terminal's frame_parse.h mirrors this module's KINDS table.

    The kind byte is written by frame.py and read by the payload, so the same
    number is written down in Python and in C.  A renumbering on one side would
    silently mis-dispatch every frame on the other: a TILES_FULL would be
    painted as a strip, or a DICT_CHUNK would be counted as a frame.  No MAME
    check would notice until the kind it exercises itself changed, which is why
    this parses the C header rather than trusting the comment that says the two
    agree.
    """
    import pathlib
    import re

    header = (pathlib.Path(__file__).resolve().parents[1] / "frame_parse.h")
    text = header.read_text()
    found = {name: int(value) for name, value in
             re.findall(r"^#define\s+FP_KIND_(\w+)\s+(\d+)\b", text, re.M)}
    assert found, "frame_parse.h has no FP_KIND_* definitions"
    assert found == KINDS
    # And the status kind the payload builds is the one frame.py reads.
    assert found["STATUS"] == KIND_STATUS