from collections import Counter, deque import json from pathlib import Path import queue import tempfile import threading from types import SimpleNamespace import unittest from unittest.mock import Mock from app.event_capture import EventCapture, PACKET from app.protected_packet_cache import ProtectedPacketCache def packet(t, key=False): pts = int(t*1e9) return PACKET.pack(pts, pts, 100000000, 4, 0, int(key), 0)+b'data' class PacketCacheTests(unittest.TestCase): def fill(self, cache, begin, end): for n in range(begin, end+1): cache.append(n*100000000, packet(n/10, n%100==0), n%100==0) def test_post_wait_retains_key_that_fifo_loses(self): old, cache = ProtectedPacketCache(), ProtectedPacketCache() for c in (old, cache): self.fill(c, 0, 95) self.assertIsNone(cache.protect('e', 9500000000)) for c in (old, cache): self.fill(c, 96, 125) self.assertNotIn(0, [p for p,_ in old.packets]) self.assertEqual(cache.rows('e',12500000000)[0][0],0) cache.release('e') self.assertEqual(cache.bytes, old.bytes) self.assertEqual(list(cache.packets), list(old.packets)) def test_overlap_shares_bytes_and_releases_independently(self): c=ProtectedPacketCache();self.fill(c,0,95) before=c.bytes self.assertIsNone(c.protect('a',9000000000)) self.assertIsNone(c.protect('b',9500000000)) self.assertEqual(c.bytes,before) self.assertIs(c.rows('a',10000000000)[0][1],c.rows('b',10000000000)[0][1]) self.fill(c,96,125);c.release('a') self.assertEqual(c.packets[0][0],0) c.release('b');self.assertGreater(c.packets[0][0],0) def test_reordered_pts_preserves_decode_order(self): c=ProtectedPacketCache() for t,key in [(0,True),(0.3,False),(0.1,False),(0.2,False)]:c.append(int(t*1e9),packet(t,key),key) self.assertIsNone(c.protect('a',200000000)) self.assertEqual([p for p,_ in c.rows('a',300000000)],[0,300000000,100000000,200000000]) def test_no_future_key_substitution(self): c=ProtectedPacketCache();c.append(10000000000,packet(10,True),True) self.assertEqual(c.protect('a',9500000000),'NO_PRE_EVENT_KEYFRAME') self.assertFalse(c.pins) def test_hard_limit_fails_pin_without_blocking_append(self): c=ProtectedPacketCache(base_bytes=120,total_bytes=600) c.append(0,packet(0,True),True);c.protect('a',0) self.fill(c,1,20) self.assertEqual(c.failures['a'],'PROTECTED_CACHE_CAPACITY') self.assertLessEqual(c.accounted,600) self.assertFalse(c.pins) c.release('a');self.assertFalse(c.failures) def test_packet_count_and_event_count_are_bounded(self): c=ProtectedPacketCache(max_events=1,max_packets=3) c.append(0,packet(0,True),True);c.protect('a',0) self.assertEqual(c.protect('b',0),'PROTECTED_CACHE_EVENT_LIMIT') self.assertEqual(c.protect('a',0),'PROTECTED_CACHE_DUPLICATE') self.fill(c,1,8);self.assertLessEqual(len(c.packets),3) self.assertIn('a',c.failures) def test_expiry_without_new_packets_and_clear(self): now=[0];c=ProtectedPacketCache(clock=lambda:now[0]) c.append(0,packet(0,True),True);c.protect('a',0) now[0]=8;c.expire() self.assertEqual(c.failures['a'],'PROTECTED_CACHE_EXPIRED') c.clear() self.assertEqual((c.bytes,c.accounted,len(c.pins),len(c.failures)),(0,0,0,0)) def test_default_budgets_and_key_diagnostics(self): c=ProtectedPacketCache();self.fill(c,0,200) self.assertEqual(c.status()['keyframe_gap_max_ns'],10000000000) self.assertEqual(c.status()['base_bytes'],32*1024**2) self.assertEqual(c.status()['total_bytes'],64*1024**2) self.assertEqual(c.protect('empty',None),'NO_COMPRESSED_VIDEO') class CaptureIntegrationTests(unittest.TestCase): def capture(self, root): c=object.__new__(EventCapture) c.cache=ProtectedPacketCache();c.packets=c.cache.packets;c.cache_status=c.cache.status() c.pending={};c.metadata=deque(maxlen=300);c.results=queue.Queue(maxsize=128) c.metrics=Counter();c.error=None;c.training=None;c.provider_key='pose';c.camera_id=3 c.stop=threading.Event();c.store=SimpleNamespace(captured=Mock(),close=Mock()) c.native=SimpleNamespace(pull=lambda:None,close=Mock(),lib=SimpleNamespace(contextra_encoded_dropped=lambda:0)) c.config=SimpleNamespace(root=Path(root));(Path(root)/'.staging').mkdir() return c def run_case(self, *, stopped=False, error=None): with tempfile.TemporaryDirectory() as root: c=self.capture(root) for n in range(96):c.cache.append(n*100000000,packet(n/10,n==0),n==0) event={'eventId':'e','ptsNs':9500000000,'runId':'run-1'} c.cache.protect('e',event['ptsNs']) c.pending['e']=(event,1e30,None) c.metadata.append({'ptsNs':9500000000}) if not stopped: for n in range(96,126):c.cache.append(n*100000000,packet(n/10,n==100),n==100) if error: c.store.captured.side_effect=RuntimeError('database unavailable') c.stop.set();c._run() self.assertFalse(c.pending) self.assertEqual(c.cache.bytes,0) self.assertFalse(c.cache.pins) c.native.close.assert_called_once();c.store.close.assert_called_once() job=json.loads((Path(root)/'.staging/e/job.json').read_text()) self.assertEqual(job['captureReason'],'STOP_PARTIAL' if stopped else 'COMPLETE') self.assertEqual(job['captureIdentity']['runId'],'run-1') if error:self.assertEqual(c.error,'EVENT_CAPTURE_RUNTIMEERROR') else:self.assertEqual(c.metrics['capture_ready'],1) def test_stop_partial_handoff(self):self.run_case(stopped=True) def test_complete_handoff(self):self.run_case() def test_failed_handoff_releases_memory_but_preserves_stage(self):self.run_case(error=True) def test_capacity_failure_is_recorded_without_media(self): with tempfile.TemporaryDirectory() as root: c=self.capture(root) c.pending['e']=({'eventId':'e','ptsNs':1},1e30,'PROTECTED_CACHE_EVENT_LIMIT') c.stop.set();c._run() c.store.captured.assert_called_once_with('e','PROTECTED_CACHE_EVENT_LIMIT') self.assertFalse(list((Path(root)/'.staging').iterdir())) def test_parallel_producers_use_existing_bounded_queue(self): with tempfile.TemporaryDirectory() as root: c=self.capture(root) r=SimpleNamespace(camera_id=3,events=(),snapshot=lambda:{'cameraId':3}) threads=[threading.Thread(target=lambda:[c.submit(r) for _ in range(100)]) for _ in range(2)] for t in threads:t.start() for t in threads:t.join() self.assertEqual(c.results.qsize(),128) self.assertEqual(c.metrics['metadata_queue_drop'],72) c._collect();self.assertEqual(c.results.qsize(),0) def test_training_uses_the_same_packet_deque(self): with tempfile.TemporaryDirectory() as root: c=self.capture(root) c.training=SimpleNamespace(collect=Mock(),consider=Mock()) c.cache.append(0,packet(0,True),True) c.stop.set();c._run() args=c.training.collect.call_args self.assertIs(args.args[0],c.packets) self.assertTrue(args.kwargs['stopped'])