forked from arti/vt1200
227 lines
8.8 KiB
Python
227 lines
8.8 KiB
Python
"""Tests for the chunked dictionary upload protocol.
|
|
|
|
The plan's test comes first, verbatim. The extras pin down what the plan text
|
|
leaves implicit but the terminal's receiver (Task 3) depends on:
|
|
|
|
* the 32 KB dictionary splits into exactly 23 independently framed packets
|
|
(payload of at most 1450 bytes: a 2-byte offset plus at most 1448 bytes of
|
|
dictionary);
|
|
* every packet is an ordinary ``KIND_DICT_CHUNK`` frame -- 16-byte header, the
|
|
header-only CRC, no geometry -- so a receiver can tell it from a strip or a
|
|
status frame and drop a corrupted header without touching the payload;
|
|
* the offset is the dictionary position of the first body byte, so a gap or a
|
|
reordered packet is visible from the offsets alone;
|
|
* ``offset_base`` shifts every offset, which is how a dictionary can be placed
|
|
at a non-zero position in the receiver's buffer.
|
|
|
|
Reassembly is asserted byte for byte: concatenating the bodies in arrival order
|
|
must give back the input, and the last offset plus the last body length must be
|
|
the end of the dictionary.
|
|
|
|
The last group of tests is Task 5's delta encoding, and it pins the *payload*,
|
|
not just the host round trip: the C decoder (``vidstream.c``'s ``paint_delta``)
|
|
reads the same layout, so what is asserted here is what ``frame_parse.h``
|
|
writes down. ``unpack_delta`` is the host's reading of that layout and shares
|
|
none of the encoder's code, so a delta that round-trips through it is not being
|
|
blessed by its own writer.
|
|
"""
|
|
import pytest
|
|
|
|
from harness.udpstream.frame import (HEADER_SIZE, KIND_DICT_CHUNK,
|
|
unpack_header)
|
|
from harness.udpstream.tiledict import (KEYFRAME_EVERY, MAX_PAYLOAD,
|
|
delta_frame, dict_chunks, keyframe_due,
|
|
payload_body, payload_offset,
|
|
unpack_delta)
|
|
|
|
#: A whole dictionary: 256 entries of 128 bytes.
|
|
DICTIONARY = bytes(range(256)) * 128
|
|
|
|
#: One full-screen index frame: 1,280 row-major tile indices (40 x 32).
|
|
FRAME = bytes(range(256)) * 5 # exactly 1,280 bytes
|
|
|
|
|
|
def test_chunks_cover_the_dictionary_once_in_order():
|
|
d = bytes(range(256)) * 128 # 32 KB
|
|
pk = dict_chunks(d)
|
|
assert b"".join(payload_body(p) for p in pk) == d
|
|
offs = [payload_offset(p) for p in pk]
|
|
assert offs == sorted(offs) and offs[0] == 0
|
|
assert offs[-1] + len(payload_body(pk[-1])) == len(d)
|
|
|
|
|
|
def test_dictionary_splits_into_23_framed_packets():
|
|
pk = dict_chunks(DICTIONARY)
|
|
assert len(pk) == 23
|
|
for packet in pk:
|
|
header = unpack_header(packet)
|
|
assert header["magic"] == b"VS"
|
|
assert header["kind"] == KIND_DICT_CHUNK
|
|
assert header["crc_ok"] is True
|
|
assert 0 < header["payload_len"] <= MAX_PAYLOAD
|
|
assert len(packet) == HEADER_SIZE + header["payload_len"]
|
|
|
|
|
|
def test_offsets_name_each_body_position():
|
|
pk = dict_chunks(DICTIONARY)
|
|
position = 0
|
|
for packet in pk:
|
|
assert payload_offset(packet) == position
|
|
position += len(payload_body(packet))
|
|
assert position == len(DICTIONARY)
|
|
|
|
|
|
def test_offset_base_shifts_every_offset():
|
|
d = bytes(3000)
|
|
base = 100 * 128 # entry 100 of a dictionary
|
|
pk = dict_chunks(d, offset_base=base)
|
|
positions = [payload_offset(p) for p in pk]
|
|
assert positions[0] == base
|
|
assert b"".join(payload_body(p) for p in pk) == d
|
|
assert positions[-1] + len(payload_body(pk[-1])) == base + len(d)
|
|
|
|
|
|
def test_body_never_crosses_the_1450_byte_payload_budget():
|
|
pk = dict_chunks(DICTIONARY)
|
|
assert max(unpack_header(p)["payload_len"] for p in pk) == MAX_PAYLOAD
|
|
assert all(len(payload_body(p)) <= MAX_PAYLOAD - 2 for p in pk)
|
|
|
|
|
|
def test_dictionary_that_would_overflow_the_u16_offset_is_rejected():
|
|
with pytest.raises(ValueError):
|
|
dict_chunks(DICTIONARY, offset_base=0x10000 - len(DICTIONARY) + 1)
|
|
with pytest.raises(ValueError):
|
|
dict_chunks(DICTIONARY, offset_base=-1)
|
|
|
|
|
|
def test_empty_dictionary_is_rejected():
|
|
with pytest.raises(ValueError):
|
|
dict_chunks(b"")
|
|
|
|
|
|
def test_payload_helpers_reject_a_frame_of_another_kind():
|
|
from harness.udpstream.frame import KIND_STRIP, pack_header
|
|
strip = pack_header(KIND_STRIP, 1, 2, 0, 0, payload=b"\x00\x00")
|
|
with pytest.raises(ValueError):
|
|
payload_offset(strip)
|
|
with pytest.raises(ValueError):
|
|
payload_body(strip)
|
|
|
|
|
|
def test_payload_helpers_reject_a_truncated_payload():
|
|
packet = bytearray(dict_chunks(bytes(100))[0])
|
|
header = unpack_header(packet)
|
|
with pytest.raises(ValueError):
|
|
payload_body(bytes(packet[:HEADER_SIZE + header["payload_len"] - 1]))
|
|
|
|
|
|
# --- Task 5: delta frames and keyframe cadence ---------------------------
|
|
#
|
|
# The layout pinned here is the one frame_parse.h writes down for the C
|
|
# decoder: a u16 count, `count` u16 row-major tile positions and `count`
|
|
# one-byte dictionary entries. Positions are two bytes because a screen has
|
|
# 1,280 tiles and one byte cannot name tile 1279 -- the *entry* fits in a byte
|
|
# (256 of them), the *position* does not. That is the one layout fact this
|
|
# group of tests exists to stop from drifting on one side.
|
|
|
|
def test_delta_of_two_changed_tiles_lists_exactly_those_two_indices():
|
|
"""The brief's test: two tiles changed, two entries and no others."""
|
|
cur = bytearray(FRAME)
|
|
cur[7] = 0x80
|
|
cur[1000] = 0xFF
|
|
|
|
payload = delta_frame(FRAME, bytes(cur))
|
|
|
|
assert payload is not None
|
|
assert unpack_delta(payload) == [(7, 0x80), (1000, 0xFF)]
|
|
# 2 bytes of count, two positions and two values.
|
|
assert len(payload) == 2 + 2 * 2 + 2
|
|
|
|
|
|
def test_identical_frames_produce_no_delta():
|
|
assert delta_frame(FRAME, FRAME) is None
|
|
assert delta_frame(FRAME, bytes(FRAME)) is None
|
|
|
|
|
|
def test_a_keyframe_is_due_at_frame_60_and_120():
|
|
assert KEYFRAME_EVERY == 60
|
|
assert keyframe_due(60) is True
|
|
assert keyframe_due(120) is True
|
|
# Frame 0 has no previous frame, so the first frame is a keyframe too; the
|
|
# 59 frames between two keyframes are deltas.
|
|
assert keyframe_due(0) is True
|
|
assert not any(keyframe_due(n) for n in range(1, 60))
|
|
assert not any(keyframe_due(n) for n in range(61, 120))
|
|
|
|
|
|
def test_a_reported_drop_forces_a_keyframe():
|
|
"""The status frame's `drops` is cumulative: any increase means a loss."""
|
|
assert keyframe_due(7, drops=3, seen_drops=2) is True
|
|
assert keyframe_due(7, drops=2, seen_drops=2) is False
|
|
assert keyframe_due(7, drops=0, seen_drops=0) is False
|
|
|
|
|
|
def test_delta_positions_are_strictly_ascending_in_scan_order():
|
|
cur = bytearray(FRAME)
|
|
for position in (1279, 0, 640, 639):
|
|
cur[position] = (cur[position] + 1) & 0xFF
|
|
|
|
payload = delta_frame(FRAME, bytes(cur))
|
|
|
|
assert [p for p, _ in unpack_delta(payload)] == [0, 639, 640, 1279]
|
|
|
|
|
|
def test_a_whole_screen_delta_is_the_largest_frame_and_stays_one_slot():
|
|
cur = bytes((b + 1) & 0xFF for b in FRAME)
|
|
|
|
payload = delta_frame(FRAME, cur)
|
|
|
|
assert len(payload) == 2 + 3 * 1280
|
|
assert unpack_delta(payload) == [(i, cur[i]) for i in range(1280)]
|
|
# A delta can never be larger than 2 + 3 * 1,280 = 3,842 bytes; both that
|
|
# and a 1,280-byte keyframe fit the payload's 24 KB slot many times over.
|
|
assert len(payload) == 3842
|
|
|
|
|
|
def test_encoder_rejects_a_frame_that_is_not_1280_index_bytes():
|
|
with pytest.raises(ValueError):
|
|
delta_frame(FRAME[:-1], FRAME)
|
|
with pytest.raises(ValueError):
|
|
delta_frame(FRAME, FRAME + b"\x00")
|
|
with pytest.raises(TypeError):
|
|
delta_frame(FRAME, "not bytes")
|
|
|
|
|
|
def test_decoder_rejects_a_list_that_disagrees_with_the_length():
|
|
"""`payload_len` is authoritative: the list cannot run past it."""
|
|
changed = bytes([FRAME[0] ^ 0xFF]) + FRAME[1:]
|
|
good = delta_frame(FRAME, changed)
|
|
|
|
with pytest.raises(ValueError):
|
|
unpack_delta(good[:-1]) # one value missing
|
|
with pytest.raises(ValueError):
|
|
unpack_delta(good + b"\x00") # a tail outside the list
|
|
with pytest.raises(ValueError):
|
|
unpack_delta(b"") # no count field at all
|
|
with pytest.raises(ValueError):
|
|
unpack_delta(b"\x00\x00") # count 0: the encoder sends None
|
|
|
|
|
|
def test_decoder_rejects_a_tile_position_outside_the_grid():
|
|
"""A position >= 1,280 must be refused, never wrapped onto another tile."""
|
|
assert unpack_delta(b"\x01\x00" + (1279).to_bytes(2, "little") + b"\x00") \
|
|
== [(1279, 0)]
|
|
|
|
for position in (1280, 5000, 0xFFFF):
|
|
with pytest.raises(ValueError):
|
|
unpack_delta(b"\x01\x00"
|
|
+ position.to_bytes(2, "little") + b"\x00")
|
|
|
|
|
|
def test_decoder_rejects_a_repeated_or_descending_position():
|
|
"""Strictly ascending: a tile is named once, in scan order."""
|
|
with pytest.raises(ValueError):
|
|
unpack_delta(b"\x02\x00" + bytes([5, 0, 5, 0]) + b"\x00\x01")
|
|
with pytest.raises(ValueError):
|
|
unpack_delta(b"\x02\x00" + bytes([6, 0, 5, 0]) + b"\x00\x01")
|