"""Frozen pre-change wire fixtures + actual serialization/path/validator functions.""" from copy import deepcopy from dataclasses import replace import json import math from pathlib import Path import re import tempfile import unittest from unittest.mock import patch from PIL import Image, ImageDraw from app.contracts.visualization import FrameResult from app.event_artifact_contract import relative_directory, build_body, media_path from app.loitering import LoiteringConfig from app.stream_settings import patch_settings from loitering_contract_helpers import ROOT, result, event_payload, node_function class BoundaryError(Exception): def __init__(self,code,status):self.code,self.status=code,status class WireTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.old=json.loads((ROOT/'tests/fixtures/loitering_legacy_wire.json').read_text()) def test_intrusion_snapshot_exactly_matches_baseline(self): self.assertEqual(result('INTRUSION').snapshot(),self.old['frame_intrusion']) def test_off_snapshot_exactly_matches_baseline(self): self.assertEqual(result(None).snapshot(),self.old['frame_off']) def test_loitering_only_changes_existing_scenario_values(self): wire=result().snapshot() normalized=json.loads(json.dumps(wire).replace('LOITERING','INTRUSION')) self.assertEqual(normalized,self.old['frame_intrusion']) self.assertEqual(wire['objects'][0]['scenarioResults'][0]['scenario'],'LOITERING') self.assertEqual(wire['events'][0]['eventId'],wire['objects'][0]['eventIds'][0]) def test_event_payload_has_no_new_fields(self): value=event_payload(result()) normalized=json.loads(json.dumps(value).replace('LOITERING','INTRUSION')) self.assertEqual(normalized,self.old['event_payload']) self.assertFalse({'dwellSeconds','thresholdSeconds','loiteringDwell'} & value.keys()) def test_body_uses_original_wrapper_and_row_schema(self): event=event_payload(result());relative=relative_directory(event) body=build_body(event,{'clip':True,'thumbnail':True},None,'https://fixture.invalid/artifacts',relative) self.assertEqual(json.loads(json.dumps(body).replace('LOITERING','INTRUSION')),self.old['event_body']) self.assertEqual(set(body),{'serverId','events'}) self.assertEqual(body['events'][0]['eventType'],'LOITERING') self.assertEqual(body['events'][0]['detail']['roiLabel'],'Zone A') def test_intrusion_body_unchanged(self): event=event_payload(result('INTRUSION')) self.assertEqual(build_body(event,{'clip':True,'thumbnail':True},None, 'https://fixture.invalid/artifacts',relative_directory(event)),self.old['event_body']) def test_invalid_absolute_time_still_blocks_delivery(self): event=event_payload(result(synced=False)) self.assertIsNone(build_body(event,{},None,'https://fixture.invalid/artifacts',relative_directory(event))) self.assertEqual(event['aiRecordedAtWallMs'],1758000005123) def test_failure_does_not_fabricate_urls(self): event=event_payload(result());body=build_body(event,{},'NO_VIDEO','https://fixture.invalid/artifacts',relative_directory(event)) row=body['events'][0] self.assertNotIn('clipUrl',row);self.assertNotIn('thumbnailUrl',row) self.assertEqual(row['detail']['artifactError'],'NO_VIDEO') def test_folder_time_still_uses_ai_recorded_not_detection(self): event=event_payload(result());expected=relative_directory(event) event['frameTime']['detectedAtWallMs']+=86400000 self.assertEqual(relative_directory(event),expected) self.assertEqual(expected.replace('LOITERING','INTRUSION'),self.old['relative']) def test_unrepresentable_timer_rejected_before_persistence(self): for name in ('dwell_seconds','grace_seconds','rearm_seconds'): with self.subTest(name=name),self.assertRaises(ValueError): patch_settings({'server_id':'s','source_id':'77','stream_id':'x'}, {'loitering':{name:1e-12}}) class MediaTests(unittest.TestCase): def setUp(self): tmp=tempfile.TemporaryDirectory();self.addCleanup(tmp.cleanup);self.root=Path(tmp.name) self.event=event_payload(result());self.relative=relative_directory(self.event) directory=self.root/self.relative;directory.mkdir(parents=True) for name in ('clip.mp4','thumbnail.jpg','event.json','README.ko.md'):(directory/name).write_bytes(b'fixture') def test_media_names_allowed_without_new_namespace(self): for name in ('clip.mp4','thumbnail.jpg'): self.assertEqual(media_path(self.root,self.relative+'/'+name),self.root/self.relative/name) def test_protected_files_missing_and_traversal_remain_hidden(self): for path in (self.relative+'/event.json',self.relative+'/README.ko.md',self.relative+'/missing.mp4', '../clip.mp4','/etc/passwd','.staging/event-01/clip.mp4',self.relative+'/../clip.mp4'): with self.subTest(path=path):self.assertIsNone(media_path(self.root,path)) def test_symlink_cannot_escape(self): clip=self.root/self.relative/'clip.mp4';clip.unlink();outside=self.root/'private';outside.write_bytes(b'x') clip.symlink_to(outside) self.assertIsNone(media_path(self.root,self.relative+'/clip.mp4')) def test_unknown_scenario_is_not_a_file_whitelist_wildcard(self): for name in ('FIGHT','LOITERING_EVIL','loitering','../bad'): with self.subTest(name=name),self.assertRaises(ValueError):relative_directory(dict(self.event,scenario=name)) def test_gateway_regex_changes_only_scenario_alternation(self): source=(ROOT/'docker/secure-gateway.conf.template').read_text() pattern=re.search(r'location ~ "([^"]+)"',source).group(1) for scenario in ('INTRUSION','LOITERING'): path='/artifacts/'+self.relative.replace('LOITERING',scenario)+'/clip.mp4' self.assertIsNotNone(re.fullmatch(pattern,path)) for name in ('event.json','README.ko.md','clip.mp4/extra'): self.assertIsNone(re.fullmatch(pattern,'/artifacts/'+self.relative+'/'+name)) self.assertIn('include /tmp/media-access.conf;',source) self.assertIn('location = /ws/v1',source) self.assertIn('proxy_set_header Authorization $http_authorization;',source) def test_overlay_preserves_dimensions_and_uses_loitering_label(self): draw=node_function('app/artifact_media.py','draw_frame',{'Image':Image,'ImageDraw':ImageDraw}) image=Image.new('RGB',(640,360)) for size in ((1280,720),(640,360)): with patch.object(ImageDraw.ImageDraw,'text',autospec=True) as text: output=draw(image,result().snapshot(),size) self.assertEqual(output.size,size) self.assertIn('LOITERING',text.call_args.args[2]) self.assertNotIn('INTRUSION',text.call_args.args[2]) def test_overlay_roi_uses_selected_scenario_config(self): draw=node_function('app/artifact_media.py','draw_frame',{'Image':Image,'ImageDraw':ImageDraw}) wire=result().snapshot();wire['objects']=[] wire['scenarioConfig']['LOITERING']['roi']=[[.2,.2],[.8,.2],[.8,.8],[.2,.8]] output=draw(Image.new('RGB',(640,360)),wire,(640,360)) self.assertEqual(output.getpixel((128,72)),(255,159,50)) def test_legacy_overlay_still_labels_intrusion(self): draw=node_function('app/artifact_media.py','draw_frame',{'Image':Image,'ImageDraw':ImageDraw}) with patch.object(ImageDraw.ImageDraw,'text',autospec=True) as text: draw(Image.new('RGB',(640,360)),result('INTRUSION').snapshot()) self.assertIn('INTRUSION',text.call_args.args[2]) class SidecarValidatorTests(unittest.TestCase): def setUp(self): self.validate=node_function('control_sidecar/main.py','validate_event',{'math':math,'BoundaryError':BoundaryError}) event=event_payload(result()) self.body=build_body(event,{'clip':True,'thumbnail':True},None,'https://fixture.invalid/artifacts',relative_directory(event)) def test_loitering_and_intrusion_use_identical_validation(self): for name in ('LOITERING','INTRUSION'): self.body['events'][0]['eventType']=name self.assertIsNone(self.validate(self.body,'server')) def test_extra_field_still_rejected(self): self.body['events'][0]['dwellSeconds']=5 with self.assertRaises(BoundaryError) as ctx:self.validate(self.body,'server') self.assertEqual(ctx.exception.code,'EVENT_FIELDS_INVALID') def test_other_scenarios_not_silently_enabled(self): for name in ('FIGHT','LOITERING_EXTRA'): self.body['events'][0]['eventType']=name with self.subTest(name=name),self.assertRaises(BoundaryError):self.validate(self.body,'server') def test_time_and_identity_checks_preserved(self): for key,value in (('cameraId',0),('detectedAtWallMs',None),('startWallTimeMs',True),('rtpTimestamp',2**32)): body=deepcopy(self.body);body['events'][0][key]=value with self.subTest(key=key),self.assertRaises(BoundaryError):self.validate(body,'server') with self.assertRaises(BoundaryError):self.validate(self.body,'other-server') def test_media_https_requirement_unchanged(self): for url in ('http://fixture.invalid/a','https://u:p@fixture.invalid/a','https://fixture.invalid/a?token=x'): self.body['events'][0]['clipUrl']=url with self.subTest(url=url),self.assertRaises(BoundaryError):self.validate(self.body,'server') if __name__=='__main__':unittest.main()