import json import tempfile import threading import time import unittest from dataclasses import FrozenInstanceError from pathlib import Path from types import SimpleNamespace from unittest.mock import Mock, patch import numpy as np from app.contracts.frames import FrameEnvelope, DetectionObservation from app.contracts.visualization import FrameResult, VisualizationCommand from app.det_result_path import DetResultPath from app.frame_ingress import FrameIngress from app.metadata_runtime import publisher_from_environment from app.metadata_publisher import MetadataPublisher, MetadataSettings from app.metadata_visual import MetadataVisual from app.native_metadata import NativeFrame, NativeDetection from app.stream_manager import StreamManager, StreamGroup from app.stream_config_store import ConfigPersistenceError from app.application.streams import StreamApplicationService def wait_for(fn): deadline = time.monotonic() + 2 while not fn() and time.monotonic() < deadline: time.sleep(.005) assert fn(), 'bounded wait expired' def frame(seq=0, run='run', generation=1): return FrameEnvelope('server', 'camera', 'stream', run, generation, seq, 0, 100, 160, 90) def result(f=None): f = f or frame() return FrameResult(f, tuple(VisualizationCommand( DetectionObservation(f, 'RFDETRMedium', 0, score, (.1, .1, .5, .5)), 'person', visible=score > .1) for score in (.05, .2))) class ConnectionTest(unittest.TestCase): def test_production_retriever_pairs_exact_envelope_and_existing_image(self): from app import multi_pipeline as mp value = frame() pixels = np.zeros((90, 160, 3), dtype=np.uint8) visual = Mock() visual.wants_image.return_value = True path = SimpleNamespace(ingress=Mock(copy=Mock(return_value=value)), visual=visual, submit=Mock()) fm = SimpleNamespace(buffer_pts=0, frame_number=0, source_width=160, source_height=90) buffer = SimpleNamespace(batch_meta=SimpleNamespace(frame_items=[fm]), extract=Mock(return_value='tensor')) copied = NativeFrame(0, 0, 160, 90, (), 160, 90) receiver = mp._make_retriever('det', 'unused', 'unused', Mock(), 0, {'n': 0}, component_id=7, result_path=path)() with patch.object(mp, 'copy_detection_frame', return_value=copied), \ patch.object(mp, '_frame_to_hwc', return_value=pixels) as extract: receiver.consume(buffer) extract.assert_called_once() # Opt-in visual only; permanent Snapshot cache removed. self.assertIs(visual.submit_image.call_args.args[0], value) self.assertIs(visual.submit_image.call_args.args[1], pixels) self.assertIs(path.submit.call_args.args[0], value) def test_shared_result_visible_projection_keeps_object_reference(self): value = result() with self.assertRaises(FrozenInstanceError): value.frame = frame(1) wire = value.snapshot() self.assertEqual([x['objectRef'] for x in wire['objects']], ['1']) self.assertEqual(len(value.commands), 2) wire['objects'].clear() self.assertEqual(len(value.snapshot()['objects']), 1) self.assertEqual(len(value.snapshot(visible_only=False)['objects']), 2) def test_path_tracks_all_but_collector_and_display_are_independent(self): tracker = Mock(update=lambda obs: [None] * len(obs)) collector, errors = Mock(), [] publisher = MetadataPublisher(MetadataSettings('unused'), sink=Mock()) path = DetResultPath(FrameIngress('server', 'camera', 'stream', 'run', 1), tracker, 'RFDETRMedium', .5, collector, lambda: True, lambda: errors.append(1), display_threshold=.1, publisher=publisher) self.addCleanup(lambda: path.close(1)) path.start() copied = NativeFrame(0, 0, 160, 90, tuple(NativeDetection(0, s, (.1, .1, .5, .5)) for s in (.05, .2)), 160, 90) path.submit(frame(), copied) wait_for(lambda: path.status().get('processed') == 1) self.assertIs(path.latest_result(), path.latest_result()) self.assertEqual(len(path.latest_result().commands), 2) self.assertEqual(collector.add.call_args.args[3]['count'], 0) payload = next(iter(publisher._pending.values()))[1] self.assertEqual(len(json.loads(payload)['objects']), 1) self.assertEqual(errors, []) path.submit(frame(1), NativeFrame(1, 0, 160, 90, (), 160, 90)) wait_for(lambda: path.status().get('processed') == 2) self.assertEqual(path.latest_result().snapshot()['objects'], []) path.close(1) self.assertIsNone(path.latest_result()) def test_metadata_env_no_implicit_connection_or_secret_exposure(self): self.assertIsNone(publisher_from_environment({})) with self.assertRaisesRegex(ValueError, 'Authenticated'): publisher_from_environment({'CONTEXTRA_METADATA_ENABLED': 'true', 'REDIS_URL': 'redis://internal'}) env = {'CONTEXTRA_METADATA_ENABLED': 'true', 'CONTEXTRA_METADATA_REDIS_URL': 'redis://writer:private@localhost/0'} publisher = publisher_from_environment(env) self.assertIsNone(publisher.sink.client) self.assertIsNone(publisher._thread) self.assertNotIn('private', repr(publisher.settings)) env['CONTEXTRA_METADATA_MAX_BYTES'] = 'private' with self.assertRaises(ValueError) as raised: publisher_from_environment(env) self.assertNotIn('private', str(raised.exception)) def test_registered_stopped_channel_has_no_fabricated_run(self): publisher = MetadataPublisher(MetadataSettings('unused'), sink=Mock()) self.addCleanup(publisher.close) self.assertTrue(publisher.channel('camera', None, 0, 'STOPPED')) self.assertIsNone(next(iter(publisher._channels.values()))['runId']) self.assertTrue(publisher.channel('camera', 'run', 1, 'STARTING')) self.assertFalse(publisher.channel('camera', None, 0, 'STOPPED')) def test_composition_injects_existing_and_new_groups_closes_once(self): with tempfile.TemporaryDirectory() as root: manager = StreamManager(root+'/base', root+'/overlay') before = manager.add({'stream_id': 'one'}) publisher = Mock() manager.configure_metadata(publisher) after = manager.add({'stream_id': 'two'}) self.assertIs(before.metadata_publisher, publisher) self.assertIs(after.metadata_publisher, publisher) publisher.start.assert_called_once() with self.assertRaises(RuntimeError): manager.configure_metadata(Mock()) self.assertTrue(manager.close_metadata()) publisher.close.assert_called_once_with(1) def test_status_heartbeat_and_connection_failure_do_not_claim_frames(self): now = [0.] sink = Mock() p = MetadataPublisher(MetadataSettings('unused', stream_ttl=3), sink, clock=lambda: now[0]) self.addCleanup(p.close) p.channel('camera', 'run', 1, 'STOPPED') p.start() wait_for(lambda: p.metrics().get('sent_status') == 1) now[0] = 2 with p._cv: p._cv.notify() wait_for(lambda: p.metrics().get('sent_status', 0) >= 2) self.assertEqual(p.metrics().get('sent_frames', 0), 0) class SettingsTest(unittest.TestCase): def setUp(self): self.root = tempfile.TemporaryDirectory() self.addCleanup(self.root.cleanup) self.manager = StreamManager(self.root.name+'/base', self.root.name+'/overlay') self.g = self.manager.add(dict(stream_id='stream', source_id='camera', server_id='server', uri='private-secret', store=False, inference_profile={'thresholds': {'score': .5}, 'roles': ['det']})) def test_patch_preserves_identity_and_does_not_start_stopped(self): before = dict(self.g.spec) rev = self.manager.settings('stream')['configured']['revision'] with patch.object(self.g, 'start', side_effect=AssertionError('unexpected start')): response = self.manager.patch_settings('stream', rev, {'display_score': .1}) self.assertEqual(response['effective']['downstream_score'], .5) self.assertEqual(response['configured']['display_score'], .1) self.assertEqual({k: v for k, v in self.g.spec.items() if k != 'visualization'}, before) self.assertNotIn('private', str(response)) with self.assertRaises(ValueError): self.manager.patch_settings('stream', rev, {'display_score': .2}) with patch.object(self.manager.config_store, 'update', side_effect=AssertionError('same config write')): self.manager.patch_settings('stream', response['configured']['revision'], {'display_score': .1}) def test_save_failure_and_invalid_input_preserve_execution(self): view = self.manager.settings('stream') rev = view['configured']['revision'] for changes in ({'display_score': None}, {'display_score': float('nan')}, {'server_id': 'x'}, {}): with self.assertRaises(ValueError): self.manager.patch_settings('stream', rev, changes) with patch.object(self.manager.config_store, 'update', side_effect=ConfigPersistenceError()): with self.assertRaises(ConfigPersistenceError): self.manager.patch_settings('stream', rev, {'display_score': .1}) self.assertEqual(view, self.manager.settings('stream')) from app.stream_settings import patch_settings nullable = dict(self.g.spec, visualization=None, inference_profile=None) changed = patch_settings(nullable, {'display_score': .1, 'downstream_score': .5}) self.assertEqual(changed['visualization']['score_threshold'], .1) self.assertEqual(changed['inference_profile']['thresholds']['score'], .5) def test_http_cas_null_unknown_and_save_failure(self): from fastapi import FastAPI from fastapi.testclient import TestClient from app.api.streams import router app = FastAPI() app.include_router(router) service = StreamApplicationService(lambda: self.manager) with patch('app.api.streams.stream_service', service), TestClient(app) as client: url = '/api/v1/streams/stream/settings' rev = client.get(url).json()['configured']['revision'] body = dict(expected_revision=rev, display_score=.1) with patch.object(self.manager.config_store, 'update', side_effect=ConfigPersistenceError()): self.assertEqual(client.patch(url, json=body).status_code, 503) self.assertEqual(client.patch(url, json={**body, 'uri': 'private'}).status_code, 422) self.assertEqual(client.patch(url, json={**body, 'display_score': True}).status_code, 422) self.assertEqual(client.patch(url, json={**body, 'display_score': None}).status_code, 409) response = client.patch(url, json=body) self.assertEqual(response.status_code, 200) self.assertNotIn('private', response.text) self.assertEqual(client.patch(url, json=body).status_code, 409) def test_concurrent_cas_only_one_save(self): rev = self.manager.settings('stream')['configured']['revision'] outcomes = [] def change(value): try: self.manager.patch_settings('stream', rev, {'display_score': value}) outcomes.append('ok') except ValueError: outcomes.append('conflict') workers = [threading.Thread(target=change, args=(x,)) for x in (.1, .2)] for worker in workers: worker.start() for worker in workers: worker.join(1) self.assertFalse(worker.is_alive()) self.assertCountEqual(outcomes, ['ok', 'conflict']) def test_running_patch_waits_for_old_worker_then_applies_once(self): from app.stream_lifecycle import RoleRunHandle gen = self.g.lifecycle.begin() self.g.lifecycle.configure_roles(gen, ['det']) self.g.lifecycle.signal_frame(gen, 'det') release, called = threading.Event(), threading.Event() handle = RoleRunHandle('det', gen) handle.bind_pipeline(SimpleNamespace(stop=called.set)) worker = threading.Thread(target=lambda: release.wait(3)) handle.bind_thread(worker) worker.start() self.g.role_handles['det'] = handle def begin(group, generation, stop): group.lifecycle.configure_roles(generation, ['det']) group.lifecycle.signal_frame(generation, 'det') try: with patch.object(StreamGroup, '_begin', begin): rev = self.manager.settings('stream')['configured']['revision'] view = self.manager.patch_settings('stream', rev, {'display_score': .1}) self.assertTrue(called.wait(1)) self.assertTrue(view['restart_pending']) self.assertEqual(view['effective']['display_score'], .5) self.assertEqual(self.g.run_generation, gen) release.set() self.assertTrue(self.g._cleanup_done.wait(2)) with self.g._teardown_lock: self.assertEqual(self.g.run_generation, gen+1) self.assertEqual(self.manager.settings('stream')['effective']['display_score'], .1) self.assertFalse(self.g.lifecycle.signal_frame(gen, 'det')) self.g.stop() self.assertTrue(self.g._cleanup_done.wait(2)) finally: release.set() worker.join(1) def test_explicit_stop_cancels_autostart_but_saved_settings_apply_on_next_start(self): from app.stream_lifecycle import RoleRunHandle gen = self.g.lifecycle.begin() release = threading.Event() handle = RoleRunHandle('det', gen) handle.bind_pipeline(SimpleNamespace(stop=lambda: None)) worker = threading.Thread(target=lambda: release.wait(3)) worker.start() handle.bind_thread(worker) self.g.role_handles['det'] = handle try: rev = self.manager.settings('stream')['configured']['revision'] self.manager.patch_settings('stream', rev, {'display_score': .1}) self.manager.stop('stream') self.assertFalse(self.g.status()['restart_pending']) release.set() self.assertTrue(self.g._cleanup_done.wait(2)) view = self.manager.settings('stream') self.assertEqual(view['configured']['display_score'], .1) self.assertEqual(view['effective']['display_score'], .5) with patch.object(StreamGroup, '_begin'): self.manager.start('stream') self.assertEqual(self.manager.settings('stream')['effective']['display_score'], .1) self.manager.stop('stream') self.assertTrue(self.g._cleanup_done.wait(2)) finally: release.set() worker.join(1) class VisualTest(unittest.TestCase): def make_visual(self, **kwargs): root = tempfile.TemporaryDirectory() self.addCleanup(root.cleanup) path = Path(root.name) (path/'capture.enabled').touch(mode=0o600) visual = MetadataVisual(path, **kwargs) self.addCleanup(lambda: visual.close(1)) return visual, path def test_exact_pair_real_encoding_and_output_matches_wire(self): import cv2 visual, root = self.make_visual() value = result() pixels = np.full((90, 160, 3), 100, dtype=np.uint8) visual.submit_image(value.frame, pixels) pixels[:] = 0 visual.submit_result(value) wait_for(lambda: visual.status().get('saved') == 1) paths = list(root.glob('capture-*/*.json')) self.assertEqual(len(paths), 1) self.assertEqual(json.loads(paths[0].read_text()), value.snapshot()) rendered = cv2.imread(str(paths[0].with_suffix('.jpg'))) from app.metadata_render import HEADER self.assertEqual(rendered.shape, (360 + HEADER, 640, 3)) self.assertGreater(rendered.mean(), 50) self.assertGreater(rendered.std(), 1) self.assertEqual(paths[0].stat().st_mode & 0o777, 0o600) self.assertEqual(paths[0].parent.stat().st_mode & 0o777, 0o700) def test_same_pts_foreign_run_mismatch_expiry_and_cap(self): now = [0.] visual, root = self.make_visual(clock=lambda: now[0]) pixels = np.zeros((90, 160, 3), dtype=np.uint8) visual.submit_image(frame(), pixels) visual.submit_result(result(frame(1))) visual.submit_result(result(frame(0, 'other', 2))) self.assertLessEqual(visual.status()['queued'], 32) self.assertEqual(visual.status().get('saved', 0), 0) now[0] = 3 wait_for(lambda: visual.status()['queued'] == 0) self.assertFalse(visual.submit_image(frame(), pixels[:20])) self.assertEqual(visual.status()['geometry_mismatch'], 1) (root/'capture.enabled').unlink() wait_for(lambda: not visual.enabled()) self.assertFalse(visual.submit_result(result())) self.assertTrue(visual.close(1)) self.assertFalse(visual.status()['worker_alive']) def test_write_failure_is_nonfatal_and_bounded(self): visual, _ = self.make_visual() with patch.object(visual, '_save', side_effect=OSError('private')): visual.submit_image(frame(), np.zeros((90, 160, 3), dtype=np.uint8)) visual.submit_result(result()) wait_for(lambda: visual.status().get('write_failed') == 1) self.assertNotIn('private', str(visual.status())) def test_three_frames_and_sixty_seconds_are_hard_capture_limits(self): now = [0.] visual, _ = self.make_visual(clock=lambda: now[0]) with patch.object(visual, '_save'): for seq in range(3): visual.submit_result(result(frame(seq))) visual.submit_image(frame(seq), np.zeros((90, 160, 3), dtype=np.uint8)) wait_for(lambda: visual.status().get('saved', 0) == seq+1) self.assertFalse(visual.wants_image(frame(3))) self.assertFalse(visual.submit_result(result(frame(3)))) other, _ = self.make_visual(clock=lambda: now[0]) now[0] = 61 self.assertFalse(other.wants_image(frame())) def test_close_timeout_retains_thread_then_finishes(self): visual, _ = self.make_visual() entered, release = threading.Event(), threading.Event() with patch.object(visual, '_save', side_effect=lambda *a: (entered.set(), release.wait(2))): visual.submit_image(frame(), np.zeros((90, 160, 3), dtype=np.uint8)) visual.submit_result(result()) self.assertTrue(entered.wait(1)) self.assertFalse(visual.close(.001)) self.assertTrue(visual._thread.is_alive()) release.set() self.assertTrue(visual.close(1))