import os import threading import time import unittest from dataclasses import replace from unittest.mock import patch from app.native_metadata import NativeFrame from app.provider_linear_results import LinearProviderResults, linear_trial_enabled from app.provider_linear_pipeline import LINEAR_QUEUE from app.provider_plan import ProviderPlan, ProviderSpec from tests.test_pr13_multi_provider import _Ingress, _Collector, frame class LinearTest(unittest.TestCase): def make(self, key='medium'): self.now=[1.] provider=ProviderSpec(key,'det','RFDETRMedium' if key=='medium' else 'RFDETRFireSmoke', ('bbox',),() if key=='medium' else ('FIRE',),key=='medium') spec=dict(stream_id='stream',server_id='server',source_id='1',camera_id=1, tracking={'enabled':False},store=False,active_learning=False,scenarios='off') if key=='fire':spec['fire']={'enabled':True,'scenario':{'enabled':True}} with patch.dict(os.environ,CONTEXTRA_PROVIDER_FRESHNESS_ENABLED='false'): c=LinearProviderResults(spec,_Ingress(),ProviderPlan((provider,),key if key=='medium' else None, provider.scenarios),_Collector(),lambda:True,lambda:None,clock=lambda:self.now[0]) c.ingress.result_gap=lambda f:False return c def submit(self,c,seq,key='medium'): f=frame(seq,seq*100000000) return c.submit(key,f,NativeFrame(seq,f.pts_ns,1280,720,(),1280,720),timing=(0.,)) def test_opt_in_and_on_rejection(self): with patch.dict(os.environ,{},clear=True):self.assertFalse(linear_trial_enabled()) with patch.dict(os.environ,CONTEXTRA_PROVIDER_LINEAR_TRIAL='bad'): with self.assertRaises(ValueError):linear_trial_enabled() c=self.make() with patch.dict(os.environ,CONTEXTRA_PROVIDER_FRESHNESS_ENABLED='true'): with self.assertRaises(ValueError): LinearProviderResults(c.spec,c.ingress,c.plan,c.collector,lambda:True,lambda:None) def test_no_preinfer_slots_and_owned_queue_is_bounded(self): c=self.make() for seq in range(1000):c.admit(frame(seq)) self.assertEqual(len(c._frames),0) for seq in range(40):self.assertTrue(self.submit(c,seq)) self.assertEqual(len(c._frames),32) self.assertEqual(c._metrics['completed_queue_drop'],8) self.assertTrue(c._gap_pending['medium']) self.assertFalse(self.submit(c,39)) def test_identity_pts_provider_and_stop_rejection(self): c=self.make();f=frame(0);copy=NativeFrame(0,0,1280,720,(),1280,720) for bad in (replace(f,run_id='foreign'),replace(f,pts_ns=1)): with self.assertRaises(ValueError):c.submit('medium',bad,copy,timing=(0.,)) with self.assertRaises(ValueError):c.submit('pose',f,copy,timing=(0.,)) c.stop_admission();self.assertFalse(c.submit('medium',f,copy,timing=(0.,))) def test_queue_age_is_after_completion_and_expiry_marks_gap(self): c=self.make();self.now[0]=10 self.assertTrue(self.submit(c,1));self.assertEqual(c._ready_locked(10.9),1) self.assertIsNone(c._ready_locked(11.01));self.assertTrue(c._gap_pending['medium']) self.assertEqual(c._metrics['completed_queue_expired'],1) self.assertEqual(c._metrics['over_freshness_medium'],1) def test_selected_fire_gap_is_carried_to_common_finalizer(self): c=self.make('fire');c.ingress.result_gap=lambda f:True self.submit(c,1,'fire');state=c._frames[1];state.state='FINALIZING' with patch('app.multi_provider_results.MultiProviderCoordinator._finalize') as finish: c._finalize(state) self.assertIn('fire',finish.call_args.args[0].gaps) def test_stop_waits_for_claimed_handoff_and_failure_is_not_hidden(self): c=self.make('fire');entered=threading.Event();release=threading.Event();handed=[] original=c._finalize def finalize(state): entered.set();self.assertTrue(release.wait(2));original(state);handed.append(1) c._finalize=finalize;self.submit(c,1,'fire') c._thread=threading.Thread(target=c._run);c._thread.start() self.assertTrue(entered.wait(2)) t=threading.Thread(target=lambda:c.close(2));t.start() time.sleep(.02);self.assertTrue(t.is_alive()) release.set();t.join(2);self.assertEqual(handed,[1]);self.assertFalse(t.is_alive()) def test_pipeline_has_no_isolation_or_completion_relay(self): import inspect from app.provider_linear_pipeline import run_provider_linear_pipeline source=inspect.getsource(run_provider_linear_pipeline) self.assertNotIn('.fork(',source);self.assertNotIn("'nvvideoconvert'",source) self.assertNotIn('contextra_completion_relay',source) self.assertEqual(LINEAR_QUEUE,{'max-size-buffers':200,'max-size-bytes':10485760, 'max-size-time':1000000000,'leaky':0}) def test_private_timing_travels_with_identity_not_public_envelope(self): from dataclasses import asdict from types import SimpleNamespace from tests.test_det_result_path import IdentityTest, Frame fixture=IdentityTest();fixture.setUp();self.addCleanup(fixture.doCleanups) f=Frame();selected=time.monotonic_ns() fixture.ingress.stamp(SimpleNamespace(frame_items=[f]),selected_monotonic_ns=selected) envelope, copied=fixture.ingress.copy(f,with_timing=True) self.assertEqual(copied,selected) self.assertNotIn('selected_monotonic_ns',asdict(envelope)) self.assertEqual(envelope,fixture.ingress.copy(f)) del f.meta[0].value['selected_monotonic_ns'] with self.assertRaises(ValueError):fixture.ingress.copy(f,with_timing=True) def test_backpressured_source_does_not_hide_outstanding_request_stall(self): c=self.make();c.admit(frame(0));c.note_request('medium') self.now[0]=11 self.assertEqual(c._check_stalls_locked(11),['medium']) self.submit(c,0);self.assertEqual(c._check_stalls_locked(14),[]) c.note_request('medium');self.assertEqual(c._check_stalls_locked(14),['medium']) self.now[0]=14;c.note_provider_progress('medium','infer_output') self.assertEqual(c._check_stalls_locked(14),[]) self.assertEqual(c._check_stalls_locked(16.01),['medium'])