"""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_lsb_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 0, pixel 7 -> bit 7, pixel 8 -> bit 0 of byte 1: the # terminal's scanout reads bit 0 of a byte as its leftmost pixel # (dwt/screenshot.py decodes live VRAM the same way). The plan carried an # MSB-first pair here (0b11000001 / 0b10000000), which is self-consistent # with an MSB-first checker but mirrors every eight pixels on a display. assert b[0] == 0b10000001 and b[1] == 0b00000001 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 _decode_like_the_terminal(row: bytes) -> list[int]: """Decode a packed row the way the VT1200 scanout does. The same rule ``dwt/screenshot.py`` uses on live VRAM: bit ``x % 8`` of byte ``x // 8`` is pixel ``x`` -- LSB first within each byte. """ return [(row[x // 8] >> (x % 8)) & 1 for x in range(1280)] def test_pack_row_decodes_back_to_the_pixels_the_terminal_shows(): """A row must survive the terminal's own LSB-first scanout decode. The failure this pins: an MSB-first packer is self-consistent with an MSB-first checker, so every MAME replay check passes while the screen mirrors every group of eight pixels. Decoding the packed row with the hardware's rule is the only reading that can catch it. """ px = [1 if i % 3 == 0 else 0 for i in range(1280)] px[0] = 0 # asymmetric ends, so a mirror cannot pass px[1279] = 1 assert _decode_like_the_terminal(Rows.pack_row(px)) == px def test_pack_row_puts_every_pixel_where_the_terminal_reads_it(): """Pixel i is bit (i % 8) of byte i // 8, i.e. LSB 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] >> (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) # The u8 header field allows 255 rows; the ring ceiling is what a huge row # count hits first, and MAX_ROWS is that ceiling (289 with a 32-deep ring). with pytest.raises(ValueError): StripSource(rows=MAX_ROWS + 1, stretch=4) def test_strip_source_caps_rows_at_the_ring_not_the_header_field(): """The terminal's 32-datagram ring, not the u8 ``rows`` field, is the limit. The ring was deepened to 32 descriptors when the dictionary upload outgrew 16 (vidstream.c, SLOT_BYTES), so the strip encoder may now send 288 rows: 32 datagrams of 1450 with the 16-byte header in the first. One row more would need a 33rd 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)) # MAX_ROWS is now the u8 header field (255), which is fewer datagrams than # the 32-deep ring could carry -- so the frame fits with room to spare. assert len(pkts) <= RING_DATAGRAMS == 32 with pytest.raises(ValueError, match=str(MAX_ROWS)): StripSource(rows=MAX_ROWS + 1, stretch=8)