"""Execute existing composition/settings methods with explicit fake infrastructure. Not an SDK/HTTP/DB integration test: exact method ASTs are isolated so no server singleton, GPU binding, PostgreSQL connection or native capture can start. """ from copy import deepcopy from pathlib import Path from types import ModuleType, SimpleNamespace import json import sys import threading import unittest from unittest.mock import Mock, patch from app import region_scenario from app.intrusion import Intrusion from app.loitering import Loitering from app.stream_settings import revision, settings_view from app.event_artifact_contract import ArtifactConfig from loitering_contract_helpers import node_function from test_loitering_settings import spec def module(name,**entries): m=ModuleType(name);m.__dict__.update(entries);return m class CompositionTests(unittest.TestCase): def prepare(self,values): fn=node_function('app/stream_manager.py','_prepare_det_results',owner='StreamGroup') s=dict(spec(),**values) group=SimpleNamespace(spec=s,stream_id='stream',metadata_publisher=None, lifecycle=SimpleNamespace(snapshot=lambda:{'run_id':'run'}), _generation_active=lambda *args:True,_fail_run=Mock()) made=[] def path_factory(*args,**kwargs): value=SimpleNamespace(start=Mock(),close=Mock(),**kwargs) made.append(value);return value def ingress(*args,**kwargs):return SimpleNamespace(sampler=SimpleNamespace(fps=10),metrics=Mock()) overrides={ 'app.frame_ingress':module('app.frame_ingress',FrameIngress=ingress,require_identity_binding=lambda:None), 'app.metadata_visual':module('app.metadata_visual',MetadataVisual=SimpleNamespace(from_environment=lambda **kw:None)), 'app.event_capture':module('app.event_capture',EventCapture=SimpleNamespace(from_environment=lambda value:None)), } with patch.dict(sys.modules,overrides),patch('app.det_result_path.DetResultPath',side_effect=path_factory),patch('app.byte_tracking.ByteTracker',return_value=Mock()): fn(group,SimpleNamespace(roles=['det'],resolved_models={'det':'RFDETRMedium'},score_threshold=.5),1,threading.Event(),Mock()) made[0].start.assert_called_once();self.assertIs(group.result_path,made[0]) return made[0] def test_stream_group_composes_loitering(self): self.assertIsInstance(self.prepare({'loitering':{'enabled':True}}).intrusion,Loitering) def test_stream_group_composes_original_intrusion(self): self.assertIsInstance(self.prepare({'intrusion':{'enabled':True}}).intrusion,Intrusion) def test_stream_group_off(self): self.assertIsNone(self.prepare({}).intrusion) def test_capture_gate_accepts_either_but_does_not_start_when_disabled(self): gate=node_function('app/event_capture.py','from_environment',{'ArtifactConfig':ArtifactConfig},owner='EventCapture') for key in ('intrusion','loitering'): factory=Mock() with patch.object(ArtifactConfig,'from_environment',return_value=SimpleNamespace(enabled=True)): gate(factory,dict(spec(),**{key:{'enabled':True}})) factory.assert_called_once() factory.reset_mock();gate(factory,spec());factory.assert_not_called() with patch.object(ArtifactConfig,'from_environment',return_value=SimpleNamespace(enabled=False)): gate(factory,dict(spec(),**{key:{'enabled':True}}));factory.assert_not_called() def test_status_names_selected_engine(self): fn=node_function('app/stream_manager.py','status',owner='StreamGroup') for key,expected in (('loitering','LOITERING'),('intrusion','INTRUSION')): engine=region_scenario.compose(dict(spec(),**{key:{'enabled':True}})) done=threading.Event();done.set() state=dict(state='running',desired_state='running',observed_state='running',run_generation=1, run_id='run',roles={},last_error=None,cleanup_error=None) group=SimpleNamespace(lifecycle=SimpleNamespace(snapshot=lambda:state),_lock=threading.Lock(),role_handles={}, stream_id='stream',site_id='site',source_type='rtsp',_cleanup_done=done,_pending_restart=None, _last_cleanup=None,started_at=None,profile=None,result_path=SimpleNamespace(intrusion=engine,status=lambda:{}), scenario_runner=None,collector=None) self.assertEqual(fn(group)['scenarios']['active'],[expected]) class ManagerCASMethods(unittest.TestCase): def setUp(self): self.fn=node_function('app/stream_manager.py','patch_settings',{'deepcopy':deepcopy},owner='StreamManager') self.saved=spec();self.actions=[] self.group=SimpleNamespace(spec=deepcopy(self.saved),_teardown_lock=threading.RLock(), observed_state='stopped',_pending_restart=None, check_restart=lambda candidate:region_scenario.validate(candidate), _apply_spec=self.apply_spec,restart=lambda candidate:self.actions.append('restart')) self.manager=SimpleNamespace(get=lambda sid:self.group,_load_base=lambda:[], _saved_spec=lambda group:deepcopy(self.saved),config_store=SimpleNamespace(update=self.persist), settings=lambda sid:settings_view(self.saved)) def apply_spec(self,candidate): self.actions.append('apply');self.group.spec=deepcopy(candidate) def persist(self,change): overlay={'added':[deepcopy(self.saved)],'removed':[]};change(overlay) self.saved=overlay['added'][0];self.actions.append('persist') def call(self,changes,expected=None): return self.fn(self.manager,'stream',revision(self.saved) if expected is None else expected,changes) def test_stopped_persists_before_apply_without_start(self): self.call({'loitering':{'enabled':True,'dwell_seconds':8}}) self.assertEqual(self.actions,['persist','apply']) self.assertEqual(self.saved['loitering']['dwell_seconds'],8) def test_running_persists_before_controlled_restart(self): self.group.observed_state='running';self.call({'loitering':{'enabled':True}}) self.assertEqual(self.actions,['persist','restart']) def test_stale_revision_never_persists(self): with self.assertRaises(ValueError):self.call({'loitering':{'enabled':True}},'0'*64) self.assertFalse(self.actions) def test_save_failure_keeps_effective_state(self): before=deepcopy(self.group.spec) self.manager.config_store.update=Mock(side_effect=OSError('fixture save failure')) with self.assertRaises(OSError):self.call({'loitering':{'enabled':True}}) self.assertEqual(self.group.spec,before);self.assertFalse(self.actions) def test_explicit_stop_pending_blocks_change(self): self.group.observed_state='stopping' with self.assertRaises(RuntimeError):self.call({'loitering':{'enabled':True}}) self.assertFalse(self.actions) def test_noop_does_not_restart(self): self.group.observed_state='running';self.call({'loitering':{'enabled':False,'dwell_seconds':5}}) self.assertFalse(self.actions) def test_read_only_base_is_not_overwritten(self): self.manager._load_base=lambda:[spec()] with self.assertRaises(ValueError):self.call({'loitering':{'enabled':True}}) self.assertFalse(self.actions) if __name__=='__main__':unittest.main()