from copy import deepcopy import importlib.util import json from pathlib import Path import tempfile import unittest from unittest.mock import Mock import yaml spec=importlib.util.spec_from_file_location('test_run_manager',Path(__file__).resolve().parents[1]/'tools/test_run_manager.py') m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m) class FakeApi: def __init__(self,registered=True): self.registered=registered;self.calls=[];self.state='running';self.run='run1';self.processed=5;self.pending=False self.conf=dict(revision='a'*64,analysisFpsTarget=10,display_score=.25,frame_time_anchor=None,intrusion={'enabled':False,'confidence':.5,'roi_name':'existing'},display_categories=['person']) self.fail_patch=False;self.hold=False def call(self,method,path,body=None): self.calls.append((method,path,body)) if path=='/api/v1/streams':return {'streams':[{'stream_id':'demo'}] if self.registered else []} if path.endswith('?start=false'): self.registered=True;self.state='stopped';self.processed=0;return {'state':'stopped'} if path.endswith('/settings'): if method=='PATCH': if self.fail_patch:raise m.TestRunError('api','HTTP 409/503') assert body['expected_revision']==self.conf['revision'] for key,value in body.items(): if key=='expected_revision':continue if isinstance(value,dict):self.conf.setdefault(key,{}).update(value) else:self.conf[key]=value if self.state=='running':self.run='run2' return {'configured':deepcopy(self.conf)} if path.endswith('/start') or path.endswith('/restart'): self.state='starting' if self.hold else 'running';self.run='run2';self.processed=0 if self.hold else 1 if path.endswith('/stop'): self.state='stopping' if self.hold else 'stopped';self.pending=self.hold return dict(state=self.state,run_id=self.run,run_generation=2,metadata_path={'processed':self.processed,'queued':0},cleanup_pending=self.pending,last_error=None,cleanup_error=None) class TestRunTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup) self.root=Path(self.temp.name);self.file=self.root/'한글 공백 #.yaml' self.data=dict(version=1,camera=dict(stream_id='demo',server_id='server-demo',source_id='camera-demo',uri='rtsp://media.invalid/demo'),settings={'display_score':.25}) self.file.write_text(yaml.safe_dump(self.data)) self.preset=m.load_preset(self.file) self.api=FakeApi() self.row=dict(self.preset['camera'],store=False,active_learning=False) self.reader=lambda _:[deepcopy(self.row)] if self.api.registered else [] self.manager=m.TestRunManager(self.preset,self.api,self.reader,sleep=lambda _:None) def writes(self):return [x for x in self.api.calls if x[0]!='GET'] def test_parse_unicode_path_and_defaults(self): self.assertEqual(self.preset['wait_seconds'],90);self.assertFalse(self.preset['debug_metrics']) def test_integer_camera_canonical_source_and_human_label(self): self.data['camera']['source_id']=77 self.data['test_label']='rtsp-test-source-02' self.file.write_text(yaml.safe_dump(self.data)) preset=m.load_preset(self.file) self.assertEqual(preset['camera']['source_id'],'77') self.assertEqual(preset['camera_id'],77) self.api.registered=False manager=m.TestRunManager(preset,self.api,lambda _:[] if not self.api.registered else [dict(preset['camera'],store=False,active_learning=False)],sleep=lambda _:None) manager.execute('start') body=self.writes()[0][2] self.assertEqual(body['camera_id'],77) self.assertNotIn('test_label',body) def test_tracking_off_partial_noop_and_conflict(self): self.data['settings']={'tracking':{'enabled':False},'intrusion':{'enabled':False}} self.file.write_text(yaml.safe_dump(self.data)) preset=m.load_preset(self.file) self.assertFalse(preset['settings']['tracking']['enabled']) self.assertEqual(m.select_changes({'tracking':{'enabled':False}}, {'tracking':{'enabled':False,'activation':.25}}),{}) self.data['settings']['intrusion']['enabled']=True self.file.write_text(yaml.safe_dump(self.data)) with self.assertRaises(m.TestRunError):m.load_preset(self.file) self.preset['settings']={'tracking':{'enabled':False},'intrusion':{'enabled':False}} self.manager.execute('restart') self.assertEqual([x[0] for x in self.writes()],['PATCH']) def test_invalid_presets(self): for changes in [{'intrusion':{'roi':[[0,0],[1,1],[0,1],[1,0]]}},{'analysisFpsTarget':True},{'display_score':float('nan')},{'frame_time_anchor':{'startWallTimeMs':True}},{'fire':{'enabled':True}},{'display_categories':['not-a-class']}]: with self.subTest(changes=changes): data=deepcopy(self.data);data['settings']=changes;self.file.write_text(yaml.safe_dump(data)) with self.assertRaises(m.TestRunError):m.load_preset(self.file) def test_unsafe_yaml_duplicate_and_alias(self): for content in ['!!python/object/apply:os.system [echo bad]','version: 1\nversion: 1','x: &x [*x]']: self.file.write_text(content) with self.assertRaises(m.TestRunError):m.load_preset(self.file) def test_status_is_read_only(self): self.manager.execute('status');self.assertEqual(self.writes(),[]) def test_start_running_noop(self): self.manager.execute('start');self.assertEqual(self.writes(),[]) def test_running_change_requires_restart(self): self.preset['settings']['display_score']=.3 with self.assertRaises(m.TestRunError):self.manager.execute('start') self.assertEqual(self.writes(),[]) def test_restart_patch_does_not_double_restart(self): self.preset['settings']['intrusion']={'enabled':True} self.manager.execute('restart') self.assertEqual([x[0] for x in self.writes()],['PATCH']) self.assertEqual(self.api.conf['intrusion']['roi_name'],'existing') def test_restart_without_changes_once(self): self.manager.execute('restart');self.assertEqual([x[1] for x in self.writes()],['/api/v1/streams/demo/restart']) def test_stopped_patch_then_start(self): self.api.state='stopped';self.preset['settings']['display_score']=.3 self.manager.execute('start');self.assertEqual([x[0] for x in self.writes()],['PATCH','POST']) def test_cas_or_storage_failure_no_start(self): self.api.state='stopped';self.api.fail_patch=True;self.preset['settings']['display_score']=.3 with self.assertRaises(m.TestRunError):self.manager.execute('start') self.assertEqual([x[0] for x in self.writes()],['PATCH']);self.assertEqual(self.api.conf['display_score'],.25) def test_identity_and_uri_conflicts(self): for key in ['server_id','source_id','uri']: with self.subTest(key=key): old=self.row[key];self.row[key]='different' with self.assertRaises(m.TestRunError):self.manager.execute('start') self.row[key]=old self.assertEqual(self.writes(),[]) def test_duplicate_source_refused(self): self.manager.read_registration=lambda _:[self.row,dict(self.row,stream_id='other')] with self.assertRaises(m.TestRunError):self.manager.execute('start') self.assertEqual(self.writes(),[]) def test_transition_and_error_refused(self): for state in ['starting','stopping','error']: self.api.state=state with self.assertRaises(m.TestRunError):self.manager.execute('restart') self.assertEqual(self.writes(),[]) def test_pending_stop_not_duplicated_and_timeout(self): self.api.state='stopping';self.api.hold=True self.manager.clock=Mock(side_effect=[0,0,100]) with self.assertRaises(m.TestRunError):self.manager.execute('stop') self.assertEqual(self.writes(),[]) def test_start_timeout_not_retried(self): self.api.state='stopped';self.api.hold=True;self.manager.clock=Mock(side_effect=[0,0,100]) with self.assertRaises(m.TestRunError):self.manager.execute('start') self.assertEqual(len(self.writes()),1) def test_stop_can_complete_with_retained_prior_runtime_error(self): self.manager.api=Mock() self.manager.api.call.return_value=dict(state='stopped',cleanup_pending=False,last_error='old source error',cleanup_error=None) self.assertEqual(self.manager.wait('stopped')['state'],'stopped') def test_status_distinguishes_effective_from_configured(self): effective=deepcopy(self.api.conf);effective['display_score']=.1 self.manager.api=Mock() self.manager.api.call.return_value=dict(configured=self.api.conf,effective=effective,restart_pending=True) out=self.manager.summary(dict(state='running'),'status') self.assertEqual(out['settings']['display_score'],.1) self.assertTrue(out['restart_pending']) def test_register_defaults_no_sender_control(self): self.api.registered=False self.manager.execute('start') body=self.writes()[0][2] self.assertFalse(body['store']);self.assertFalse(body['active_learning']);self.assertFalse(body['autostart']) self.assertEqual(self.writes()[0][1],'/api/v1/streams?start=false') self.assertTrue(all(x[1].startswith('/api/v1/streams') for x in self.api.calls)) def test_lock_and_bounded_receipt(self): with m.control_lock(self.preset,self.root/'state'): with self.assertRaises(m.TestRunError): with m.control_lock(self.preset,self.root/'state'):pass m.record_result(self.preset,'start',self.manager,{'state':'running'},self.root/'state') receipt=next((self.root/'state').glob('*.last.json')) self.assertNotIn(self.preset['camera']['uri'],receipt.read_text()) self.assertEqual(receipt.stat().st_mode&0o777,0o600) if __name__=='__main__':unittest.main()