"""Targeted regression fixtures for the agreed FALL review boundaries.""" import ast from dataclasses import asdict, replace import json from pathlib import Path import unittest from app.fall_config import FallConfig from app.fall_geometry import PoseFeatures from app.fall_scenario import FallEvidence, _compare, _Hold from fall_fixtures import Sequence, make_frame, make_pose class CueTests(unittest.TestCase): def evidence(self, **options): s=Sequence();s.arm();r=s.step(.7,**options) return s,r.decisions[0].evidence def test_a_b(self): _,e=self.evidence(angle=30,hip=(500,510),span=400,missing=(13,14)) self.assertEqual(e.cues,(True,True,False));self.assertTrue(e.transition) def test_a_c(self): _,e=self.evidence(angle=30,hip=(500,400),span=280,missing=(13,14)) self.assertEqual(e.cues,(True,False,True));self.assertTrue(e.transition) def test_b_c(self): _,e=self.evidence(angle=0,hip=(500,510),span=280,missing=(13,14)) self.assertEqual(e.cues,(False,True,True));self.assertTrue(e.transition) self.assertFalse(e.down) # usable vertical torso cannot be overridden by span def test_a_only(self): _,e=self.evidence(angle=30,hip=(500,400),span=400,missing=(13,14)) self.assertEqual(e.cues,(True,False,False));self.assertFalse(e.transition) def test_threshold_exact_boundaries(self): _,e=self.evidence(angle=25,hip=(500,500),span=280,missing=(13,14)) self.assertEqual(e.cues,(True,True,True)) self.assertAlmostEqual(e.hip_drop_ratio,.25) self.assertAlmostEqual(e.recent_span_ratio,.70) def test_unknown_not_lower_vote_threshold(self): for cues,expect in (((True,True,None),True),((True,False,None),False), ((True,None,None),False),((None,None,None),False)): e=FallEvidence(PoseFeatures(),cues=cues) self.assertEqual(e.transition,expect) def test_knee_soft_guard_logic(self): g=PoseFeatures(knee_guard=True) self.assertFalse(FallEvidence(g,cues=(False,True,True)).transition) self.assertFalse(FallEvidence(g,cues=(None,True,True)).transition) self.assertTrue(FallEvidence(g,cues=(True,True,False)).transition) def test_knee_guard_real_fixture(self): _,e=self.evidence(angle=0,hip=(500,510),span=280,knee_bent=True) self.assertTrue(e.geometry.knee_guard) self.assertFalse(e.transition) def test_source_change_a_b_unavailable(self): _,e=self.evidence(angle=40,hip=(500,520),missing=(6,12)) self.assertEqual(e.geometry.torso_source,'left') self.assertIsNotNone(e.geometry.torso_angle) self.assertIsNone(e.cues[0]);self.assertIsNone(e.cues[1]) def test_ankle_disappearance_not_collapse(self): s,e=self.evidence(missing=(15,16)) self.assertEqual(e.geometry.span_source,'mid_knee') self.assertIsNone(e.recent_span_ratio);self.assertFalse(e.transition) self.assertEqual(s.engine.inspect(1)['scale'],('mid_ankle',400)) def test_denominator_frozen_on_ankle_loss(self): _,e=self.evidence(hip=(500,500),missing=(15,16)) self.assertEqual(e.body_scale_source,'mid_ankle') self.assertEqual(e.body_scale_value,400) self.assertAlmostEqual(e.hip_drop_ratio,.25) def test_no_body_scale_b_unavailable(self): s=Sequence() for i in range(7):s.step(i/10,missing=(13,14,15,16)) e=s.step(.7,hip=(500,510),missing=(13,14,15,16)).decisions[0].evidence self.assertIsNone(e.cues[1]);self.assertIsNone(s.engine.inspect(1)['scale']) def test_same_fallback_source_needs_new_history(self): s=Sequence();s.arm();s.step(.7,missing=(15,16)) e=s.step(.8,missing=(15,16)).decisions[0].evidence self.assertAlmostEqual(e.recent_span_ratio,1) def test_slow_motion_not_old_reference_transition(self): s=Sequence();s.arm() for i in range(1,61): r=s.step(.6+i*.1,angle=i*.5,hip=(500,400+i*3),span=400-i*3,missing=(13,14)) self.assertFalse(r.event_candidates);self.assertFalse(r.decisions[0].evidence.transition) def test_gap_breaks_old_cues(self): s=Sequence();s.arm();e=s.down(1.2).decisions[0].evidence self.assertEqual(e.cues,(None,None,None)) def test_recent_baseline_not_minimum_or_maximum_search(self): value,cue=_compare(50,((0,40,None),(1,0,None)),25) self.assertEqual(value,10);self.assertFalse(cue) def test_down_fallback_only_matching_stable_source(self): s=Sequence();s.arm();s.step(.7,angle=40,hip=(500,510),span=200,missing=(13,14)) r=s.step(.8,angle=80,hip=(500,520),span=200,missing=(11,12)) e=r.decisions[0].evidence self.assertIsNone(e.geometry.torso_angle);self.assertEqual(e.down_source,'span:mid_ankle') self.assertTrue(e.down);self.assertEqual(r.decisions[0].state,'DOWN_CANDIDATE') def test_down_fallback_missing_reference_unavailable(self): s=Sequence() for i in range(7):s.step(i/10,missing=(13,14,15,16)) r=s.step(.7,angle=80,span=100,missing=(11,12)) self.assertIsNone(r.decisions[0].evidence.upright_span_ratio) self.assertIsNone(r.decisions[0].evidence.down) class TransitionBoundaryTests(unittest.TestCase): def transition(self): s=Sequence();s.arm() s.step(.7,angle=40,hip=(500,510),span=280,missing=(13,14)) self.assertEqual(s.engine.inspect(1)['state'],'TRANSITION') return s def test_down_at_deadline_allowed(self): s=self.transition() for i in range(8,15):s.step(i/10,angle=40,hip=(500,510),span=280,missing=(13,14)) self.assertEqual(s.down(1.5).decisions[0].state,'DOWN_CANDIDATE') def test_down_after_deadline_rejected(self): s=self.transition() for i in range(8,16):s.step(i/10,angle=40,hip=(500,510),span=280,missing=(13,14)) r=s.down(1.51) self.assertEqual(r.decisions[0].state,'UNKNOWN');self.assertFalse(r.event_candidates) self.assertEqual(s.engine.status()['transition_expired'],1) def test_transition_deadline_not_extended_by_repeated_votes(self): s=self.transition() for i in range(8,15): s.step(i/10,angle=40,hip=(500,510),span=280,missing=(13,14)) self.assertEqual(s.engine.inspect(1)['transition_pts'],700_000_000) def test_timeout_returns_stable_upright(self): s=self.transition() for i in range(8,17):s.step(i/10) self.assertEqual(s.engine.inspect(1)['state'],'UPRIGHT') self.assertEqual(s.engine.status()['transition_expired'],1) def test_transition_long_feature_gap_reset(self): s=self.transition() for t in (.8,.9,1.,1.1,1.2):s.step(t,missing=(5,6)) self.assertEqual(s.engine.inspect(1)['state'],'TRANSITION') s.step(1.201,missing=(5,6)) self.assertEqual(s.engine.inspect(1)['state'],'UNKNOWN') def test_hold_source_change_does_not_add_bridge(self): h=_Hold();h.step(True,0,500_000_000,source='torso:mid') h.step(True,100_000_000,500_000_000,source='torso:mid') h.step(True,200_000_000,500_000_000,source='torso:left') self.assertEqual(h.total,100_000_000) def test_down_entered_before_deadline_can_finish_after_it(self): s=self.transition() for i in range(8,14):s.step(i/10,angle=40,hip=(500,510),span=280,missing=(13,14)) events=[] for i in range(14,22):events.extend(s.down(i/10).event_candidates) self.assertEqual(len(events),1) self.assertEqual(events[0].frame.pts_ns,2_100_000_000) class IsolationAndFixtureTests(unittest.TestCase): def test_two_people_independent(self): s=Sequence();e=s.engine;events=[] for i in range(15): f=make_frame(i,i/10) a=make_pose(f) if i<7 else make_pose(f,angle=80,hip=(500,520),span=200) b=make_pose(f,hip=(700,400)) r=e.evaluate(f,(a,b),(1,2));events.extend(r.event_candidates) self.assertEqual([c.track_id for c in events],[1]) self.assertEqual(e.inspect(2)['state'],'UPRIGHT') def test_empty_frames_are_not_bboxes(self): s=Sequence();s.confirm();r=s.step(1.5,present=False) self.assertFalse(r.decisions);self.assertFalse(r.event_candidates) self.assertTrue(s.engine.inspect(1)['latched']) def test_low_person_score_freezes_not_recovery(self): s=Sequence();s.confirm() for i in range(15,40):s.step(i/10,score=.1) self.assertTrue(s.engine.inspect(1)['latched']) self.assertEqual(s.engine.inspect(1)['state'],'FALL_CONFIRMED') def test_repeat_sequence_deterministic(self): a,b=Sequence(),Sequence() self.assertEqual(asdict(a.confirm()),asdict(b.confirm())) def test_no_wall_clock_import_or_sleep(self): root=Path(__file__).resolve().parents[1] forbidden={'time','datetime','threading','asyncio','socket','requests','httpx','numpy','cv2','supervision'} for name in ('fall_config','fall_geometry','fall_scenario'): tree=ast.parse((root/'app'/f'{name}.py').read_text()) for node in ast.walk(tree): if isinstance(node,ast.Import):self.assertFalse({a.name.split('.')[0] for a in node.names}&forbidden) if isinstance(node,ast.ImportFrom):self.assertNotIn((node.module or '').split('.')[0],forbidden) if isinstance(node,ast.Call) and isinstance(node.func,ast.Name):self.assertNotIn(node.func.id,{'open','eval','exec'}) def test_status_does_not_advance_or_expire(self): s=Sequence();s.confirm();before=s.engine.inspect(1) for _ in range(50):s.engine.status() self.assertEqual(before,s.engine.inspect(1)) def test_all_diagnostics_json_finite(self): s=Sequence();s.confirm() json.dumps(s.engine.status(),allow_nan=False) json.dumps(s.engine.inspect(1),allow_nan=False) def test_standing_walking_no_events(self): s=Sequence() for i in range(100): r=s.step(i/10,hip=(500+i,400+(i%3)),angle=i%4) self.assertFalse(r.event_candidates) def test_brief_bending_then_recovery_no_event(self): s=Sequence();s.arm();events=[] for t,angle in ((.7,40),(.8,70),(.9,20),(1.,0),(1.1,0)): events.extend(s.step(t,angle=angle,hip=(500,410),span=300).event_candidates) self.assertFalse(events) def test_narrow_bbox_no_shortcut(self): s=Sequence();s.arm();events=[] for i in range(7,15):events.extend(s.down(i/10,box=(.4,.1,.5,.95)).event_candidates) self.assertEqual(len(events),1) self.assertEqual(events[0].bbox_xyxy,(.4,.1,.5,.95)) def test_public_serializer_not_defined(self): s=Sequence();c=s.confirm() self.assertFalse(hasattr(c,'payload')) self.assertFalse(hasattr(c,'event_id')) self.assertFalse(hasattr(c,'clipUri')) def test_state_capacity_frees_only_by_expiry(self): s=Sequence(max_states=1);s.arm() s.step(5.6,present=False) r=s.step(5.7,track_id=2) self.assertEqual(r.decisions[0].state,'UNKNOWN') def test_same_integer_and_string_id_not_reassociated(self): s=Sequence();s.confirm();r=s.down(1.5,track_id='1') self.assertEqual(r.decisions[0].state,'UNKNOWN');self.assertTrue(s.engine.inspect(1)['latched']) if __name__=='__main__':unittest.main()