forked from arti/vt1200
158 lines
6.1 KiB
Python
158 lines
6.1 KiB
Python
"""Tests for the strip encoder: 1-bpp row packing and frame chunking.
|
|
|
|
The plan's three tests come first, verbatim. The extras pin down what the plan
|
|
text leaves implicit but the terminal depends on: the frame is split at 1450
|
|
bytes with only the last packet short, and a payload whose length disagrees with
|
|
``rows`` is rejected here instead of on the wire.
|
|
|
|
Note the third plan test checks ``crc_ok`` *with the payload*: since ruling R14
|
|
the header CRC covers header[0:12] only, so the same result must hold with or
|
|
without the body -- the payload argument is an advisory length check, not part
|
|
of the CRC.
|
|
"""
|
|
import pytest
|
|
|
|
from harness.udpstream.frame import HEADER_SIZE, KIND_STRIP, unpack_header
|
|
from harness.udpstream.vidsrc import (
|
|
MAX_ROWS, RING_DATAGRAMS, Rows, StripSource)
|
|
|
|
#: Largest UDP payload the transport promises per packet.
|
|
CHUNK = 1450
|
|
|
|
|
|
def test_pack_row_is_msb_first_and_160_bytes():
|
|
px = [0] * 1280
|
|
px[0] = 1
|
|
px[7] = 1
|
|
px[8] = 1
|
|
b = Rows.pack_row(px)
|
|
assert len(b) == 160
|
|
# Pixel 0 -> bit 7, pixel 7 -> bit 0, pixel 8 -> bit 7 of byte 1.
|
|
# NOTE: the plan/brief write this as 0b11000001, which is 0xC1 and encodes
|
|
# pixels 0, 1 and 7 -- three pixels from a two-pixel segment of the row, so
|
|
# no packing rule can produce it from px[0] and px[7]. MSB-first, as the
|
|
# same brief's prose states and as the terminal's bit-addressed row needs,
|
|
# is 0b10000001. See the task-2 report.
|
|
assert b[0] == 0b10000001 and b[1] == 0b10000000
|
|
|
|
|
|
def test_strip_source_rejects_stretch_that_does_not_divide_1024():
|
|
with pytest.raises(ValueError):
|
|
StripSource(rows=100, stretch=7)
|
|
|
|
|
|
def test_strip_frame_round_trips_through_the_header():
|
|
src = StripSource(rows=8, stretch=128)
|
|
rows = bytes([0xAA]) * (8 * 160)
|
|
pkts = src.encode(frame_no=3, rows_1bpp=rows)
|
|
head = unpack_header(pkts[0][:16])
|
|
assert head["kind"] == 1 and head["rows"] == 8 and head["stretch"] == 128
|
|
assert head["payload_len"] == 8 * 160
|
|
assert b"".join(pkts)[16:] == rows
|
|
assert unpack_header(pkts[0][:16], b"".join(pkts)[16:])["crc_ok"] is True
|
|
|
|
|
|
# --- packing -------------------------------------------------------------
|
|
|
|
def test_pack_row_puts_every_pixel_where_the_terminal_reads_it():
|
|
"""Pixel i is bit (7 - i % 8) of byte i // 8, i.e. MSB first, and no bit leaks."""
|
|
px = [1 if i % 3 == 0 else 0 for i in range(1280)]
|
|
b = Rows.pack_row(px)
|
|
assert len(b) == 160
|
|
for i, v in enumerate(px):
|
|
assert (b[i // 8] >> (7 - i % 8)) & 1 == v, i
|
|
|
|
|
|
def test_pack_row_rejects_a_row_that_is_not_1280_pixels():
|
|
with pytest.raises(ValueError):
|
|
Rows.pack_row([1] * 1279)
|
|
with pytest.raises(ValueError):
|
|
Rows.pack_row([1] * 1281)
|
|
|
|
|
|
# --- framing -------------------------------------------------------------
|
|
|
|
def test_encode_splits_the_frame_at_1450_bytes_and_only_the_last_is_short():
|
|
"""A full-screen frame is 20,496 B including the header: 15 packets."""
|
|
src = StripSource(rows=128, stretch=8)
|
|
rows = bytes([0x5A]) * (128 * 160)
|
|
pkts = src.encode(frame_no=1, rows_1bpp=rows)
|
|
|
|
assert all(len(p) <= CHUNK for p in pkts)
|
|
assert all(len(p) == CHUNK for p in pkts[:-1])
|
|
assert len(pkts) == 15
|
|
assert sum(len(p) for p in pkts) == HEADER_SIZE + 128 * 160
|
|
assert b"".join(pkts)[HEADER_SIZE:] == rows
|
|
|
|
|
|
def test_a_small_strip_is_a_single_packet():
|
|
src = StripSource(rows=8, stretch=128)
|
|
rows = bytes([0x11]) * (8 * 160)
|
|
pkts = src.encode(frame_no=2, rows_1bpp=rows)
|
|
assert len(pkts) == 1
|
|
assert len(pkts[0]) == HEADER_SIZE + 8 * 160
|
|
assert unpack_header(pkts[0][:HEADER_SIZE], pkts[0][HEADER_SIZE:])["crc_ok"] is True
|
|
|
|
|
|
def test_every_packet_carries_the_same_valid_header():
|
|
"""The header travels once, in the first packet, and the body follows it."""
|
|
src = StripSource(rows=8, stretch=128)
|
|
rows = bytes([0x0F]) * (8 * 160)
|
|
pkts = src.encode(frame_no=0x1234, rows_1bpp=rows)
|
|
|
|
head = unpack_header(pkts[0][:HEADER_SIZE])
|
|
assert head["magic"] == b"VS"
|
|
assert head["kind"] == KIND_STRIP
|
|
assert head["frame_no"] == 0x1234
|
|
assert head["payload_len"] == 8 * 160
|
|
assert head["rows"] == 8 and head["stretch"] == 128
|
|
assert unpack_header(pkts[0][:HEADER_SIZE], b"".join(pkts)[HEADER_SIZE:])["crc_ok"] is True
|
|
# The header-only check gives the same answer (ruling R14): the body is
|
|
# covered by the Ethernet FCS, not by this CRC.
|
|
assert head["crc_ok"] is True
|
|
|
|
|
|
def test_encode_rejects_a_payload_that_is_not_rows_times_160():
|
|
src = StripSource(rows=8, stretch=128)
|
|
with pytest.raises(ValueError):
|
|
src.encode(frame_no=0, rows_1bpp=bytes(8 * 160 - 1))
|
|
with pytest.raises(ValueError):
|
|
src.encode(frame_no=0, rows_1bpp=bytes(8 * 160 + 1))
|
|
|
|
|
|
def test_encode_rejects_a_frame_number_that_does_not_fit_u16():
|
|
src = StripSource(rows=8, stretch=128)
|
|
with pytest.raises(ValueError):
|
|
src.encode(frame_no=0x10000, rows_1bpp=bytes(8 * 160))
|
|
|
|
|
|
# --- geometry validation -------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("stretch", [0, 7, 1023, 1025, 2048])
|
|
def test_strip_source_rejects_impossible_stretch(stretch):
|
|
with pytest.raises(ValueError):
|
|
StripSource(rows=8, stretch=stretch)
|
|
|
|
|
|
def test_strip_source_rejects_rows_that_do_not_fit_the_header():
|
|
with pytest.raises(ValueError):
|
|
StripSource(rows=0, stretch=128)
|
|
with pytest.raises(ValueError):
|
|
StripSource(rows=256, stretch=4)
|
|
|
|
|
|
def test_strip_source_caps_rows_at_the_ring_not_the_header_field():
|
|
"""The terminal's 16-datagram ring, not the u8 ``rows`` field, is the limit.
|
|
|
|
144 rows is 23,040 body bytes: 16 datagrams of 1450 with the 16-byte header
|
|
in the first. A 145th row would need a 17th datagram, which the receiver
|
|
drops (RX_DESC in vidstream.c), so the encoder must refuse it here -- with a
|
|
message that names the cap -- rather than put an unreceivable frame on the
|
|
wire.
|
|
"""
|
|
src = StripSource(rows=MAX_ROWS, stretch=8)
|
|
pkts = src.encode(frame_no=1, rows_1bpp=bytes(MAX_ROWS * 160))
|
|
assert len(pkts) == RING_DATAGRAMS == 16
|
|
with pytest.raises(ValueError, match=str(MAX_ROWS)):
|
|
StripSource(rows=MAX_ROWS + 1, stretch=8)
|