"""Latest-AU H.264 mmap: overwrite unread payloads; seq is monotonic.""" from nal_shm import AU_MAX, NalReader, NalWriter def test_writer_seq_starts_at_one_and_increments(tmp_path): path = str(tmp_path / "cam-01.h264") w = NalWriter(path, max_payload=64) assert w.write(b"\x00\x00\x00\x01\x65" + b"x" * 10) == 1 assert w.write(b"\x00\x00\x00\x01\x41" + b"y" * 10) == 2 w.close() def test_reader_returns_latest_payload(tmp_path): path = str(tmp_path / "cam-01.h264") w = NalWriter(path, max_payload=64) w.write(b"first-access-unit-xxxx") w.write(b"second-access-unit-yyy") r = NalReader(path) seq, payload = r.read() assert seq == 2 assert payload.startswith(b"second-access-unit") w.close() r.close() def test_reader_none_when_no_new_au(tmp_path): path = str(tmp_path / "cam-01.h264") w = NalWriter(path, max_payload=64) w.write(b"only-once") r = NalReader(path) assert r.read()[0] == 1 assert r.read() is None w.close() r.close() def test_write_drops_oversize_payload(tmp_path): path = str(tmp_path / "cam-01.h264") w = NalWriter(path, max_payload=8) assert w.write(b"123456789") == 0 w.close() def test_au_max_is_bounded(): assert AU_MAX >= 64 * 1024 assert AU_MAX <= 2 * 1024 * 1024