"""FALL R2 contracts only: no network, live streams, models or Event delivery.""" import json,itertools,tempfile,unittest from copy import deepcopy from dataclasses import replace from unittest.mock import patch from app.fall_runtime import FallSettings,FallScenario from app.fall_r2.config import R2ConfigError from app.event_cooldown import EventCooldown from app.secure_control import Start,ControlService,command_error from app.stream_settings import revision,patch_settings from app.contracts.frames import PoseObservation from app.pose_config import POSE_MODEL from app.region_scenario import alerted from fall_fixtures import make_frame from test_loitering_settings import FakeManager,spec,request def saved():return dict(spec(),fall=FallSettings.parse({'enabled':True,'engine':'r2'}).settings_payload()) def start(raw,old=None): manager=FakeManager(saved() if old is None else old);body=request('FALL') next(s for s in body['scenarios'] if s['scenarioCode']=='FALL')['parameters']=raw return manager,ControlService(manager,'server',spec()).start_camera(Start(json.dumps(body))) def pose(f,score=.9,joint=.9): k=[(0.,0.,0.)]*17 for a,h,n,e,dy in ((5,11,13,15,-.02),(6,12,14,16,.02)): for i,x,y in ((a,.3,.50+dy),(h,.5,.52+dy),(n,.65,.525+dy),(e,.8,.53+dy)):k[i]=(x,y,joint) return PoseObservation(f,POSE_MODEL,score,(.1,.1,.9,.9),tuple(k)) class WebTests(unittest.TestCase): def test_floor_and_noop(self): _,a=start({'person_confidence':.15,'pose_confidence':.05,'down_angle_deg':90,'recovery_hold_seconds':999}) _,b=start({'person_confidence':.25,'pose_confidence':.1},a) self.assertEqual(a,b);self.assertEqual(revision(a),revision(b)) self.assertEqual((a['fall']['person_score'],a['fall']['joint_score']),(.25,.1)) def test_high_values_omission_and_reset(self): _,a=start({'fall_r2':{'schema_version':1},'person_confidence':.35,'pose_confidence':.3,'cooldown_seconds':12}) _,b=start({},a);self.assertEqual(a,b) _,c=start({'fall_r2':{'schema_version':1,'reset_to_defaults':True}},a) self.assertEqual((c['fall']['person_score'],c['fall']['joint_score'],c['event_cooldowns']['FALL']),(.25,.1,10.)) self.assertEqual(a['fall']['roi'],c['fall']['roi']) def test_old_container_ignored(self): _,a=start({}) for raw in (None,[], 'old'): self.assertEqual(start(raw)[1],a) def test_invalid_new_preserves_execution(self): for raw in (None,{}, {'schema_version':True},{'schema_version':2},{'schema_version':1,'down_angle_deg':20},{'schema_version':1,'reset_to_defaults':False}): m=FakeManager(saved());before=deepcopy(m.group.spec);body=request('FALL') next(s for s in body['scenarios'] if s['scenarioCode']=='FALL')['parameters']={'fall_r2':raw} with self.assertRaises(ValueError):ControlService(m,'server',spec()).start_camera(Start(json.dumps(body))) self.assertEqual(m.group.spec,before);self.assertFalse(m.applied) def test_invalid_old_ignored_new_rejected(self): for v in (None,'0.1',True,-1,2,10**400): self.assertEqual(start({'person_confidence':v})[1]['fall']['person_score'],.25) with self.assertRaises(ValueError):start({'fall_r2':{'schema_version':1},'person_confidence':v}) def test_reset_conflict(self): with self.assertRaises(ValueError):start({'fall_r2':{'schema_version':1,'reset_to_defaults':True},'pose_confidence':.2}) def test_legacy_rollback_and_not_enabled(self): self.assertNotIn('engine',FallSettings.parse({}).settings_payload()) x=patch_settings(saved(),{'fall':{'engine':'legacy'}});self.assertNotIn('r2',x['fall']) with self.assertRaisesRegex(ValueError,'NOT_ENABLED'):start({'fall_r2':{'schema_version':1}},spec()) def test_safe_error(self): self.assertEqual(command_error(R2ConfigError('FALL_R2_RESET_INVALID')),'FALL_R2_RESET_INVALID') self.assertEqual(command_error(R2ConfigError('private arbitrary text')),'RUNTIME_COMMAND_REJECTED') def test_persistence(self): from app.stream_manager import StreamManager from app.stream_config_store import ConfigPersistenceError with tempfile.TemporaryDirectory() as root: m=StreamManager(root+'/base',root+'/overlay');g=m.add(dict(spec(),uri='rtsp://fixture.invalid/video')) rev=m.settings('stream')['configured']['revision'];change={'fall':{'enabled':True,'engine':'r2','person_score':.15,'joint_score':.05}} with patch.object(g,'start',side_effect=AssertionError('no Start')):out=m.patch_settings('stream',rev,change) self.assertEqual(out['effective']['fall']['engine'],'r2') self.assertEqual(m.config_store.read()['added'][0]['fall']['joint_score'],.1) rev=out['configured']['revision'] with patch.object(m.config_store,'update',side_effect=AssertionError('no-op')):m.patch_settings('stream',rev,change) with patch.object(m.config_store,'update',side_effect=ConfigPersistenceError()): with self.assertRaises(ConfigPersistenceError):m.patch_settings('stream',rev,{'fall':{'joint_score':.2}}) self.assertEqual(m.settings('stream'),out) class RuntimeTests(unittest.TestCase): def scenario(self,**cfg): ids=itertools.count() return FallScenario(FallSettings.parse(dict(enabled=True,engine='r2',**cfg)),event_id_factory=lambda:'event-'+str(next(ids))) def test_current_hold_rearm_cooldown_null_id(self): e=self.scenario();clock=[0.];gate=EventCooldown(10,clock=lambda:clock[0]);e.engine.event_admission=gate.allow def step(t,joint=.9,present=True): clock[0]=t;f=make_frame(round(t*1000),t) return e.evaluate(f,(pose(f,joint=joint),) if present else (), (None,) if present else ()) rows,events=step(0);self.assertTrue(alerted(rows[0]));self.assertEqual(len(events),1);self.assertIsNone(events[0].track_id) rows,events=step(.5,.01);self.assertTrue(alerted(rows[0]));self.assertFalse(events) rows,events=step(1.1,.01);self.assertFalse(alerted(rows[0])) step(2,.01);self.assertFalse(e.engine.notification.latched) self.assertFalse(step(3)[1]);self.assertFalse(e.engine.notification.latched) self.assertEqual(len(step(10)[1]),1);self.assertFalse(step(19.9)[1]);self.assertEqual(len(step(20)[1]),1) self.assertFalse(step(31,present=False)[1]);self.assertFalse(e.engine.notification.latched) def test_multiple_people_one_event_and_roi(self): e=self.scenario();f=make_frame(0,0);rows,events=e.evaluate(f,(pose(f),pose(f)),(1,2)) self.assertEqual(len(events),1);self.assertTrue(all(alerted(x) for x in rows)) e=self.scenario(roi=[[0.,0.],[.05,0.],[.05,.05],[0.,.05]]) self.assertFalse(e.evaluate(f,(pose(f),),(None,))[1]) def test_missing_and_high_scores(self): e=self.scenario(joint_score=.3);f=make_frame(0,0) self.assertFalse(e.evaluate(f,(pose(f,joint=.1),),(1,))[1]) e=self.scenario();self.assertTrue(e.evaluate(f,(pose(f,joint=.1),),(1,))[1]) def test_boundary_and_same_frame(self): e=self.scenario();f=make_frame(0,0) with self.assertRaises(ValueError):e.evaluate(f,(pose(make_frame(1,.1)),),(1,)) e.evaluate(f,(pose(f),),(1,));self.assertFalse(e.evaluate(f,(pose(f),),(1,))[1]) f=make_frame(1,.1,run_id='new') with self.assertRaises(ValueError):e.evaluate(f,(pose(f),),(1,)) e.clear();self.assertEqual(len(e.evaluate(f,(pose(f),),(None,))[1]),1) def test_no_old_bbox_on_missing(self): e=self.scenario();f=make_frame(0,0);e.evaluate(f,(pose(f),),(1,)) rows,events=e.evaluate(make_frame(1,.1),(),(),gap_before=True) self.assertEqual(rows,());self.assertFalse(events) def test_actual_redis_and_event_contract(self): from app.event_artifact_contract import build_body from app.contracts.frames import DetectionObservation from app.contracts.visualization import FrameResult,VisualizationCommand e=self.scenario();f=make_frame(0,0);p=pose(f);rows,events=e.evaluate(f,(p,),(None,)) cmd=VisualizationCommand(DetectionObservation(f,p.model_key,0,p.score,p.bbox_xyxy),'person',color='#FF3030',track_id=None,scenario=rows[0],pose=p) snap=FrameResult(f,(cmd,),scenario_config=e.config).snapshot() self.assertEqual(snap['objects'][0]['color'],'#FF3030');self.assertIsNone(snap['objects'][0]['trackId']) payload=events[0].payload();payload['frameTime']['detectedAtWallMs']=1000 item=build_body(payload,{'clip':True,'thumbnail':True},None,'https://fixture.invalid','fixture')['events'][0] self.assertEqual(item['eventType'],'FALL');self.assertTrue(item['clipUrl']);self.assertTrue(item['thumbnailUrl']) if __name__=='__main__':unittest.main()