import copy from collections import Counter, deque from dataclasses import asdict, replace import json import queue from types import SimpleNamespace import unittest from unittest.mock import Mock, patch from app.contracts.frames import FrameEnvelope from app.contracts.visualization import frame_snapshot from app.event_artifact_contract import build_body from app.event_capture import EventCapture from app.event_time_fallback import event_diagnostic, fallback_time from app.frame_time import FrameAnchor, RtpFrameClock, validate_carried_time from app.intrusion import IntrusionEvent class EventTimeFallbackTests(unittest.TestCase): def packet(self, raw=100, rate=90000, ssrc=1): return dict(timestamp=raw, clock_rate=rate, ssrc=ssrc) def clock(self, anchor=None): a = anchor or FrameAnchor(startWallTimeMs=100000).frame_time() return RtpFrameClock(a, clock_ns=lambda: 200000000000) def frame(self, time, run='run-1', generation=1): return FrameEnvelope('server', '3', 'stream', run, generation, 2, 200000000, 200100, 1920, 1080, frame_time=time) def event(self, time): f = self.frame(time) e = IntrusionEvent('event-1', f, 1, 'roi-1', '0', .9, (.1,.1,.3,.3), ai_recorded_at_wall_ms=200200) p = e.payload() p['cameraId'] = 3 p['eventTimeFallback'] = event_diagnostic(f, 'pose') return e, p def invalid_time(self): c = self.clock() c.resolve(self.packet()) return c.resolve(self.packet(99)) def test_normal_output_unchanged(self): t = self.clock().resolve(self.packet(90001)) self.assertEqual(t.source_time_wall_ms, 101000) e, payload = self.event(t) self.assertIsNone(payload.pop('eventTimeFallback')) result = build_body(payload, {}, None, 'http://unused.invalid', 'unused')['events'][0] self.assertEqual(result['detectedAtWallMs'], 101000) self.assertNotIn('eventTime', result['detail']) def test_three_causes_stick_until_new_clock_and_frame_transport(self): for packet, reason in ((self.packet(99), 'RTP_TIMESTAMP_BACKWARD'), (self.packet(100, ssrc=2), 'SSRC_CHANGED'), (self.packet(100, rate=8000), 'RTP_CLOCK_RATE_CHANGED')): with self.subTest(reason=reason): c = self.clock() c.resolve(self.packet()) first = c.resolve(packet) self.assertEqual(first.invalidation_reason, reason) for p in (self.packet(200), None, self.packet(300)): t = c.resolve(p) self.assertEqual(t.invalidation_reason, reason) self.assertEqual(t.invalidated_at_wall_ms, 200000) self.assertIsNone(t.source_time_wall_ms) restored = validate_carried_time(json.loads(json.dumps(asdict(t))), c.anchor) self.assertEqual(restored, t) self.assertEqual(fallback_time(self.event(t)[1]), 200200) self.assertEqual(c.invalidated_frames, 4) fresh = self.clock().resolve(self.packet()) self.assertIsNone(fresh.invalidation_reason) self.assertEqual(fresh.sync_status, 'ESTIMATED') def test_private_diagnostic_does_not_change_redis_snapshot(self): t = self.invalid_time() original = replace(t, invalidation_reason=None, invalidated_at_wall_ms=None, invalidated_frames=0) self.assertEqual(frame_snapshot(self.frame(t), 3), frame_snapshot(self.frame(original), 3)) def test_multi_roi_event_preserves_time_and_rule_identity(self): from app.multi_roi.adapter import normalize from app.multi_roi.policy import DEFAULT from app.multi_roi.runtime import RuleRuntime from multi_roi_fixtures import request, frame, observation doc = request('INTRUSION', 'INTRUSION') for rule in doc['rules']: rule['parameters'] = dict(dwell_seconds=.1, cooldown_seconds=10) runtime = RuleRuntime(normalize(doc, DEFAULT).payload(), clock=lambda: 10.) try: for i in range(4): f = replace(frame(i), frame_time=self.invalid_time()) result = runtime.person(f, (observation(f),), (None,), [1], (0,)) if result.events: break self.assertEqual({e.rule_id for e in result.events}, {0, 1}) snapshot = result.snapshot() for event in result.events: payload = event.payload() payload['eventTimeFallback'] = event_diagnostic(event.frame, 'medium') wire = build_body(payload, {}, None, 'https://unused.invalid', 'unused')['events'][0] self.assertEqual(wire['detectedAtWallMs'], event.ai_recorded_at_wall_ms) self.assertEqual(wire['detail']['ruleId'], event.rule_id) self.assertEqual(wire['detail']['roiId'], payload['roiId']) self.assertEqual(wire['detail']['eventTime']['runId'], f.run_id) self.assertIsNone(wire['detail']['frameTime']['detectedAtWallMs']) self.assertEqual(snapshot, result.snapshot()) finally: runtime.close() def test_event_wire_fallback_and_frozen_retry(self): _, payload = self.event(self.invalid_time()) original = copy.deepcopy(payload) first = build_body(payload, {'clip':True}, None, 'http://unused.invalid', 'unused') with patch('time.time', return_value=999999999): self.assertEqual(build_body(payload, {'clip':True}, None, 'http://unused.invalid', 'unused'), first) row = first['events'][0] self.assertEqual(row['detectedAtWallMs'], 200200) self.assertEqual(row['detail']['eventTime']['source'], 'AI_WALL_FALLBACK') self.assertEqual(row['detail']['frameTime']['syncStatus'], 'RAW_ONLY') self.assertIsNone(row['detail']['frameTime']['detectedAtWallMs']) self.assertEqual(payload, original) payload.pop('eventTimeFallback') self.assertIsNone(build_body(payload, {}, None, 'http://unused.invalid', 'unused')) def test_other_invalid_inputs_and_run_identity_still_block(self): _, base = self.event(self.invalid_time()) for value in (None, 0, -1, True, '200200', float('nan'), 2**53): payload = copy.deepcopy(base);payload['aiRecordedAtWallMs'] = value self.assertIsNone(fallback_time(payload)) for key, value in (('startWallTimeMs', None), ('startWallTimeMs', 0), ('timeSource', 'SYNTHETIC_TEST_ANCHOR'), ('syncStatus', 'ESTIMATED')): payload = copy.deepcopy(base);payload['frameTime'][key] = value self.assertIsNone(fallback_time(payload)) for key, value in (('runId', 'old-run'), ('runGeneration', 2), ('cameraId', '4'), ('reason', 'UNKNOWN'), ('invalidatedFrames', 0)): payload = copy.deepcopy(base);payload['eventTimeFallback'][key] = value self.assertIsNone(fallback_time(payload)) c = RtpFrameClock(None);c.resolve(self.packet()) self.assertIsNone(event_diagnostic(self.frame(c.resolve(self.packet(99))), 'pose')) for bad in (self.packet(True), self.packet(1, rate=0), {}): with self.assertRaises(ValueError): self.clock().resolve(bad) def test_corrupt_private_identity_rejected(self): c = self.clock();c.resolve(self.packet());t=c.resolve(self.packet(99)) for changes in (dict(invalidation_reason='UNKNOWN'), dict(invalidated_frames=True), dict(invalidated_at_wall_ms=0), dict(source_time_wall_ms=100001)): with self.assertRaises(ValueError): validate_carried_time(dict(asdict(t), **changes), c.anchor) def test_capture_attaches_once_per_event_without_external_io(self): e, payload = self.event(self.invalid_time()) capture = object.__new__(EventCapture) capture.native = SimpleNamespace(pull=lambda:None) capture.results = queue.Queue();capture.metadata = deque(maxlen=300) capture.camera_id = 3;capture.provider_key = 'pose';capture.training = None capture.metrics = Counter();capture.pending = {} capture.store = SimpleNamespace(accept=Mock(return_value=(True,None))) snap = dict(frame_snapshot(e.frame, 3), scenarioConfig={}) for _ in range(2): capture.results.put(SimpleNamespace(camera_id=3, events=(e,), snapshot=lambda:snap)) with patch('builtins.print') as log: capture._collect() self.assertEqual(log.call_count, 1) self.assertEqual(capture.metrics['event_time_fallback_candidates'], 2) admitted = capture.store.accept.call_args.args[0] self.assertEqual(fallback_time(admitted), 200200) self.assertNotIn('eventTimeFallback', capture.metadata[0])