import json import os import queue import threading import tempfile import unittest from unittest.mock import patch from app.channel_process import ProcessSlots from app.metadata_publisher import MetadataPublisher, MetadataSettings, camera_key from app.provider_delivery import ProviderPublisher, SnapshotRelay, provider_spec, runtime_plan from app.provider_plan import plan from app.provider_process_group import ProviderProcessGroup from app.stream_config_store import ConfigPersistenceError from app.stream_manager import StreamManager from tests.test_channel_process import fixture_child, wait def spec(): return dict(stream_id='camera-3', source_id='3', camera_id=3, server_id='server', uri='unused', pose=dict(enabled=True, show_joints=True), fall=dict(enabled=True), fight=dict(enabled=True), fire=dict(enabled=True, scenario=dict(enabled=True)), crowd=dict(enabled=True), tracking=dict(enabled=True), store=False, active_learning=False, scenarios='off') def provider_fixture(*args, snapshot_output=None, snapshot_limit=None): child_spec = args[0] if child_spec.get('_test_crash_fire') and child_spec['_provider_runtime']['provider'].key == 'fire': os._exit(9) fixture_child(*args) class ProviderProcessTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.slots = ProcessSlots(lock_path=self.temp.name+'/owner.lock') self.addCleanup(self.slots.close) def test_projection_keeps_identity_and_only_owned_scenarios(self): original = spec() selected = plan(original) self.assertEqual(selected.keys, ('pose', 'fire', 'crowd')) for provider in selected.providers: child = provider_spec(original, provider, selected.keys) self.assertEqual(runtime_plan(child).keys, (provider.key,)) self.assertEqual(child['camera_id'], 3) self.assertEqual(child['source_id'], '3') self.assertEqual(child['stream_id'], original['stream_id']) self.assertEqual(child['tracking']['enabled'], provider.person_owner) self.assertEqual(child['fall']['enabled'], provider.key == 'pose') self.assertEqual(child['fire']['scenario']['enabled'], provider.key == 'fire') self.assertEqual(child['crowd']['enabled'], provider.key == 'crowd') self.assertTrue(original['fall']['enabled']) def test_provider_queue_and_sequence_are_independent_and_terminal_clears(self): publisher = MetadataPublisher(MetadataSettings('unused'), sink=object()) publisher.channel('3', 'run', 1, 'RUNNING') base = dict(cameraId=3, runId='run', runGeneration=1) for provider, seq in [('pose',100), ('fire',1), ('crowd',1), ('pose',101)]: self.assertTrue(publisher.submit(dict(base, providerKey=provider, frameSeq=seq))) key = camera_key('3') self.assertEqual({k[1] for k in publisher._pending if isinstance(k,tuple)}, {'pose','fire','crowd'}) self.assertFalse(publisher.submit(dict(base, providerKey='fire',frameSeq=1))) self.assertFalse(publisher.submit(dict(base, providerKey='fire',frameSeq=2,runId='old'))) publisher.channel('3','run',1,'STOPPED') self.assertEqual(list(publisher._pending),[key]) self.assertEqual(publisher._bytes,0) self.assertFalse(publisher.submit(dict(base,providerKey='fire',frameSeq=3))) def test_relay_has_three_fields_and_preserves_numeric_camera_and_pts(self): class Output(queue.Queue): def cancel_join_thread(self): pass output=Output(maxsize=1) relay=SnapshotRelay(output,('3','run',1),128*1024) provider=plan(spec()).providers[1] publisher=ProviderPublisher(relay,provider,('pose','fire','crowd')) value=dict(cameraId=3,runId='run',runGeneration=1,frameSeq=4,ptsNs=123,objects=[]) self.assertTrue(publisher.submit(value)) got=json.loads(output.get_nowait()) self.assertEqual(got['providerKey'],'fire') self.assertEqual(got['providerModel'],provider.model) self.assertEqual(got['requiredProviders'],['pose','fire','crowd']) self.assertEqual(got['ptsNs'],123) self.assertNotIn('providerKey',value) self.assertFalse(publisher.submit(dict(value,cameraId=4))) def test_split_cas_noop_and_save_failure_do_not_start(self): with patch.dict(os.environ,CONTEXTRA_STREAM_EXECUTION_MODE='process'): manager=StreamManager(self.temp.name+'/base',self.temp.name+'/settings.yaml') try: group=manager.add(spec()) current=manager.settings(group.stream_id)['configured']['revision'] manager.patch_settings(group.stream_id,current,{'analysisFpsTarget':10}) self.assertEqual(manager.settings(group.stream_id)['configured']['revision'],current) with patch.object(manager.config_store,'update',side_effect=ConfigPersistenceError()): with self.assertRaises(ConfigPersistenceError): manager.patch_settings(group.stream_id,current,{'display_score':.4}) self.assertEqual(manager.settings(group.stream_id)['configured']['revision'],current) self.assertFalse(group._split_active) self.assertEqual(group._children,{}) with self.assertRaises(ValueError): group.check_restart(dict(spec(),store=True)) finally: self.assertTrue(manager.shutdown_channels()) def test_split_duplicate_start_stop_reap_and_single_restart(self): group=ProviderProcessGroup(spec(),slots=self.slots,child_target=provider_fixture) with patch('app.provider_process_group.run_channel',provider_fixture): try: t=threading.Thread(target=group.start);t.start();t.join(3) wait(lambda:group.state=='running' and len(group._children)==3) wait(lambda:all(c.state=='running' for c in group._children.values())) run=group.lifecycle.snapshot()['run_id'] pids={c._process.pid for c in group._children.values()} self.assertEqual(len(pids),3) self.assertEqual(group.start(),1) self.assertTrue(all(c.lifecycle.snapshot()['run_id']==run for c in group._children.values())) group.stop();self.assertTrue(group._cleanup_done.wait(10)) self.assertTrue(group._last_cleanup['completed']) self.assertTrue(all(c._process is None for c in group._children.values())) single=dict(spec(),fire={'enabled':False},crowd={'enabled':False}) group.restart(single) wait(lambda:group.state=='running' and group.run_generation==2) self.assertIsNone(group._split_plan) finally: group.stop();self.assertTrue(group._cleanup_done.wait(10)) def test_local_crash_degrades_without_stopping_other_providers(self): group=ProviderProcessGroup(dict(spec(),_test_crash_fire=True),slots=self.slots) with patch('app.provider_process_group.run_channel',provider_fixture): try: group.start() wait(lambda:group.state=='running' and group.status().get('degraded')) self.assertTrue(group.spec['fire']['scenario']['enabled']) self.assertIn(dict(name='FIRE',reason='provider_unavailable'),group.status()['scenarios']['disabled']) self.assertEqual(group._children['pose'].state,'running') self.assertEqual(group._children['crowd'].state,'running') finally: group.stop();self.assertTrue(group._cleanup_done.wait(10)) if __name__ == '__main__': unittest.main()