"""SCENARIO-FANOUT-01: fixture-only final worker/control/transport boundaries.""" from copy import deepcopy from dataclasses import replace import json from pathlib import Path from types import SimpleNamespace import unittest from unittest.mock import Mock, patch from app.region_scenario import compose from app.fire_scenario import Fire, FireScenarioConfig from app.fire_config import FIRE_MODEL from app.pose_config import execution_plan, planned_profile from app.provider_plan import plan from app.secure_control import ControlService, Start, ControlError from app.stream_settings import patch_settings from app.det_result_path import DetResultPath from app.metadata_samples import MetadataSamples from app.event_artifact_contract import build_body, relative_directory from control_sidecar.main import validate_event from test_loitering_settings import FakeManager, request, spec from test_loitering_connection import ResultWorkerTests, CopiedDetection, Sink from loitering_contract_helpers import WALL_MS class FanoutTests(unittest.TestCase): make = ResultWorkerTests.make feed = ResultWorkerTests.feed def test_both_events_independent_same_frame_one_tracker_call(self): engine = compose(dict(spec(), intrusion={'enabled': True, 'dwell_seconds': .5}, loitering={'enabled': True, 'dwell_seconds': .5})) path, sink, errors = self.make(engine) calls = Mock(wraps=path.tracker.update) path.tracker.update = calls for i in range(6): result = self.feed(path, i, i/10) self.assertEqual(calls.call_count, 6) self.assertEqual(len(result.events), 2) snap = sink.rows[-1] self.assertEqual(set(snap['scenarioConfig']), {'INTRUSION', 'LOITERING'}) self.assertEqual(len(set(snap['objects'][0]['eventIds'])), 2) self.assertEqual({e['scenario'] for e in snap['events']}, {'INTRUSION', 'LOITERING'}) self.assertEqual(len(snap['objects'][0]['scenarioResults']), 2) self.assertEqual(snap['objects'][0]['color'], '#FF3030') self.assertFalse(self.feed(path, 6, .6).events) path.close(2) self.assertFalse(path._thread.is_alive()) self.assertTrue(all(not e.states for e in engine.engines)) errors.assert_not_called() def test_roi_is_independent_and_alert_color_is_union(self): engine = compose(dict(spec(), intrusion={'enabled': True, 'dwell_seconds': .5}, loitering={'enabled': True, 'dwell_seconds': .5, 'roi': [[.6, 0], [1, 0], [1, 1], [.6, 1]]})) path, sink, _ = self.make(engine) for i in range(6): result = self.feed(path, i, i/10) self.assertEqual([e.payload()['scenario'] for e in result.events], ['INTRUSION']) self.assertEqual([d['state'] for d in sink.rows[-1]['objects'][0]['scenarioResults']], ['INTRUSION', 'OUTSIDE']) self.assertEqual(sink.rows[-1]['objects'][0]['color'], '#FF3030') class WebTests(unittest.TestCase): def test_medium_template_fire_and_return_noop(self): template = json.loads(Path('configs/control_templates/medium.json').read_text()) manager = FakeManager() service = ControlService(manager, 'server', template) fire = service.start_camera(Start(json.dumps(request('FIRE')))) self.assertEqual(execution_plan(fire)['model'], FIRE_MODEL) self.assertEqual(planned_profile(fire)['det_model'], FIRE_MODEL) self.assertTrue(fire['tracking']['enabled']) self.assertTrue(fire['fire']['scenario']['enabled']) manager = FakeManager(fire) service = ControlService(manager, 'server', template) self.assertEqual(fire, service.start_camera(Start(json.dumps(request('FIRE'))))) person = service.start_camera(Start(json.dumps(request('INTRUSION', 'LOITERING')))) self.assertEqual(execution_plan(person)['model'], 'RFDETRMedium') self.assertTrue(person['tracking']['enabled']) self.assertEqual(person['visualization']['categories'], ['person']) self.assertFalse(person['secure_control_projection']['skippedScenarios']) def test_fire_combines_with_person_and_pose_scenarios(self): for names in [('FIRE','INTRUSION'), ('FIRE','LOITERING'), ('FIRE','FALL')]: saved = spec() manager = FakeManager(saved) service = ControlService(manager, 'server', saved) value = service.start_camera(Start(json.dumps(request(*names)))) planned = plan(value) self.assertIn('fire', planned.keys) if 'FALL' in names: self.assertIn('pose', planned.keys) else: self.assertIn('medium', planned.keys) self.assertTrue(manager.applied) def test_fire_partial_settings_preserves_timers(self): manager = FakeManager() s = ControlService(manager, 'server', spec()).start_camera(Start(json.dumps(request('FIRE')))) s = patch_settings(s, {'fire': {'scenario': {'flame_window_seconds': 3}}}) s = patch_settings(s, {'fire': {'scenario': {'flame_confidence': .7}}}) self.assertEqual(s['fire']['scenario']['flame_window_seconds'], 3) class FireWorkerTests(unittest.TestCase): feed = ResultWorkerTests.feed def test_fire_no_tracking_smoke_hidden_event_ref_matches(self): engine = Fire(FireScenarioConfig(enabled=True), wall_ns=lambda: WALL_MS*1_000_000) ingress = SimpleNamespace(identity=('server', '77', 'stream', 'run', 1)) sink, errors = Sink(), Mock() with patch.object(MetadataSamples, 'from_environment', return_value=None): path = DetResultPath(ingress, None, FIRE_MODEL, .5, Mock(add=Mock(return_value=1)), lambda: True, errors, publisher=sink, fire=engine, camera_id=77) path.start() self.addCleanup(lambda: path.close(2)) for i in range(11): result = self.feed(path, i, i/10, [CopiedDetection(class_id=0, score=.7), CopiedDetection(class_id=1, score=.5), CopiedDetection(class_id=1, score=.9)]) snap = sink.rows[-1] self.assertEqual(len(snap['objects']), 3) self.assertEqual(len(result.events), 1) self.assertTrue(all(o['trackId'] is None for o in snap['objects'])) self.assertEqual({o['classId'] for o in snap['objects']},{0,1}) event = snap['events'][0] self.assertEqual(event['objectRef'], '2') self.assertEqual(event['bbox'], snap['objects'][2]['bbox']) self.assertEqual(snap['scenarioConfig']['FIRE']['roi_name'], 'Fire ROI') errors.assert_not_called() if __name__ == '__main__': unittest.main()