import threading import time import unittest from dataclasses import replace from unittest.mock import Mock from app.contracts.frames import FrameEnvelope, DetectionObservation from app.contracts.visualization import VisualizationCommand, frame_snapshot from app.metadata_publisher import MetadataPublisher, MetadataSettings, RedisSnapshotSink, camera_key from app.application.camera_commands import CameraCommands, CommandUnavailable def frame(seq=0, generation=1): return FrameEnvelope("server", "001", "stream", "run" + str(generation), generation, seq, None, 10, 100, 100) def wait_until(predicate): deadline = time.monotonic() + 1 while not predicate() and time.monotonic() < deadline: time.sleep(0.001) assert predicate() class SnapshotTest(unittest.TestCase): def test_empty_and_null(self): result = frame_snapshot(frame(), "001") self.assertEqual(result["objects"], []) self.assertIsNone(result["frameTime"]["rtpTimestamp"]) self.assertIsNone(result["frameTime"]["detectedAtWallMs"]) def test_command_copy_and_identity(self): f = frame() d = DetectionObservation(f, "medium", 0, .7, (0, 0, 1, 1)) c = VisualizationCommand(d, "person", event_ids=("event",)) result = frame_snapshot(f, "001", [c]) result["objects"][0]["bbox"][0] = .5 self.assertEqual(d.bbox_xyxy[0], 0) self.assertIsNone(result["objects"][0]["trackId"]) with self.assertRaises(ValueError): frame_snapshot(frame(1), "001", [c]) with self.assertRaises(ValueError): frame_snapshot(f, 1) class PublisherTest(unittest.TestCase): def make(self, **options): self.sink = Mock() self.publisher = MetadataPublisher(MetadataSettings("unused", **options), self.sink) self.addCleanup(self.publisher.close) self.publisher.channel("001", "run1", 1, "RUNNING") return self.publisher def test_constructor_no_io_and_latest(self): p = self.make() self.sink.write.assert_not_called() p.submit(frame_snapshot(frame(), "001")) p.submit(frame_snapshot(frame(1), "001")) self.assertEqual(p.metrics()["queued"], 1) self.assertEqual(p.metrics()["replaced"], 1) p.start() wait_until(lambda: p.metrics().get("sent") == 1) self.assertIn('"frameSeq":1', self.sink.write.call_args.args[2]) def test_fences_limits_and_stop(self): p = self.make(max_message_bytes=2) self.assertFalse(p.submit(frame_snapshot(frame(), "001"))) self.assertEqual(p.metrics()["message_limit"], 1) self.assertFalse(p.channel("001", "other", 1, "RUNNING")) p.channel("001", "run1", 1, "STOPPED") self.assertFalse(p.submit(frame_snapshot(frame(1), "001"))) self.assertFalse(p.channel("001", "run1", 1, "RUNNING")) self.assertTrue(p.channel("001", "run2", 2, "RUNNING")) self.assertFalse(p.submit(frame_snapshot(frame(2), "001"))) def test_byte_cap_and_channel_cap(self): p = self.make(max_bytes=2, max_channels=1) self.assertFalse(p.submit(frame_snapshot(frame(), "001"))) self.assertEqual(p.metrics()["queuedBytes"], 0) self.assertFalse(p.channel("other", "r", 1, "RUNNING")) def test_stale_age_and_failure(self): p = self.make() now = [0] p.clock = lambda: now[0] p.submit(frame_snapshot(frame(), "001")) now[0] = 2 self.sink.write.side_effect = OSError("private connection details") p.start() wait_until(lambda: p.metrics().get("send_error") == 1) self.assertEqual(p.metrics()["expired"], 1) self.assertIsNone(self.sink.write.call_args.args[2]) def test_fairness_and_shutdown_pending_io(self): p = self.make() entered, release = threading.Event(), threading.Event() def block(*args): entered.set() release.wait(1) self.sink.write.side_effect = block p.start() self.assertTrue(entered.wait(1)) p.channel("other", "r", 1, "RUNNING") p.submit(frame_snapshot(frame(), "001")) p.submit(frame_snapshot(frame(1), "001")) self.assertEqual(list(p._pending), [camera_key("other"), camera_key("001")]) self.assertFalse(p.close(.001)) release.set() self.assertTrue(p.close(1)) self.assertFalse(p.submit(frame_snapshot(frame(2), "001"))) def test_sink_transaction_and_no_duplicate_global_frame(self): client = Mock() sink = RedisSnapshotSink(MetadataSettings("unused"), client) sink.write("key", dict(runId="r", state="RUNNING"), "{}") client.pipeline.assert_called_once_with(transaction=True) pipe = client.pipeline.return_value pipe.xadd.assert_called_once() pipe.set.assert_called_once() pipe.execute.assert_called_once() sink.write("key", dict(runId="r", state="STOPPED"), None) self.assertEqual(pipe.xadd.call_count, 1) self.assertEqual(pipe.delete.call_count, 2) def test_stop_follows_inflight_and_discards_pending_frame(self): p = self.make() entered, release = threading.Event(), threading.Event() writes = [] def write(key, status, payload): if not writes: entered.set() release.wait(1) writes.append((status["state"], payload)) self.sink.write.side_effect = write p.submit(frame_snapshot(frame(), "001")) p.start() self.assertTrue(entered.wait(1)) p.submit(frame_snapshot(frame(1), "001")) p.channel("001", "run1", 1, "STOPPED") release.set() wait_until(lambda: len(writes) == 2) self.assertEqual(writes[-1], ("STOPPED", None)) self.assertEqual(p.metrics()["queuedBytes"], 0) def test_failure_recovers_on_next_frame_not_automatic_retry(self): p = self.make() self.sink.write.side_effect = [OSError(), None] p.start() wait_until(lambda: p.metrics().get("send_error") == 1) p.submit(frame_snapshot(frame(), "001")) wait_until(lambda: p.metrics().get("sent") == 1) self.assertEqual(self.sink.write.call_count, 2) def test_close_failure_is_sticky(self): p = self.make() self.sink.close.side_effect = OSError("private detail") p.start() self.assertFalse(p.close(1)) self.assertFalse(p.close(1)) self.assertEqual(p.metrics()["close_error"], 1) class CommandsTest(unittest.TestCase): def test_dedupe_busy_shutdown_and_late_result(self): service = Mock() entered, release = threading.Event(), threading.Event() def work(request): entered.set() release.wait(1) return "done" service.start_camera.side_effect = work commands = CameraCommands(service, capacity=1, history=1) self.addCleanup(commands.close) job = commands.submit("001", "request") self.assertTrue(entered.wait(1)) self.assertIs(commands.submit("001", "request"), job) with self.assertRaises(CommandUnavailable): commands.submit("other", "request") with self.assertRaises(CommandUnavailable): commands.submit("001", action="stop") self.assertFalse(commands.close(.001)) release.set() self.assertEqual(job.future.result(1), "done") self.assertTrue(commands.close(1)) self.assertIs(commands.get(job.operation_id), job) def test_queued_cancel_and_failure_result(self): service = Mock() gate = threading.Event() service.start_camera.side_effect = lambda _: gate.wait(1) commands = CameraCommands(service, capacity=2) self.addCleanup(commands.close) job = commands.submit("1", "request") wait_until(lambda: service.start_camera.called) pending = commands.submit("2", "request") commands.close(.001) self.assertRaises(CommandUnavailable, pending.future.result) gate.set() job.future.result(1)