diff --git a/tests/test_perception_streaming_ingress.py b/tests/test_perception_streaming_ingress.py index 5ca5268..3677f95 100644 --- a/tests/test_perception_streaming_ingress.py +++ b/tests/test_perception_streaming_ingress.py @@ -8,7 +8,7 @@ from dataclasses import replace import pytest -from k1link.compute.live_perception import LiveIngressEvent +from k1link.compute.live_perception import LiveIngressEvent, LivePerceptionIngress from k1link.perception import streaming_wire as wire from k1link.perception.graph_contracts import GraphState from k1link.perception.realtime_contract import StreamStart @@ -311,3 +311,28 @@ def test_sender_timeout_closes_transport_instead_of_building_backlog(): finally: left.close() right.close() + + +def test_existing_raw_first_ingress_events_feed_new_writer_without_wire_v1_changes(tmp_path): + source = LivePerceptionIngress() + source.begin_session("capture-1") + source.open_consumer("stream-adapter") + assert source.take_next("stream-adapter", timeout=0).modality == "control" + assert source.publish( + modality="lidar", + source_id="existing-raw-topic", + source_sequence=12, + captured_at_epoch_ns=1799999999123456789, + received_monotonic_ns=9007199254740993, + payload=b"unchanged raw payload", + ) + raw = source.take_next("stream-adapter", timeout=0) + legacy = raw.wire_bytes() + with harness(tmp_path) as (sock, receiver, _, received, _): + sender = StreamingSender(sock, start(), "capture-1", 1, lambda: None) + sender.send(raw) + sender.end() + assert receiver.join() and receiver.terminal == "end" + assert received == [raw] and raw.wire_bytes() == legacy + source.close_consumer("stream-adapter") + source.close()