from __future__ import annotations import json from pathlib import Path import pytest from k1link.mqtt.capture import FRAME_HEADER, RAW_MAGIC from k1link.viewer.replay import ReplayFormatError, detect_replay_format, iter_replay_messages def _write_native(path: Path, topic: str, payload: bytes) -> None: topic_raw = topic.encode() path.write_bytes( RAW_MAGIC + FRAME_HEADER.pack(len(topic_raw), len(payload)) + topic_raw + payload ) def test_native_replay_uses_aligned_metadata_timing(tmp_path: Path) -> None: capture = tmp_path / "mqtt.raw.k1mqtt" _write_native(capture, "lixel/application/report/lio_pose", b"pose") (tmp_path / "mqtt.metadata.jsonl").write_text( json.dumps( { "record_type": "message", "sequence": 1, "received_at_epoch_ns": 123_000_000_000, "received_monotonic_ns": 456, } ) + "\n" ) assert detect_replay_format(capture) == "k1mqtt" message = list(iter_replay_messages(capture))[0] assert message.topic.endswith("lio_pose") assert message.payload == b"pose" assert message.received_at_epoch_ns == 123_000_000_000 assert message.received_monotonic_ns == 456 def test_legacy_tsv_replay_validates_and_decodes_payload(tmp_path: Path) -> None: capture = tmp_path / "payloads.tsv" capture.write_bytes(b"1784124315.186225000\tRealtimePath\t4\t0001aaff\n") assert detect_replay_format(capture) == "legacy_tsv" message = list(iter_replay_messages(capture))[0] assert message.received_at_epoch_ns == 1_784_124_315_186_225_000 assert message.payload == b"\x00\x01\xaa\xff" @pytest.mark.parametrize( "line", [ b"1\tonly\tthree\n", b"bad\ttopic\t1\t00\n", b"1\ttopic\t2\t00\n", b"1\ttopic\t1\tzz\n", ], ) def test_legacy_tsv_rejects_unreviewed_shapes(tmp_path: Path, line: bytes) -> None: capture = tmp_path / "payloads.tsv" capture.write_bytes(line) with pytest.raises(ReplayFormatError): list(iter_replay_messages(capture))