import json from pathlib import Path import tempfile import threading import time import unittest from unittest.mock import patch from app.event_artifact_contract import ArtifactConfig, build_body, media_path, relative_directory from app.event_artifact_service import ArtifactService, file_app, post_event def event(): return dict(cameraId=77, sourceId='77', eventId='event-test-1', scenario='INTRUSION', serverId='test-ai', streamId='test-stream', runId='run-1', runGeneration=1, frameSeq=8, ptsNs=800000000, trackId=3, roiId='roi-1', aiRecordedAtWallMs=1789610000123, frameTime=dict(startWallTimeMs=1789610000000, rtpTimestamp=9000, rtpClockRate=90000, detectedAtWallMs=1789610000100)) class MemoryLedger: def __init__(self): self.attempts = 0 self.body = None self.state = 'PENDING' self.lock = threading.Lock() def prepare_delivery(self, row, body): self.body = self.body or body self.state = 'PENDING' if body else 'BLOCKED_TIME' return self.body def begin_attempt(self, row): with self.lock: if self.attempts == 2 or self.state not in ('PENDING', 'RETRY'): return None self.attempts += 1 self.state = 'SENDING' return self.attempts def finish_attempt(self, row, state, status, response, error): self.state = state class ArtifactTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.config = ArtifactConfig(self.root, 'unused', 'http://web.invalid/api/v1/events', 'http://ai.invalid/artifacts') def row(self): e = event() return dict(event_id=e['eventId'], payload=e, media={'clip': True, 'thumbnail': True}, relative_path=relative_directory(e), error=None, body=None) def test_path_clock_and_fixed_path_not_detected_time(self): e = event() path = relative_directory(e) self.assertIn('.123+0900_INTRUSION_event-test-1', path) e['frameTime']['detectedAtWallMs'] = None self.assertEqual(relative_directory(e), path) for camera in ('../x', '/root', True, ''): e['cameraId'] = camera with self.assertRaises(ValueError): relative_directory(e) def test_wire_type_extras_and_time_block(self): row = self.row() body = build_body(event(), row['media'], None, self.config.public_base, row['relative_path']) self.assertEqual(set(body), {'events', 'serverId'}) wire = body['events'][0] self.assertIs(type(wire['cameraId']), int) self.assertEqual(wire['score'], .7) self.assertFalse(wire['detail']['scoreMeasured']) self.assertNotIn('rtpClockRate', wire) self.assertEqual(wire['startWallTimeMs'], event()['frameTime']['startWallTimeMs']) e = event(); e['frameTime']['detectedAtWallMs'] = None self.assertIsNone(build_body(e, {}, 'NO_VIDEO', self.config.public_base, row['relative_path'])) def test_http_only_completed_media_no_listing_or_symlinks(self): from fastapi.testclient import TestClient relative = self.row()['relative_path'] folder = self.root/relative folder.mkdir(parents=True) (folder/'clip.mp4').write_bytes(b'media') (folder/'event.json').write_text('{}') (self.root/'secret').write_bytes(b'private') (folder/'thumbnail.jpg').symlink_to(self.root/'secret') client = TestClient(file_app(self.root)) self.assertEqual(client.get('/artifacts/'+relative+'/clip.mp4').content, b'media') for path in (relative, relative+'/event.json', relative+'/thumbnail.jpg', '.staging/test/clip.mp4', '../secret', 'README.ko.md'): self.assertEqual(client.get('/artifacts/'+path).status_code, 404) self.assertIsNone(media_path(self.root, '/etc/passwd')) def test_delivery_max_two_frozen_body_and_no_retry_400(self): service = ArtifactService(self.config) service.store = MemoryLedger() with patch('app.event_artifact_service.post_event', return_value=(False, True, 503, None, 'HTTP_REJECTED')) as post: service._deliver(self.row()) service._deliver(self.row()) self.assertEqual(post.call_count, 2) self.assertEqual(post.call_args_list[0], post.call_args_list[1]) self.assertEqual(service.store.attempts, 2) service.store = MemoryLedger() with patch('app.event_artifact_service.post_event', return_value=(False, False, 400, None, 'HTTP_REJECTED')) as post: service._deliver(self.row()) self.assertEqual(post.call_count, 1) service.store = MemoryLedger() row = self.row(); row['payload']['frameTime']['detectedAtWallMs'] = None with patch('app.event_artifact_service.post_event') as post: service._deliver(row) post.assert_not_called() self.assertEqual(service.store.state, 'BLOCKED_TIME') def test_artifact_failure_has_no_urls_but_event_preserved(self): row = self.row() wire = build_body(event(), {}, 'NO_VIDEO', self.config.public_base, row['relative_path'])['events'][0] self.assertNotIn('clipUrl', wire) self.assertNotIn('thumbnailUrl', wire) self.assertEqual(wire['eventId'], event()['eventId']) self.assertEqual(wire['detail']['artifactError'], 'NO_VIDEO') def test_http_ack_counts_and_malformed_response(self): from unittest.mock import MagicMock for payload, expected in [({'result': True, 'data': {'accepted': 1, 'duplicated': 0}}, True), ({'result': True, 'data': {'accepted': 0, 'duplicated': 1}}, True), ({'result': True, 'data': {'accepted': 0, 'duplicated': 0}}, False), ({'result': False}, False)]: response = MagicMock(status=200) response.read.return_value = json.dumps(payload).encode() response.__enter__.return_value = response with patch('urllib.request.build_opener') as opener: opener.return_value.open.return_value = response self.assertEqual(post_event('http://web.invalid/events', {})[0], expected) def test_render_metadata_absent_not_reused(self): from PIL import Image from app.artifact_media import draw_frame rgb = Image.new('RGB', (320, 180), 'white') empty = draw_frame(rgb, None) self.assertEqual(empty.size, (1280, 720)) snap = dict(objects=[dict(visible=True, bbox=[.1,.1,.2,.2],color='#FF3030', labelText='person',confidence=.8,trackId=3,eventIds=['x'])]) actual = draw_frame(rgb, snap) self.assertNotEqual(actual.getpixel((128, 100)), empty.getpixel((128,100))) def test_sparse_pts_durations_preserve_time(self): from app.artifact_media import frame_duration pts = [0,100_000_000,300_000_000,500_000_000] durations = [frame_duration(a,b,None) for a,b in zip(pts,pts[1:])] durations.append(frame_duration(pts[-1],None,600_000_000)) self.assertEqual(sum(durations),600_000_000) with self.assertRaises(ValueError): frame_duration(1,1,None) def test_http_only_never_initializes_or_recovers_ledger(self): from dataclasses import replace from unittest.mock import MagicMock service=ArtifactService(replace(self.config,enabled=False,port=0)) service.store=MagicMock() service.start() try: self.assertFalse(service.status()['enabled']) self.assertEqual(service.status()['workers'],0) service.store.initialize.assert_not_called() service.store.recover.assert_not_called() finally:self.assertTrue(service.close(3)) def test_four_render_workers_and_total_shutdown_budget(self): from unittest.mock import MagicMock service=ArtifactService(self.config) service.store=MagicMock() service.store.claim.side_effect=lambda kind:dict(event_id=threading.current_thread().name,claim_token='x',channel_key='x') entered=threading.Barrier(5) def render(row): entered.wait(timeout=3) service.stop.wait(3) service._render=render for i in range(4): name='test-render-'+str(i);service.activity[name]='idle' w=threading.Thread(target=service._run,args=('render',name));service.workers.append(w);w.start() entered.wait(timeout=3) self.assertEqual(service.status()['activity']['render'],4) start=time.monotonic() self.assertTrue(service.close(3)) self.assertLess(time.monotonic()-start,3) def test_warning_includes_reservations(self): service=ArtifactService(self.config) service.cached={'storage':{'observed_bytes':self.config.warning-1,'reserved_bytes':1}} self.assertTrue(service.status()['storage_warning']) service.cached['storage']['reserved_bytes']=0 self.assertFalse(service.status()['storage_warning']) if __name__ == '__main__': unittest.main()