diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 5c1cd35..145dba4 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -8,7 +8,7 @@ import time import numpy as np import pytest -from orisys import FrameSnapshot, ProducedFrame, StreamPipeline +from orisys import FrameSnapshot, PauseLevel, ProducedFrame, StreamPipeline class FakeSensor: @@ -651,3 +651,126 @@ def test_legacy_sensor_with_only_reset_is_reset_on_the_flow_stage(): assert sensor.reset_calls == 1 assert sensor.reset_threads == ["orisys-pipeline-flow"] assert sensor.reset_at_frame == [0], "the fallback reset must precede compute_flow" + + +class PauseResetFakeSensor(FakeSensor): + """A split-reset fake whose flow stage parks at chosen frame numbers. + + Parking inside ``compute_flow`` is what makes a pause deterministic: the + flow stage has already decided the current frame's version and cannot claim + a new request until the gate is released. + """ + + def __init__(self, *, gates=(1,)) -> None: + super().__init__() + self._gate_calls = {int(gate) for gate in gates} + self.gate_reached = {call: threading.Event() for call in self._gate_calls} + self.release_gate = {call: threading.Event() for call in self._gate_calls} + self.hhd_calls = 0 + self.image_reset_threads: list[str] = [] + self.force_reset_threads: list[str] = [] + self.force_reset_at_hhd_calls: list[int] = [] + + def compute_flow(self, img=None, **kwargs) -> None: + FakeSensor.compute_flow(self, img, **kwargs) + call = self._flow_calls + if call in self._gate_calls: + self.gate_reached[call].set() + self.release_gate[call].wait(timeout=5.0) + + def compute_hhd(self, flow=None, **kwargs) -> None: + self.hhd_calls += 1 + FakeSensor.compute_hhd(self, flow, **kwargs) + + def reset_image_state(self) -> None: + self.image_reset_threads.append(threading.current_thread().name) + + def reset_force_state(self) -> None: + self.force_reset_threads.append(threading.current_thread().name) + self.force_reset_at_hhd_calls.append(self.hhd_calls) + + def reset(self) -> None: + # Mirrors Sensor.reset(): both halves immediately, on the caller's thread. + self.reset_calls += 1 + self.reset_at_frame.append(self._flow_calls) + self.reset_image_state() + self.reset_force_state() + + +def test_resume_coalesces_with_a_reset_request_queued_while_paused(): + """A request queued while flow cannot claim it shares the resume's target.""" + sensor = PauseResetFakeSensor(gates=(1,)) + frames: list[FrameSnapshot] = [] + pipe = StreamPipeline( + sensor, make_producer(sensor, 8), on_frame=frames.append, depth=2 + ) + pipe.start() + try: + assert sensor.gate_reached[1].wait(timeout=5.0), "flow never reached its gate" + pipe.pause(PauseLevel.CAPTURE_ONLY) + assert pipe.tracker_reset_seq == 0 + request = pipe.request_reset() + assert request == 1 + pipe.resume(reset=True) + # The flow stage is still parked, so resume's own request coalesces. + assert pipe.request_reset() == request, ( + "resume allocated a new target although the flow stage had not claimed " + "the queued request" + ) + sensor.release_gate[1].set() + assert _wait_until( + lambda: any(snap.tracker_reset_seq >= request for snap in frames) + ), "the pipeline never published the resumed reference version" + versions = [snap.tracker_reset_seq for snap in frames] + assert 0 in versions, versions + assert max(versions) == request, versions + assert sensor.image_reset_threads == ["orisys-pipeline-flow"] + assert sensor.force_reset_threads == ["orisys-pipeline-hhd"] + finally: + sensor.release_gate[1].set() + pipe.stop() + + +def test_resume_after_a_claimed_reset_gets_a_new_target(): + """Once flow has claimed a request, resume's reset must not merge into it.""" + sensor = PauseResetFakeSensor(gates=(1, 3)) + frames: list[FrameSnapshot] = [] + pipe = StreamPipeline( + sensor, make_producer(sensor, 10), on_frame=frames.append, depth=2 + ) + pipe.start() + try: + assert sensor.gate_reached[1].wait(timeout=5.0), "flow never reached gate 1" + first = pipe.request_reset() + assert first == 1 + sensor.release_gate[1].set() + assert _wait_until( + lambda: any(snap.tracker_reset_seq >= first for snap in frames) + ), "the first requested version was never published" + assert sensor.gate_reached[3].wait(timeout=5.0), "flow never reached gate 3" + assert pipe.tracker_reset_seq == first + + pipe.pause(PauseLevel.CAPTURE_ONLY) + second = pipe.request_reset() + assert second == first + 1 + pipe.resume(reset=True) + assert pipe.request_reset() == second, ( + "resume merged into a target the flow stage had already claimed" + ) + sensor.release_gate[3].set() + assert _wait_until( + lambda: any(snap.tracker_reset_seq >= second for snap in frames) + ), "the resumed version was never published" + versions = [snap.tracker_reset_seq for snap in frames] + assert {first, second}.issubset(set(versions)), versions + assert max(versions) == second, versions + # Each new reference version resets the image and the force state once. + assert len(sensor.image_reset_threads) == 2, sensor.image_reset_threads + assert len(sensor.force_reset_threads) == 2, sensor.force_reset_threads + assert sensor.force_reset_at_hhd_calls == sorted( + sensor.force_reset_at_hhd_calls + ), sensor.force_reset_at_hhd_calls + finally: + sensor.release_gate[1].set() + sensor.release_gate[3].set() + pipe.stop()