"""No network, models, real StreamGroup, Redis or Event delivery.""" from copy import deepcopy from dataclasses import replace import json import tempfile import unittest from unittest.mock import patch from app.fight_runtime import FightSettings, FightScenario from app.fight_r5.config import R5Config, R5ConfigError from app.fight_r5.core import R5Core from app.event_cooldown import EventCooldown from app.stream_settings import patch_settings, settings_view, revision from app.secure_control import ControlService, Start from app.contracts.frames import DetectionObservation from app.contracts.visualization import FrameResult, VisualizationCommand from app.region_scenario import alerted from test_fight import frame, person from test_loitering_settings import FakeManager, spec, request def saved(): return dict(spec(),fight=FightSettings.parse({'enabled':True,'engine':'r5'}).settings_payload()) def start(parameters, original=None): old=saved() if original is None else original manager=FakeManager(old);body=request('FIGHT') next(s for s in body['scenarios'] if s['scenarioCode']=='FIGHT')['parameters']=parameters result=ControlService(manager,'server',spec()).start_camera(Start(json.dumps(body))) return manager,result class ConfigTests(unittest.TestCase): def test_legacy_default_and_explicit_rollback(self): self.assertNotIn('engine',FightSettings.parse({}).settings_payload()) out=patch_settings(saved(),{'fight':{'engine':'legacy'}}) self.assertNotIn('r5',out['fight']);self.assertNotIn('engine',out['fight']) def test_floor_same_effective_start(self): _,a=start({'person_confidence':.15,'pose_confidence':.05,'arm_motion_ratio':-999,'required_bursts':999}) _,b=start({'person_confidence':.25,'pose_confidence':.10},a) self.assertEqual(a,b);self.assertEqual(revision(a),revision(b)) self.assertEqual((a['fight']['person_score'],a['fight']['joint_score']),(.25,.1)) def test_high_scores_preserved(self): _,out=start({'person_confidence':.4,'pose_confidence':.2}) self.assertEqual((out['fight']['person_score'],out['fight']['joint_score']),(.4,.2)) def test_old_optional_container_is_fail_soft(self): _,expected=start({}) for value in (None,[], 'legacy-invalid-container'): _,out=start(value) self.assertEqual(out,expected) def test_new_tuning_persists_when_old_web_omits_it(self): _,out=start({'fight_r5':{'schema_version':1,'min_samples':3},'cooldown_seconds':8}) _,again=start({},out) self.assertEqual(out,again) self.assertEqual(settings_view(again)['fight']['r5']['min_samples'],3) _,reset=start({'fight_r5':{'schema_version':1,'reset_to_defaults':True}},again) self.assertEqual(reset['fight']['r5'],R5Config().payload()) def test_new_block_requires_ai_activation(self): with self.assertRaisesRegex(R5ConfigError,'NOT_ENABLED'): start({'fight_r5':{'schema_version':1}},spec()) def test_invalid_explicit_request_preserves_saved_and_run(self): invalid=[None,{}, {'schema_version':2},{'schema_version':True}, {'schema_version':1,'surprise':1},{'schema_version':1,'window_seconds':4}, {'schema_version':1,'min_samples':True},{'schema_version':1,'min_samples':16}, {'schema_version':1,'reset_to_defaults':False}] for value in invalid: manager=FakeManager(saved());before=deepcopy(manager.group.spec);body=request('FIGHT') next(s for s in body['scenarios'] if s['scenarioCode']=='FIGHT')['parameters']={'fight_r5':value} with self.subTest(value=value),self.assertRaises(ValueError): ControlService(manager,'server',spec()).start_camera(Start(json.dumps(body))) self.assertFalse(manager.applied);self.assertEqual(manager.group.spec,before) def test_invalid_common_values_old_ignore_new_reject(self): for value in (None,'0.1',True,-1,2,10**400): _,out=start({'person_confidence':value}) self.assertEqual(out['fight']['person_score'],.25) with self.subTest(value=str(value)[:20]),self.assertRaises(ValueError): start({'person_confidence':value,'fight_r5':{'schema_version':1}}) def test_nan_never_reaches_storage(self): with self.assertRaises(ValueError):start({'person_confidence':float('nan')}) def test_async_operation_preserves_safe_error_code(self): from app.secure_control import command_error self.assertEqual(command_error(R5ConfigError('FIGHT_R5_SCHEMA_VERSION_REQUIRED')),'FIGHT_R5_SCHEMA_VERSION_REQUIRED') self.assertEqual(command_error(R5ConfigError('unsafe arbitrary text')),'RUNTIME_COMMAND_REJECTED') def test_new_block_sample_constraints(self): for value in (0,-1,17,2.5): with self.assertRaises(ValueError):R5Config(min_samples=value) with self.assertRaises(ValueError):R5Config(window_seconds=.3,contact_min_samples=5) class RuntimeTests(unittest.TestCase): def scenario(self): return FightScenario(FightSettings.parse({'enabled':True,'engine':'r5'}),event_id_factory=lambda:'r5-event',wall_ns=lambda:1000000000) def test_one_person_stationary_and_adjacent_no_hit(self): for two in (False,True): e=self.scenario() for i in range(30): f=frame(i/10);poses=(person(f,450),person(f,650)) if two else (person(f,450),) rows,events=e.evaluate(f,poses,(1,2) if two else (1,)) self.assertFalse(events);self.assertFalse(any(alerted(r) for r in rows)) def test_same_frame_validation_and_duplicates(self): e=self.scenario();f=frame(0) with self.assertRaises(ValueError):e.evaluate(f,(person(frame(.1),450),),(1,)) with self.assertRaises(ValueError):e.evaluate(f,(person(f,450),person(f,650)),(1,1)) e.evaluate(f,(person(f,450),),(1,));self.assertFalse(e.evaluate(f,(person(f,450),),(1,))[1]) def test_gap_and_run_boundary_clear_motion(self): e=self.scenario() for i in range(3): f=frame(i/10);e.evaluate(f,(person(f,450),person(f,650)),(1,2)) f=frame(.3);e.evaluate(f,(person(f,450),person(f,650)),(1,2),gap_before=True) self.assertFalse(e.engine.last_result['observable']) f=replace(frame(.4),run_id='new') with self.assertRaises(ValueError):e.evaluate(f,(person(f,450),person(f,650)),(1,2)) e.clear();e.evaluate(f,(person(f,450),person(f,650)),(1,2)) self.assertFalse(e.engine.last_result['observable']) def test_red_hold_no_new_event_and_null_track_is_not_invented(self): e=self.scenario();f=frame(0);poses=(person(f,450),person(f,650)) current=dict(hit=True,quiet=False,current_orders=[0,1],display_orders=[0,1],episode=dict(event=True,latched=True)) hold=dict(hit=False,quiet=False,current_orders=[],display_orders=[0,1],episode=dict(event=False,latched=True)) with patch.object(e.engine.core,'step',side_effect=[(current,{}),(hold,{})]): rows,events=e.evaluate(f,poses,(None,2)) self.assertIsNone(events[0].track_id) self.assertEqual(events[0].object_ref,'0') f=frame(.1);poses=(person(f,450),person(f,650));rows,events=e.evaluate(f,poses,(None,2)) self.assertFalse(events);self.assertTrue(all(alerted(r) for r in rows)) commands=tuple(VisualizationCommand(DetectionObservation(f,p.model_key,0,p.score,p.bbox_xyxy), 'person',color='#FF3030',track_id=tid,scenario=r,pose=p) for p,tid,r in zip(poses,(None,2),rows)) payload=FrameResult(f,commands,scenario_config=e.config).snapshot() self.assertEqual([x['color'] for x in payload['objects']],['#FF3030','#FF3030']) self.assertIsNone(payload['objects'][0]['trackId']);self.assertEqual(payload['events'],[]) self.assertTrue(all(r[0].state=='FIGHT' for r in rows)) def test_external_cooldown_only_gates_event(self): e=self.scenario();gate=EventCooldown(10,clock=lambda:0) self.assertTrue(gate.allow('FIGHT'));e.engine.event_admission=gate.allow f=frame(0);value=dict(hit=True,quiet=False,current_orders=[0,1],display_orders=[0,1],episode=dict(event=True,latched=True)) with patch.object(e.engine.core,'step',return_value=(value,{})): rows,events=e.evaluate(f,(person(f,450),person(f,650)),(1,2)) self.assertFalse(events);self.assertTrue(all(alerted(r) for r in rows));self.assertTrue(all(r[0].event_id is None for r in rows)) def test_existing_event_body_and_media_fields(self): from app.event_artifact_contract import build_body e=self.scenario();f=frame(0) value=dict(hit=True,quiet=False,current_orders=[0,1],display_orders=[0,1],episode=dict(event=True,latched=True)) with patch.object(e.engine.core,'step',return_value=(value,{})): _,events=e.evaluate(f,(person(f,450),person(f,650)),(None,2)) event=events[0].payload();event['frameTime']['detectedAtWallMs']=1000 body=build_body(event,{'clip':True,'thumbnail':True},None,'https://fixture.invalid','fixture') item=body['events'][0] self.assertEqual(item['eventType'],'FIGHT');self.assertIsNone(item['detail']['trackId']) self.assertTrue(item['clipUrl'].endswith('/clip.mp4'));self.assertTrue(item['thumbnailUrl'].endswith('/thumbnail.jpg')) class PersistenceTests(unittest.TestCase): def test_r5_settings_roundtrip_cas_noop_and_failed_save(self): from app.stream_manager import StreamManager from app.stream_config_store import ConfigPersistenceError with tempfile.TemporaryDirectory() as root: manager=StreamManager(root+'/base',root+'/overlay') group=manager.add(dict(spec(),uri='rtsp://fixture.invalid/video')) initial=manager.settings('stream')['configured']['revision'] changes={'fight':{'engine':'r5','enabled':True,'person_score':.1,'joint_score':.05}} with patch.object(group,'start',side_effect=AssertionError('no real start')): result=manager.patch_settings('stream',initial,changes) self.assertEqual(result['effective']['fight']['engine'],'r5') self.assertEqual(result['effective']['fight']['person_score'],.25) stored=manager.config_store.read()['added'][0] self.assertEqual(FightSettings.parse(stored['fight']).r5.payload(),R5Config().payload()) rev=result['configured']['revision'] with patch.object(manager.config_store,'update',side_effect=AssertionError('no-op must not save')): manager.patch_settings('stream',rev,changes) with self.assertRaises(ValueError):manager.patch_settings('stream',initial,changes) with patch.object(manager.config_store,'update',side_effect=ConfigPersistenceError()): with self.assertRaises(ConfigPersistenceError): manager.patch_settings('stream',rev,{'fight':{'r5':{'schema_version':1,'min_samples':3}}}) self.assertEqual(manager.settings('stream'),result) if __name__=='__main__':unittest.main()