import threading import time import unittest from dataclasses import replace from types import SimpleNamespace from unittest.mock import patch from app.provider_branch import LATEST_QUEUE, load_completion_relay from app.multi_provider_results import public_positions, MultiProviderCoordinator from app.contracts.frames import FrameEnvelope, DetectionObservation from app.contracts.visualization import VisualizationCommand from app.provider_plan import ProviderPlan, ProviderSpec from app.fire_scenario import Fire, FireScenarioConfig def frame(seq, seconds=0): return FrameEnvelope('server', '1', 'stream', 'run', 1, seq, int(seconds*1e9), 1, 1280, 720) class Boundaries(unittest.TestCase): def test_sdk_borrowed_frame_is_copied_before_iterator_advances(self): from app.single_decode_pipeline import copy_single_envelope wrapper = SimpleNamespace(value=None) def items(): wrapper.value = frame(7) yield wrapper wrapper.value = None class Ingress: def copy(self, borrowed): if borrowed.value is None: raise RuntimeError('Expired SDK wrapper') return borrowed.value result = copy_single_envelope(SimpleNamespace(frame_items=items()), Ingress()) self.assertEqual(result.frame_seq, 7) self.assertIsNone(wrapper.value) with self.assertRaisesRegex(RuntimeError, 'Expired SDK wrapper'): Ingress().copy(list(items())[0]) def test_sdk_single_frame_contract_is_preserved(self): from app.single_decode_pipeline import copy_single_envelope ingress = SimpleNamespace(copy=lambda value: value) for values in ((), (frame(1), frame(2))): with self.assertRaises(ValueError): copy_single_envelope(SimpleNamespace(frame_items=iter(values)), ingress) def test_default_detection_and_pose_display_keep_person_owner(self): from app.provider_plan import plan self.assertEqual(plan({}).person_owner,'medium') self.assertEqual(plan({'pose':{'enabled':True}}).person_owner,'pose') def test_huge_tuning_number_is_fail_soft(self): from app.scenario_parameters import sanitize accepted,ignored=sanitize('FALL',{'person_confidence':10**500}) self.assertEqual(accepted,{}) self.assertEqual(len(ignored),1) def test_queue_contract(self): self.assertEqual(LATEST_QUEUE, dict(**{'max-size-buffers':1, 'max-size-bytes':0,'max-size-time':0,'leaky':2})) def test_public_cap_keeps_event_fire_and_crowd_labels(self): f = frame(0) def command(model, cls=0, score=.9, label='box', visible=True): return VisualizationCommand(DetectionObservation( f,model,cls,score,(.1,.1,.3,.4)),label,visible=visible) commands = [command('RFDETRKeypointPreview', visible=False) for _ in range(300)] commands += [command('RFDETRFireSmoke',0),command('RFDETRFireSmoke',1)] commands += [command('Awiros_person_and_head_detection',0,label='person | P=3'), command('Awiros_person_and_head_detection',1,label='head | H=3')] events = [SimpleNamespace(object_ref='299')] selected = public_positions(commands,events) self.assertEqual(len(selected),300) self.assertTrue({299,300,301,302,303} <= set(selected)) self.assertEqual(selected,public_positions(commands,events)) def test_mandatory_overflow_is_explicit(self): f=frame(0) commands=[VisualizationCommand(DetectionObservation(f,'RFDETRMedium',0,.5,(0.,0.,1.,1.)),'person')]*301 with self.assertRaises(RuntimeError): public_positions(commands,[SimpleNamespace(object_ref=str(i)) for i in range(301)]) def test_fire_short_gap_does_not_advance_clear(self): cfg=FireScenarioConfig(enabled=True,clear_seconds=.4) fire=Fire(cfg) fire.event_id='episode' fire.evaluate(frame(0,0),()) fire.evaluate(frame(1,.3),(),gap_before=True) fire.evaluate(frame(2,.5),()) self.assertEqual(fire.event_id,'episode') fire.evaluate(frame(3,.7),()) self.assertIsNone(fire.event_id) def coordinator(self): from tests.test_pr13_multi_provider import _Ingress,_Collector p=ProviderSpec('fire','det','RFDETRFireSmoke',('bbox',),('FIRE',)) spec=dict(stream_id='stream',server_id='server',source_id='1',camera_id=1, fire={'enabled':True,'scenario':{'enabled':True}}, store=False,active_learning=False,scenarios='off') return MultiProviderCoordinator(spec,_Ingress(),ProviderPlan((p,),None,('FIRE',)), _Collector(),lambda:True,lambda:None) def test_stop_drains_claimed_event_before_close(self): from tests.test_pr13_multi_provider import _Copied c=self.coordinator();entered=threading.Event();release=threading.Event();handoff=[] original=c._finalize def finish(state): self.assertEqual(state.state,'FINALIZING') entered.set();self.assertTrue(release.wait(2));original(state);handoff.append(1) c._finalize=finish c.admit(frame(1));c.submit('fire',frame(1),_Copied()) c._thread=threading.Thread(target=c._run);c._thread.start() self.assertTrue(entered.wait(2)) done=threading.Event() t=threading.Thread(target=lambda:(c.close(2),done.set()));t.start() time.sleep(.02);self.assertFalse(done.is_set()) release.set();t.join(2) self.assertTrue(done.is_set());self.assertEqual(handoff,[1]) def test_eviction_marks_fire_gap_and_disable_removes_only_open_results(self): from tests.test_pr13_multi_provider import _Copied c=self.coordinator() for seq in range(3):c.admit(frame(seq)) self.assertTrue(c._gap_pending['fire']) c.submit('fire',frame(2),_Copied()) c.mark_provider_unavailable('fire') self.assertNotIn('fire',c._frames[2].results) class GstRelay(unittest.TestCase): def test_actual_cpu_gst_relay(self): import gi gi.require_version('Gst','1.0') from gi.repository import Gst Gst.init(None) load_completion_relay() pipeline=Gst.parse_launch('fakesrc num-buffers=3 ! queue max-size-buffers=1 ! contextra_completion_relay name=relay ! fakesink name=out signal-handoffs=true') seen=[] relay=pipeline.get_by_name('relay') sink=pipeline.get_by_name('out') sink.connect('handoff',lambda *args:(seen.append(1),relay.set_property('complete',True))) try: pipeline.set_state(Gst.State.PLAYING) msg=pipeline.get_bus().timed_pop_filtered(3*Gst.SECOND,Gst.MessageType.ERROR|Gst.MessageType.EOS) self.assertIsNotNone(msg) self.assertEqual(msg.type,Gst.MessageType.EOS) self.assertEqual(len(seen),3) finally: relay.set_property('disabled',True);pipeline.set_state(Gst.State.NULL) def test_actual_latest_waiting_and_slow_branch_isolation(self): import gi gi.require_version('Gst','1.0') from gi.repository import Gst Gst.init(None);load_completion_relay() pipeline=Gst.parse_launch( 'appsrc name=source is-live=true format=time ! tee name=t ' 't. ! queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=2 ! ' 'contextra_completion_relay name=slow ! fakesink name=s sync=false async=false signal-handoffs=true ' 't. ! queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=2 ! ' 'contextra_completion_relay name=fast ! fakesink name=f sync=false async=false signal-handoffs=true') source=pipeline.get_by_name('source');slow=pipeline.get_by_name('slow');fast=pipeline.get_by_name('fast') first=threading.Event();slow_seen=[];fast_seen=[] def slow_output(sink,buffer,pad): slow_seen.append(buffer.pts);first.set() if len(slow_seen)>1:slow.set_property('complete',True) pipeline.get_by_name('s').connect('handoff',slow_output) pipeline.get_by_name('f').connect('handoff',lambda sink,b,p:(fast_seen.append(b.pts),fast.set_property('complete',True))) try: pipeline.set_state(Gst.State.PLAYING) b=Gst.Buffer.new();b.pts=1;source.emit('push-buffer',b) self.assertTrue(first.wait(2)) for pts in (2,3,4): b=Gst.Buffer.new();b.pts=pts;source.emit('push-buffer',b) deadline=time.monotonic()+2 while 4 not in fast_seen and time.monotonic()