from copy import deepcopy from concurrent.futures import ThreadPoolExecutor import json from pathlib import Path from threading import Event from unittest.mock import patch import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from app.multi_roi.errors import RoiError from app.multi_roi.policy import PolicyStore, DEFAULT, atomic, digest from app.secure_control import ControlService, Start, build_router from app.application.camera_commands import CameraCommands, CommandUnavailable from multi_roi_fixtures import Manager, request, LEFT TOKEN = 'synthetic-local-test-token' HEADERS = {'Authorization':'Bearer '+TOKEN} PATH = '/internal/secure-control/roi-policy' def change(store, **changes): status = store.status() return store.apply(status['revision'], changes, configured_revision=status['configuredRevision']) def service(tmp_path): m = Manager() s = ControlService(m, 'server', dict(store=False), roi_policy_path=tmp_path/'policy.json') change(s.roi_policy, mode='multi_roi') return m, s def test_defaults_explicit_activation_restart_and_invalid_direct_edit(tmp_path): path = tmp_path/'policy.json' atomic(path, dict(DEFAULT, mode='multi_roi')) store = PolicyStore(path) assert store.value['mode'] == 'legacy' and store.status()['pending'] change(store) assert PolicyStore(path).value['mode'] == 'multi_roi' path.write_text('{invalid') assert store.status()['error'] == 'INVALID_POLICY' assert PolicyStore(path).value['mode'] == 'multi_roi' def test_policy_conflict_keeps_effective_and_disk(tmp_path): m, s = service(tmp_path) d = request('FIRE') d['rois'] = [dict(roiId=1,points=LEFT),dict(roiId=2,points=LEFT)] d['rules'][0]['roiId'] = 1 s.start_camera(Start(json.dumps(d))) before = s.roi_policy.applied_path.read_bytes() with pytest.raises(RoiError, match='POLICY_LIMIT_CONFLICT') as error: change(s.roi_policy, max_rois=1) assert error.value.details['cameraId'] == 1 assert s.roi_policy.applied_path.read_bytes() == before assert next(iter(m._groups.values())).observed_state == 'running' d['rois'].pop() s.start_camera(Start(json.dumps(d))) change(s.roi_policy, max_rois=1) assert s.roi_policy.value['max_rois'] == 1 def test_reserved_change_rejects_existing_regular_full_frame_id(tmp_path): m, s = service(tmp_path) d = request('FIRE') d['rois'] = [dict(roiId=9,points=[[0,0],[1,0],[1,1],[0,1]])] d['rules'][0]['roiId'] = 9 s.start_camera(Start(json.dumps(d))) with pytest.raises(RoiError, match='POLICY_ID_CONFLICT'): change(s.roi_policy, full_frame_roi_id=9) def test_same_request_id_change_stop_and_full_snapshot(tmp_path): m, s = service(tmp_path) d = request('INTRUSION','LOITERING') a = s.start_camera(Start(json.dumps(d))) d['rules'].reverse() again = s.start_camera(Start(json.dumps(d))) assert again.summary['action'] == 'RUN_REUSED' and m.starts == 1 d['rules'][0]['ruleId'] = 50 s.start_camera(Start(json.dumps(d))) assert m.starts == 2 d['rules'] = [] out = s.start_camera(Start(json.dumps(d))) assert len(out.summary['removedRules']) == 2 assert not out.group.spec['intrusion']['enabled'] assert not out.group.spec['loitering']['enabled'] s.stop_camera(1) s.start_camera(Start(json.dumps(d))) assert m.starts == 4 def test_invalid_start_keeps_existing_run_and_never_falls_back(tmp_path): m, s = service(tmp_path) a = s.start_camera(Start(json.dumps(request('FIRE')))) before = deepcopy(a.group.spec) d = request('FIRE') d['rules'][0]['roiId'] = 500 with pytest.raises(RoiError, match='ROI_REFERENCE_NOT_FOUND'): s.start_camera(Start(json.dumps(d))) assert a.group.spec == before and m.starts == 1 change(s.roi_policy, mode='legacy') with pytest.raises(RoiError, match='CONTRACT_MODE_MISMATCH'): s.start_camera(Start(json.dumps(request('FIRE')))) def test_concurrent_writers_stale_cas_and_persistence_failure(tmp_path): s = PolicyStore(tmp_path/'policy.json') before = s.status() def apply(n): try: s.apply(before['revision'], {'max_rois':n}, configured_revision=before['configuredRevision']) return True except RoiError: return False with ThreadPoolExecutor(2) as workers: assert sum(workers.map(apply, (10,11))) == 1 old = s.value with patch('app.multi_roi.policy.atomic', side_effect=OSError('synthetic-secret')): with pytest.raises(RoiError, match='POLICY_PERSISTENCE_FAILED') as error: change(s, max_rois=20) assert 'synthetic-secret' not in str(error.value.payload()) assert s.value == old and PolicyStore(s.path).value == old def test_crash_between_desired_and_applied_requires_explicit_resume(tmp_path): s = PolicyStore(tmp_path/'policy.json') original = atomic def fail_applied(path, value): if path == s.applied_path: raise OSError('disk') original(path,value) with patch('app.multi_roi.policy.atomic', side_effect=fail_applied): with pytest.raises(RoiError): change(s, max_rois=12) restarted = PolicyStore(s.path) assert restarted.value['max_rois'] == 8 and restarted.status()['pending'] status = restarted.status() restarted.apply(status['revision'], configured_revision=status['configuredRevision']) assert restarted.value['max_rois'] == 12 def test_symlink_and_permissions(tmp_path): target = tmp_path/'real.json' target.write_text(json.dumps(DEFAULT)) link = tmp_path/'link.json' link.symlink_to(target) with pytest.raises(RoiError): PolicyStore(link) target.chmod(0o640) s = PolicyStore(target) change(s, max_rois=9) assert target.stat().st_mode & 0o777 == 0o640 assert s.applied_path.stat().st_mode & 0o777 == 0o600 def test_same_pending_request_and_stop_reaped(tmp_path): m, s = service(tmp_path) entered, release = Event(), Event() original = m.apply_and_start def delayed(spec): entered.set() assert release.wait(3) return original(spec) m.apply_and_start = delayed owner = CameraCommands(s, capacity=4, history=8) try: command = Start(json.dumps(request('FIRE'))) job = owner.submit(1, command, 'start') assert entered.wait(2) assert owner.submit(1, command, 'start') is job with pytest.raises(CommandUnavailable): owner.submit(1, None, 'stop') release.set() job.future.result(3) # close joins the worker, so no active entry can race the next command owner. assert owner.close(3) owner = CameraCommands(s, capacity=4, history=8) stop = owner.submit(1, None, 'stop') stop.future.result(3) assert next(iter(m._groups.values())).observed_state == 'stopped' finally: release.set() assert owner.close(3) def test_authenticated_api_mode_policy_conflict_and_safe_error(tmp_path): m = Manager() app = FastAPI() app.include_router(build_router(m, TOKEN, 'server', {}, roi_policy_path=tmp_path/'policy.json')) with TestClient(app) as client: assert client.get(PATH).status_code == 401 assert not (tmp_path/'policy.json').exists() status = client.get(PATH,headers=HEADERS).json() body = dict(expectedRevision=status['revision'],configuredRevision=status['configuredRevision'], changes={'mode':'multi_roi'}) assert client.patch(PATH,json=body,headers=HEADERS).status_code == 200 d = request('FIRE') response = client.post('/internal/secure-control/start',json=d,headers=HEADERS) assert response.status_code == 202 and response.json()['change']['action'] == 'NEW_RUN_REQUESTED' d['rtspUrl'] = 'rtsp://user:secret@example.invalid/video' d['rules'][0]['roiId'] = 99 rejected = client.post('/internal/secure-control/start',json=d,headers=HEADERS) assert rejected.status_code == 422 assert rejected.json()['error'] == 'ROI_REFERENCE_NOT_FOUND' assert 'secret' not in rejected.text and 'user:' not in rejected.text assert rejected.json()['existingRunPreserved'] is True m.fail = True changed = request('INTRUSION') failure = client.post('/internal/secure-control/start',json=changed,headers=HEADERS) assert failure.status_code == 409 and failure.json()['observedState'] == 'ERROR'