"""PR13 CPU-only contracts. No DeepStream/Triton/GPU/runtime start.""" import unittest from types import SimpleNamespace from app.contracts.frames import FrameEnvelope from app.provider_plan import ProviderPlan, ProviderSpec, plan from app.scenario_parameters import apply, sanitize from app.multi_provider_results import MultiProviderCoordinator class _Metrics: def snapshot(self): return {} def close(self): pass class _Ingress: def __init__(self): self.sampler = SimpleNamespace(fps=10.0) self.metrics = _Metrics() self.identity = ('server','1','stream','run',1) class _Collector: def __init__(self): import threading self.lock = threading.Lock() self.frames = {} def add(self, *args, **kwargs): return 0 class _Copied: width = 1280 height = 720 detections = () def frame(seq, pts=0): return FrameEnvelope('server','1','stream','run',1,seq,pts,1,1280,720) class ProviderPlanTest(unittest.TestCase): def test_all_six_uses_pose_fire_crowd_and_pose_owns_person(self): spec = dict( server_id='server', source_id='1', tracking={'enabled':True}, intrusion={'enabled':True}, loitering={'enabled':True}, fall={'enabled':True}, fight={'enabled':True}, fire={'enabled':True,'scenario':{'enabled':True}}, crowd={'enabled':True}, pose={'enabled':False}, store=False, active_learning=False, scenarios='off') value = plan(spec) self.assertEqual(value.person_owner, 'pose') self.assertEqual(value.keys, ('pose','fire','crowd')) self.assertEqual(set(value.requested_scenarios), {'INTRUSION','LOITERING','FALL','FIGHT','FIRE','CROWD'}) def test_medium_owns_person_without_pose_scenarios(self): value = plan(dict( server_id='server',source_id='1',tracking={'enabled':True}, intrusion={'enabled':True},loitering={'enabled':True}, store=False,active_learning=False,scenarios='off')) self.assertEqual(value.person_owner,'medium') self.assertEqual(value.keys,('medium',)) def test_fire_and_crowd_do_not_create_person_owner(self): value = plan(dict( server_id='server',source_id='1',tracking={'enabled':True}, fire={'enabled':True,'scenario':{'enabled':True}}, crowd={'enabled':True},store=False,active_learning=False,scenarios='off')) self.assertIsNone(value.person_owner) self.assertEqual(value.keys,('fire','crowd')) class ParameterTest(unittest.TestCase): def test_invalid_unknown_and_null_are_fail_soft(self): accepted, ignored = sanitize('FALL', { 'down_angle_deg':'bad', 'unknown':1, 'pose_confidence':None, 'recovery_hold_seconds':2}) self.assertEqual(accepted, {'recovery_hold_seconds':2.0}) self.assertEqual(len(ignored),3) def test_fall_and_fight_pose_confidence_are_scenario_local(self): spec = dict( fall={'enabled':True,'joint_score':.1}, fight={'enabled':True,'joint_score':.2}) self.assertTrue(apply(spec, { 'scenarioCode':'FALL','parameters':{'pose_confidence':.35}})) self.assertEqual(spec['fall']['joint_score'],.35) self.assertEqual(spec['fight']['joint_score'],.2) self.assertTrue(apply(spec, { 'scenarioCode':'FIGHT','parameters':{'pose_confidence':.4}})) self.assertEqual(spec['fall']['joint_score'],.35) self.assertEqual(spec['fight']['joint_score'],.4) def test_invalid_inter_field_override_rolls_back_whole_scenario(self): spec = {'fall': {'enabled':True,'down_angle_deg':50, 'knee_guard_torso_max_deg':45}} before = dict(spec['fall']) # 40 violates upright < knee torso < down; whole FALL override is reverted. self.assertFalse(apply(spec, { 'scenarioCode':'FALL','parameters':{ 'down_angle_deg':40,'recovery_hold_seconds':2}})) self.assertEqual(spec['fall'], before) class CoordinatorAcceptanceTest(unittest.TestCase): def make(self, clock): provider = ProviderSpec('medium','det','RFDETRMedium',('bbox',),(),False) p = ProviderPlan((provider,),None,()) spec = dict( stream_id='stream',server_id='server',source_id='1',camera_id=1, tracking={'enabled':True},store=False,active_learning=False, scenarios='off',visualization={'score_threshold':.25}) return MultiProviderCoordinator( spec,_Ingress(),p,_Collector(),lambda:True, lambda:None,clock=clock) def test_late_result_is_health_response_but_not_frame_result(self): now=[0.0] c=self.make(lambda:now[0]) f=frame(0) self.assertTrue(c.admit(f)) now[0]=.101 self.assertFalse(c.submit('medium',f,_Copied(),response_at=now[0])) self.assertEqual(c._providers['medium']['responses'],1) self.assertEqual(c._providers['medium']['late'],1) self.assertEqual(c._providers['medium']['state'],'ACTIVE') def test_eviction_and_no_backward_acceptance(self): now=[0.0] c=self.make(lambda:now[0]) self.assertTrue(c.admit(frame(0))) now[0]=.01; self.assertTrue(c.admit(frame(1))) now[0]=.02; self.assertTrue(c.admit(frame(2))) self.assertNotIn(0,c._frames) self.assertEqual(c._metrics['frame_evicted'],1) self.assertFalse(c.submit('medium',frame(0),_Copied(),response_at=.02)) def test_stall_needs_input_and_no_response_for_two_seconds(self): now=[0.0] c=self.make(lambda:now[0]) c.startup_s = 2.0 c.admit(frame(0)) now[0]=1.9 with c._cv: self.assertEqual(c._check_stalls_locked(now[0]),[]) now[0]=2.0 with c._cv: self.assertEqual(c._check_stalls_locked(now[0]),['medium']) def test_timing_preserves_acceptance_and_separates_stages(self): now = [0.0] c = self.make(lambda: now[0]) f = frame(0) c.admit(f) self.assertEqual(c.provider_gate_time('medium', 0, .02), 0.) now[0] = .08 self.assertTrue(c.submit('medium', f, _Copied(), .08, timing=(0., .02, .03, .06, .08))) rows = c.status()['provider_timing']['medium'] self.assertAlmostEqual(rows['selected_to_gate']['sum_ms'], 20) self.assertAlmostEqual(rows['gate_to_infer']['sum_ms'], 10) self.assertAlmostEqual(rows['infer_to_callback']['sum_ms'], 30) self.assertAlmostEqual(rows['callback_copy']['sum_ms'], 20) self.assertEqual(c._metrics['accepted_result'], 1) rows['callback_copy']['count'] = -1 self.assertEqual(c.status()['provider_timing']['medium']['callback_copy']['count'], 1) def test_diagnostic_rejection_reasons_and_history_are_bounded(self): now = [0.0] c = self.make(lambda: now[0]) c.admit(frame(0)) now[0] = .101 self.assertFalse(c.submit('medium', frame(0), _Copied())) self.assertEqual(c._metrics['rejected_medium_deadline'], 1) for seq in range(1, 100): c.admit(frame(seq)) self.assertLessEqual(len(c._retired), 64) self.assertFalse(c.submit('medium', frame(97), _Copied())) self.assertEqual(c._metrics['rejected_medium_evicted'], 1) c._retire_locked(98, 'finalizing') c._frames.pop(98) self.assertFalse(c.submit('medium', frame(98), _Copied())) self.assertEqual(c._metrics['rejected_medium_finalizing'], 1) if __name__ == '__main__': unittest.main()