root
95bb7c50ee
Publish and subscribe over wss://livekit.uni-wh.de:7800 and refuse cleartext ws://. Conference room is uwh-telhai. Includes the uncommitted encoded H.264 publish path, Rally hairpin, and KMS wall overlay.
47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
"""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
|