"""Config/control contracts; no network or running StreamGroup is used.""" from copy import deepcopy import json import threading from types import SimpleNamespace import unittest from app import region_scenario from app.intrusion import Intrusion from app.loitering import Loitering, LoiteringConfig from app.stream_settings import patch_settings, settings_view, revision from app.secure_control import ControlService, Start, validate_start, ControlError, CODES def spec(): return dict(stream_id='stream', server_id='server', source_id='77', camera_id=77, store=False, active_learning=False, scenarios='off', tracking={'enabled': True}, inference_profile={'roles':['det'], 'det_model':'RFDETRMedium'}) def request(*enabled, rois=None): return dict(cameraId=77, rtspUrl='rtsp://fixture.invalid/video', startWallTimeMs=1758000000000, rtpClockRate=90000, scenarios=[dict(scenarioCode=n, enabled=n in enabled, confidence=None, rois=[] if rois is None else deepcopy(rois)) for n in sorted(CODES)]) class FakeManager: def __init__(self, saved=None): self._commands=threading.RLock();self.applied=[] self.group=None if saved is None else SimpleNamespace(spec=deepcopy(saved),observed_state='running',_pending_restart=None) def camera_stream_id(self, camera): if self.group is None:raise KeyError(camera) return self.group.spec['stream_id'] def get(self,sid):return self.group def _saved_spec(self,g):return deepcopy(g.spec) def apply_and_start(self,s):self.applied.append(deepcopy(s));return s class SettingsTests(unittest.TestCase): def test_default_off_and_factory(self): s=spec();self.assertIsNone(region_scenario.compose(s)) self.assertFalse(settings_view(s)['loitering']['enabled']) s=patch_settings(s,{'loitering':{'enabled':True}}) self.assertIsInstance(region_scenario.compose(s),Loitering) def test_original_intrusion_factory(self): self.assertIsInstance(region_scenario.compose(dict(spec(),intrusion={'enabled':True})),Intrusion) def test_noop_revision_stable(self): s=spec();n=patch_settings(s,{'loitering':{'dwell_seconds':5.,'enabled':False}}) self.assertEqual(s,n);self.assertEqual(revision(s),revision(n)) def test_partial_patch_persists_values_without_source_mutation(self): s=patch_settings(spec(),{'loitering':{'enabled':True,'dwell_seconds':9.,'grace_seconds':.7}}) before=deepcopy(s);n=patch_settings(s,{'loitering':{'confidence':.6,'point_y':.9}}) self.assertEqual(s,before);self.assertEqual(n['loitering']['dwell_seconds'],9.) self.assertEqual(n['loitering']['grace_seconds'],.7) self.assertNotEqual(revision(s),revision(n)) def test_invalid_keeps_original(self): s=spec();before=deepcopy(s) for v in ({'surprise':1},{'roi':[]},{'enabled':1},{'dwell_seconds':float('nan')},None): with self.subTest(v=v),self.assertRaises(ValueError):patch_settings(s,{'loitering':v}) self.assertEqual(s,before) def test_tracking_required(self): s=patch_settings(spec(),{'loitering':{'enabled':True}}) with self.assertRaises(ValueError):patch_settings(s,{'tracking':{'enabled':False}}) def test_shared_provider_fanout(self): s=patch_settings(spec(),{'loitering':{'enabled':True},'intrusion':{'enabled':True}}) self.assertIsInstance(region_scenario.compose(s), region_scenario.RegionFanout) def test_switch_both_directions_atomic_candidate(self): s=patch_settings(spec(),{'intrusion':{'enabled':True}}) for change in ({'loitering':{'enabled':True},'intrusion':{'enabled':False}}, {'intrusion':{'enabled':False},'loitering':{'enabled':True}}): self.assertIsInstance(region_scenario.compose(patch_settings(s,change)),Loitering) s=patch_settings(s,{'intrusion':{'enabled':False},'loitering':{'enabled':True}}) self.assertIsInstance(region_scenario.compose(patch_settings(s,{'loitering':{'enabled':False},'intrusion':{'enabled':True}})),Intrusion) def test_single_person_provider(self): s=patch_settings(dict(spec(),pose={'enabled':True}),{'loitering':{'enabled':True}}) self.assertIsInstance(region_scenario.compose(s),Loitering) for extra in ({'fire':{'enabled':True}}, {'inference_profile':{'roles':['det','pose']}}, {'inference_profile':{'roles':['det'],'det_model':'other'}}): s=dict(spec(),**extra) with self.subTest(extra=extra),self.assertRaises(ValueError):patch_settings(s,{'loitering':{'enabled':True}}) def test_identity_required(self): for k in ('server_id','source_id'): s=spec();s.pop(k) with self.assertRaises(ValueError):patch_settings(s,{'loitering':{'enabled':True}}) def test_state_not_transferred_after_setting_change(self): old=region_scenario.compose(dict(spec(),loitering={'enabled':True})) old.states[7]=object() new=region_scenario.compose(patch_settings(dict(spec(),loitering={'enabled':True}),{'loitering':{'dwell_seconds':8}})) self.assertFalse(new.states);self.assertEqual(new.config.dwell_seconds,8) class WebProjectionTests(unittest.TestCase): def project(self,d,saved=None): m=FakeManager(saved);svc=ControlService(m,'server',spec()) out=svc.start_camera(Start(json.dumps(d)));return m,out def test_existing_fields_map_first_roi(self): rois=[dict(label='first',points=[[.1,.1],[.9,.1],[.9,.9],[.1,.9]]), dict(label='second',points=[[0,0],[1,0],[1,1]])] _,s=self.project(request('LOITERING',rois=rois)) self.assertEqual(s['loitering']['roi_name'],'first') self.assertEqual(s['loitering']['roi'],rois[0]['points']) self.assertEqual(s['loitering']['dwell_seconds'],5.) self.assertNotIn('LOITERING',s['secure_control_projection']['skippedScenarios']) def test_timers_survive_repeated_web_start(self): old=dict(spec(),loitering=LoiteringConfig(enabled=True,dwell_seconds=9,grace_seconds=.8).payload()) _,a=self.project(request('LOITERING'),old) _,b=self.project(request('LOITERING'),a) self.assertEqual(a,b);self.assertEqual(a['loitering']['dwell_seconds'],9) self.assertEqual(a['loitering']['grace_seconds'],.8) def test_missing_null_empty_rois_equivalent(self): out=[] for mode in ('missing','null','empty'): d=request('LOITERING') for r in d['scenarios']: if mode=='missing':r.pop('rois') elif mode=='null':r['rois']=None _,s=self.project(d);out.append(s) self.assertEqual(out[0],out[1]);self.assertEqual(out[0],out[2]) def test_disable_preserves_local_timers(self): old=dict(spec(),loitering={'enabled':True,'dwell_seconds':8.}) _,s=self.project(request(),old) self.assertFalse(s['loitering']['enabled']);self.assertEqual(s['loitering']['dwell_seconds'],8.) def test_disabled_absent_does_not_change_legacy_spec(self): _,s=self.project(request('INTRUSION')) self.assertNotIn('loitering',s);self.assertTrue(s['intrusion']['enabled']) def test_invalid_combination_does_not_apply(self): m=FakeManager();svc=ControlService(m,'server',spec()) with self.assertRaises(ControlError):svc.start_camera(Start(json.dumps(request('INTRUSION','FIRE')))) self.assertFalse(m.applied) def test_web_timer_fields_still_rejected(self): d=request('LOITERING');next(x for x in d['scenarios'] if x['scenarioCode']=='LOITERING')['dwell_seconds']=3 with self.assertRaises(ControlError):validate_start(d) def test_other_scenarios_remain_skipped(self): _,s=self.project(request('LOITERING','FIGHT')) self.assertEqual(set(s['secure_control_projection']['skippedScenarios']),{'FIGHT'}) def test_fullframe_replaces_old_roi_name(self): old=dict(spec(),loitering={'enabled':True,'roi_name':'old'}) _,s=self.project(request('LOITERING'),old) self.assertEqual(s['loitering']['roi_name'],'Loitering ROI') def test_invalid_roi_cannot_persist(self): d=request('LOITERING',rois=[dict(label='cross',points=[[0,0],[1,1],[1,0],[0,1]])]) m=FakeManager();svc=ControlService(m,'server',spec()) with self.assertRaises(ControlError):svc.start_camera(Start(json.dumps(d))) self.assertFalse(m.applied) if __name__=='__main__':unittest.main()