"""CPU-only candidate tests. No live Manager, Runtime, GPU, Redis or PostgreSQL.""" import asyncio from contextlib import asynccontextmanager from copy import deepcopy from dataclasses import replace import hashlib import importlib.util import json from pathlib import Path import sys import threading import time import types import httpx import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from starlette.datastructures import Headers from control_sidecar.auth import Auth from control_sidecar.config import Config from control_sidecar.main import create_app, validate_event, BoundedRequests from control_sidecar.protocol import BoundaryError, bearer, loads from control_sidecar.transport import Transport from app.secure_control import ControlService, Start, ControlError, CODES, build_router, validate_start from app.secure_event_transport import RelayTransport ROOT = Path(__file__).resolve().parents[1] CONTROL = 'control-test-token-00000000' RELAY = 'relay-test-token-0000000000' MANAGER = 'manager-test-token-000000' AI_TOKEN = 'ai-test-token-00000000000' # Non-production sentinels used only by fakes. No deployable credentials are stored. def configuration(**kw): values = dict(server_id='AI-test', manager='https://manager.example.invalid', runtime='http://analytics:5000', manager_ca='unused-with-mocks', runtime_token=CONTROL, relay_token=RELAY, login_body={}, password='fake') values.update(kw) return Config(**values) def start_body(camera=2): return {'cameraId': camera, 'rtspUrl': 'rtsp://camera.example.invalid/test', 'startWallTimeMs': 1758000000000, 'rtpClockRate': 90000, 'recCameraId': '00000002', 'recServerId': 'REC-test', 'scenarios': [dict(scenarioCode=code, enabled=code=='INTRUSION', rois=[], parameters={}) for code in sorted(CODES)]} def test_top_level_scenario_confidence_is_removed_and_parameters_are_fail_soft(): body = start_body() body['scenarios'][0]['confidence'] = .5 with pytest.raises(ControlError, match='SCENARIO_FIELDS_INVALID'): validate_start(body) body = start_body() body['scenarios'][0]['parameters'] = { 'unknown': 1, 'person_confidence': None, 'dwell_seconds': 'bad'} # Tuning errors are sanitized later and do not weaken the strict Start envelope. validate_start(body) def event_body(): return {'serverId':'AI-test','events':[dict(eventId='test-event',cameraId=2,eventType='INTRUSION', detectedAtWallMs=1758000005000,startWallTimeMs=1758000000000,rtpTimestamp=450000,score=.7, clipUrl='https://ai.example.invalid:9443/artifacts/cameras/2/2026-09-17/AI_120000.000+0900_INTRUSION_test-event/clip.mp4', thumbnailUrl=None,detail={'aiRecordedAtWallMs':1758000005010,'scoreMeaning':'TEMPORARY_FIXED'})]} class FakeIdentity: def __init__(self): self.closed=False self.invalidated=False self.checked=[] async def start(self): pass async def close(self): self.closed=True def current(self): if self.invalidated: raise BoundaryError('AUTH_NOT_READY') return AI_TOKEN def invalidate(self, used_token=None): self.invalidated=True async def authorize(self, token): self.checked.append(token) if token != MANAGER: raise BoundaryError('TOKEN_AUTHORITY_INVALID',403) class FakeTransport: def __init__(self,status=200,body=None): self.status=status self.body=json.dumps(body or {'result':True,'data':{'accepted':1,'duplicated':0}}).encode() self.calls=[] self.closed=False async def request(self, method, url, **kwargs): self.calls.append((method,url,kwargs)) return self.status,self.body async def close(self): self.closed=True @pytest.mark.parametrize('raw', [b'{"x":1,"x":2}', b'{"x":NaN}', b'[]', b'\xff', b'{']) def test_strict_json(raw): with pytest.raises(BoundaryError): loads(raw) def test_duplicate_authorization_rejected(): with pytest.raises(BoundaryError): bearer(Headers(raw=[(b'authorization',('Bearer '+MANAGER).encode())]*2)) @pytest.mark.parametrize('field,value', [('cameraId',True),('detectedAtWallMs',None), ('startWallTimeMs',0),('rtpTimestamp',2**32),('score',float('inf')),('score',True), ('eventType','UNKNOWN'),('clipUrl','http://ai.example.invalid/a'),('thumbnailUrl',123)]) def test_event_contract_rejections(field,value): body=event_body();body['events'][0][field]=value with pytest.raises(BoundaryError):validate_event(body,'AI-test') @pytest.mark.parametrize('status', [200,400,401,409,429,500]) def test_relay_returns_exact_response_and_one_request(status): tx,auth=FakeTransport(status),FakeIdentity() app=create_app(configuration(),tx,auth) body=json.dumps(event_body(),ensure_ascii=False,indent=2).encode() with TestClient(app) as client: response=client.post('/internal/events',content=body,headers={'Authorization':'Bearer '+RELAY}) assert response.status_code==status and response.content==tx.body assert len(tx.calls)==1 and tx.calls[0][2]['body']==body assert tx.calls[0][2]['token']==AI_TOKEN assert auth.invalidated==(status==401) assert auth.closed and tx.closed def test_unauthorized_relay_and_invalid_time_never_reach_manager(): tx,auth=FakeTransport(),FakeIdentity() with TestClient(create_app(configuration(),tx,auth)) as c: assert c.post('/internal/events',json=event_body()).status_code==401 bad=event_body();bad['events'][0]['detectedAtWallMs']=None assert c.post('/internal/events',json=bad,headers={'Authorization':'Bearer '+RELAY}).status_code==422 assert c.get('/internal/media-auth',headers={'Authorization':'Bearer wrong-test-token-000'}).status_code==403 assert not tx.calls def test_event_only_http_override_keeps_manager_and_single_attempt(): from control_sidecar.config import event_target target = event_target('http://manager.example.invalid:9090/api/v1/events', configuration().manager) tx, auth = FakeTransport(), FakeIdentity() cfg = configuration(event_endpoint=target) with TestClient(create_app(cfg, tx, auth)) as client: response = client.post('/internal/events', json=event_body(), headers={'Authorization': 'Bearer '+RELAY}) assert response.status_code == 200 and len(tx.calls) == 1 assert tx.calls[0][1] == target and tx.calls[0][2]['token'] == AI_TOKEN assert cfg.manager == 'https://manager.example.invalid' assert event_target('', cfg.manager) == cfg.manager + '/api/v1/events' @pytest.mark.parametrize('target', ['http://other.example.invalid/api/v1/events', 'http://manager.example.invalid:9090/login', 'http://manager.example.invalid/api/v1/events?x=1', 'http://user:pass@manager.example.invalid/api/v1/events']) def test_event_override_rejects_unrelated_destination(target): from control_sidecar.config import event_target with pytest.raises(ValueError): event_target(target, configuration().manager) def test_media_auth_no_event_post(): tx,auth=FakeTransport(),FakeIdentity() with TestClient(create_app(configuration(),tx,auth)) as c: assert c.get('/internal/media-auth',headers={'Authorization':'Bearer '+MANAGER}).status_code==204 assert c.get('/ws/v1').status_code==404 assert c.get('/docs').status_code==404 assert c.get('/ready').json()['scope']=='AUTH_TRANSPORT_ONLY' assert not tx.calls def test_control_auth_and_response_meaning(): tx=FakeTransport(202,{'result':'COMMAND_ACCEPTED','cameraId':2,'observedState':'STARTING'}) auth=FakeIdentity() with TestClient(create_app(configuration(),tx,auth)) as c: h={'Authorization':'Bearer '+MANAGER} assert c.post('/api/v1/cameras/start',json=start_body(),headers=h).json()=={'result':'SUCCESS','cameraId':2} assert tx.calls[0][1]=='http://analytics:5000/internal/secure-control/start' assert tx.calls[0][2]['token']==CONTROL assert json.loads(tx.calls[0][2]['body'])==start_body() tx.body=json.dumps({'result':'COMMAND_ACCEPTED','cameraId':2,'observedState':'COMMAND_PENDING'}).encode() assert c.post('/api/v1/cameras/start',json=start_body(),headers=h).json()=={'result':'SUCCESS','cameraId':2} assert c.post('/api/v1/cameras/start',content='{',headers=h).json()['result']=='FAIL' assert c.post('/api/v1/cameras/start',json=start_body()).status_code==401 assert len(tx.calls)==2 def test_sidecar_concurrency_and_deadline_cancel_the_work(): async def check(): active=0 maximum=0 ended=0 async def downstream(scope,receive,send): nonlocal active,maximum,ended active+=1;maximum=max(maximum,active) try:await asyncio.sleep(.2) finally:active-=1;ended+=1 app=BoundedRequests(downstream,seconds=.04,limit=2) async def invoke(): messages=[] async def receive():return {'type':'http.request','body':b'','more_body':False} async def send(message):messages.append(message) await app({'type':'http','path':'/test'},receive,send) return messages[0]['status'] states=await asyncio.gather(*[invoke() for _ in range(5)]) assert sorted(states)==[503,503,503,504,504] assert maximum==2 and ended==2 and active==app.active==0 asyncio.run(check()) @pytest.mark.parametrize('code', [302,307,308]) def test_http_transport_does_not_follow_redirects(code): async def check(): calls=[] def handler(request): calls.append(request) return httpx.Response(code,headers={'location':'https://other.example.invalid'}) client=httpx.AsyncClient(transport=httpx.MockTransport(handler),follow_redirects=False) tx=Transport(None,client=client) with pytest.raises(BoundaryError) as e:await tx.request('POST','https://manager.example.invalid',body=b'{}') assert e.value.code=='UPSTREAM_REDIRECT_REFUSED' and len(calls)==1 await tx.close();assert client.is_closed asyncio.run(check()) def test_http_transport_caps_response_and_deadline(): async def check(): tx=Transport(None,client=httpx.AsyncClient(transport=httpx.MockTransport(lambda r:httpx.Response(200,content=b'x'*65537)))) with pytest.raises(BoundaryError) as e:await tx.request('POST','https://manager.example.invalid') assert e.value.code=='UPSTREAM_BODY_TOO_LARGE';await tx.close() async def slow(request):await asyncio.sleep(.2);return httpx.Response(200) tx=Transport(None,client=httpx.AsyncClient(transport=httpx.MockTransport(slow))) with pytest.raises(BoundaryError):await tx.request('POST','https://manager.example.invalid',seconds=.02) await tx.close() asyncio.run(check()) def login_response(actor='AI',server='AI-test'): return {'result':True,'error':None,'serverTimeMs':int(time.time()*1000),'data':{ 'token':AI_TOKEN,'serverId':server,'actorType':actor,'accountId':'a','permission':'admin', 'expiresAtWallMs':int(time.time()*1000)+60000,'mustChangePassword':False, 'replacedSession':False,'revokedPreviousTokens':0}} def test_auth_reuses_token_until_refresh_and_closes(): async def check(): tx=FakeTransport(body=login_response());auth=Auth(configuration(),tx) await auth.start();assert auth.current()==AI_TOKEN and len(tx.calls)==1 assert auth.task is not None await auth.close();assert auth.task is None with pytest.raises(BoundaryError):auth.current() asyncio.run(check()) @pytest.mark.parametrize('actor,server',[('MANAGER','AI-test'),('AI','other')]) def test_auth_rejects_login_identity(actor,server): async def check(): auth=Auth(configuration(),FakeTransport(body=login_response(actor,server))) with pytest.raises(BoundaryError):await auth.login() asyncio.run(check()) @pytest.mark.parametrize('active,actor,permission',[(False,'MANAGER','admin'),(True,'AI','admin'),(True,'MANAGER','user')]) def test_introspection_requires_manager_admin(active,actor,permission): async def check(): tx=FakeTransport(body={'result':True,'error':None,'data':{ 'active':active,'actorType':actor,'permission':permission,'accountId':'x'}}) auth=Auth(configuration(),tx);auth.token=AI_TOKEN with pytest.raises(BoundaryError) as exc:await auth.authorize(MANAGER) assert exc.value.status==403 and len(tx.calls)==1 await auth.close() asyncio.run(check()) class Group: def __init__(self,spec): self.spec=deepcopy(spec);self.observed_state='stopped';self._pending_restart=None class Manager: def __init__(self): self._commands=threading.RLock() spec=dict(stream_id='web-cam08-test',source_id='2',camera_id=2,server_id='AI-test', source_type='rtsp',uri='rtsp://camera.example.invalid/old',analysis_fps_target=10, tracking={'enabled':True,'activation':.25},intrusion={'enabled':False,'dwell_seconds':.5}, inference_profile={'det_model':'RFDETRMedium'},store=False,active_learning=False, source_options={'latency_ms':200},autostart=False,pose={'enabled':False}) self.group=Group(spec);self.applications=0;self.starts=0;self.stops=0 def camera_stream_id(self,source): if self.group is None or self.group.spec['source_id']!=source:raise KeyError(source) return self.group.spec['stream_id'] def get(self,sid):return self.group def _saved_spec(self,group):return group.spec def apply_and_start(self,spec): self.applications+=1 if self.group is None:self.group=Group(spec) if self.group.spec!=spec or self.group.observed_state=='stopped':self.starts+=1 self.group.spec=deepcopy(spec);self.group.observed_state='starting' return self.group def stop(self,sid): self.stops+=1;self.group.observed_state='stopped';return self.group def test_mapping_preserves_cli_identity_model_source_knobs_and_deduplicates(): m=Manager();before=deepcopy(m.group.spec);service=ControlService(m,'AI-test') req=Start(json.dumps(start_body())) service.start_camera(req);service.start_camera(req) spec=m.group.spec assert spec['stream_id']=='web-cam08-test' and spec['camera_id']==2 and spec['source_id']=='2' for key in ['inference_profile','analysis_fps_target','source_options','store','active_learning','pose']: assert spec[key]==before[key] assert spec['intrusion']['dwell_seconds']==.5 and spec['intrusion']['confidence']==.5 assert spec['frame_time_anchor']=={'startWallTimeMs':1758000000000,'synthetic':False} assert spec['tracking']['enabled'] and m.starts==1 assert 'time_anchor' not in spec def test_first_roi_only_and_scenarios_off_does_not_stop_tracking(): m=Manager();d=start_body() s=next(x for x in d['scenarios'] if x['scenarioCode']=='INTRUSION') s['rois']=[{'label':'first','points':[[0,0],[1,0],[0,1]]},{'label':'second','points':[[0,0],[.5,0],[0,.5]]}] service=ControlService(m,'AI-test');service.start_camera(Start(json.dumps(d))) assert m.group.spec['intrusion']['roi_name']=='first' for s in d['scenarios']:s['enabled']=False service.start_camera(Start(json.dumps(d))) assert not m.group.spec['intrusion']['enabled'] and m.group.spec['tracking']['enabled'] @pytest.mark.parametrize('roi_form', ['missing', 'null', 'empty']) def test_unspecified_roi_uses_full_frame_and_normalizes(roi_form): m = Manager() m.group.spec['intrusion']['roi_name'] = 'previous-area' d = start_body() for scenario in d['scenarios']: if roi_form == 'missing': scenario.pop('rois') elif roi_form == 'null': scenario['rois'] = None validate_start(d) assert d == start_body() service = ControlService(m, 'AI-test') service.start_camera(Start(json.dumps(d))) assert m.group.spec['intrusion']['roi'] == [[0., 0.], [1., 0.], [1., 1.], [0., 1.]] assert 'roi_name' not in m.group.spec['intrusion'] service.start_camera(Start(json.dumps(start_body()))) assert m.starts == 1 @pytest.mark.parametrize('rois', [{}, '', False, [None], [{'label': 'bad', 'points': [[0, 0], [2, 0], [0, 1]]}]]) def test_malformed_roi_is_not_replaced_with_full_frame(rois): d = start_body() d['scenarios'][0]['rois'] = rois with pytest.raises(ControlError): validate_start(d) def test_replayed_anchor_keeps_synthetic_provenance(): m=Manager();d=start_body() m.group.spec['frame_time_anchor']={'startWallTimeMs':d['startWallTimeMs'],'synthetic':True} service=ControlService(m,'AI-test') service.start_camera(Start(json.dumps(d))) assert m.group.spec['frame_time_anchor']['synthetic'] is True d['startWallTimeMs']+=1 service.start_camera(Start(json.dumps(d))) assert m.group.spec['frame_time_anchor']['synthetic'] is False @pytest.mark.parametrize('mutation,expected',[ (lambda d:d.update(cameraId=True),'CAMERA_ID_INVALID'), (lambda d:d.update(rtpClockRate=1),'CLOCK_RATE_UNSUPPORTED'), (lambda d:d.update(startWallTimeMs=None),'ANCHOR_INVALID'), (lambda d:d.update(scenarios=[]),'SCENARIO_SNAPSHOT_REQUIRED'), (lambda d:d.update(rtspUrl='http://camera.example.invalid'),'RTSP_INVALID'), (lambda d:d.update(extra=1),'START_FIELDS_INVALID'), ]) def test_control_rejects_without_side_effects(mutation,expected): d=start_body();mutation(d);m=Manager() with pytest.raises(ControlError) as e:ControlService(m,'AI-test').start_camera(Start(json.dumps(d))) assert str(e.value)==expected and m.applications==0 def test_stop_cleanup_and_server_conflicts_are_not_overwritten(): m=Manager();service=ControlService(m,'AI-test');m.group.observed_state='stopping' with pytest.raises(ControlError,match='STOP_CLEANUP_PENDING'):service.start_camera(Start(json.dumps(start_body()))) m.group.observed_state='stopped';m.group.spec['server_id']='other' with pytest.raises(ControlError,match='SERVER_ID_CONFLICT'):service.stop_camera(2) assert m.applications==m.stops==0 def test_new_camera_requires_explicit_template_and_missing_stop_is_idempotent(): m=Manager();m.group=None;service=ControlService(m,'AI-test') assert service.stop_camera(99) is None with pytest.raises(ControlError,match='CAMERA_NOT_PROVISIONED'):service.start_camera(Start(json.dumps(start_body()))) service=ControlService(m,'AI-test',{'inference_profile':{'det_model':'RFDETRMedium'},'store':False}) service.start_camera(Start(json.dumps(start_body()))) assert m.group.spec['stream_id'].startswith('camera-') and not m.group.spec['autostart'] def test_internal_route_auth_and_command_history_cleanup(): m=Manager();app=FastAPI();app.include_router(build_router(m,CONTROL,'AI-test')) with TestClient(app) as c: assert c.post('/internal/secure-control/start',json=start_body()).status_code==401 h={'Authorization':'Bearer '+CONTROL} r=c.post('/internal/secure-control/start',json=start_body(),headers=h) assert r.status_code==202 and r.json()['observedState']=='STARTING' assert c.get('/internal/secure-control/operations/'+r.json()['operationId'],headers=h).json()['status']=='COMMAND_APPLIED' assert c.post('/internal/secure-control/stop',json={'cameraId':2},headers=h).status_code==202 assert not any(t.name=='camera-control' and t.is_alive() for t in threading.enumerate()) def test_pending_command_is_not_cancelled_by_response_budget(): m=Manager();original=m.apply_and_start def slow(spec):time.sleep(.45);return original(spec) m.apply_and_start=slow app=FastAPI();app.include_router(build_router(m,CONTROL,'AI-test')) with TestClient(app) as c: h={'Authorization':'Bearer '+CONTROL} r=c.post('/internal/secure-control/start',json=start_body(),headers=h) assert r.json()['observedState']=='COMMAND_PENDING' time.sleep(.15) assert c.get('/internal/secure-control/operations/'+r.json()['operationId'],headers=h).json()['status']=='COMMAND_APPLIED' assert m.applications==1 @pytest.mark.parametrize('state', ['ERROR', 'UNRECOGNIZED']) def test_known_error_or_invalid_runtime_state_is_not_success(state): tx=FakeTransport(202,{'result':'COMMAND_ACCEPTED','cameraId':2,'observedState':state}) with TestClient(create_app(configuration(),tx,FakeIdentity())) as c: r=c.post('/api/v1/cameras/start',json=start_body(),headers={'Authorization':'Bearer '+MANAGER}) assert r.json().get('result')!='SUCCESS' assert len(tx.calls)==1 def test_late_save_failure_remains_queryable_and_redacted(): m=Manager() def reject(spec): time.sleep(.45) raise RuntimeError('private-credential-bearing failure') m.apply_and_start=reject app=FastAPI();app.include_router(build_router(m,CONTROL,'AI-test')) with TestClient(app) as c: h={'Authorization':'Bearer '+CONTROL} r=c.post('/internal/secure-control/start',json=start_body(),headers=h) assert r.json()['observedState']=='COMMAND_PENDING' time.sleep(.15) state=c.get('/internal/secure-control/operations/'+r.json()['operationId'],headers=h).json() assert state['status']=='FAILED' and state['error']=='RUNTIME_COMMAND_REJECTED' assert 'private-credential' not in json.dumps(state) assert m.starts==0 def test_relay_transport_opt_in_and_exactly_one_invocation(tmp_path): assert RelayTransport.from_environment('https://legacy.example.invalid',{}) is None secret=tmp_path/'relay';secret.write_text(RELAY) env={'CONTEXTRA_EVENT_TRANSPORT':'sidecar','CONTEXTRA_RELAY_TOKEN_FILE':str(secret)} endpoint='http://ai-control-sidecar:8080/internal/events' relay=RelayTransport.from_environment(endpoint,env) calls=[] def send(url,body,**kw):calls.append((url,body,kw));return (True,False,200,{},None) body=event_body();relay.post(body,send) assert len(calls)==1 and calls[0][1] is body and calls[0][2]['authorization']=='Bearer '+RELAY with pytest.raises(ValueError):RelayTransport.from_environment('https://manager.example.invalid/api/v1/events',env) def artifact_module(monkeypatch): # Isolate the integration seam; do NOT load or contact the real DB module. contract=types.ModuleType('app.event_artifact_contract') for name in ['ArtifactConfig','build_body','media_path','STAGE','STAGE_DESCRIPTION','RESERVATION']: setattr(contract,name,None) store=types.ModuleType('app.event_artifact_store');store.ArtifactStore=lambda c:None monkeypatch.setitem(sys.modules,contract.__name__,contract) monkeypatch.setitem(sys.modules,store.__name__,store) name='artifact_under_secure_test' source=importlib.util.find_spec('app.event_artifact_service').origin spec=importlib.util.spec_from_file_location(name,source) module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module) return module def test_artifact_attempt_owner_unchanged_and_only_one_delivery_path(monkeypatch): module=artifact_module(monkeypatch) svc=module.ArtifactService(types.SimpleNamespace(public_base='https://ai.example.invalid/artifacts',endpoint='unused')) class Store: attempts=0 def prepare_delivery(self,row,body):return body def begin_attempt(self,row): if self.attempts>=2:return None self.attempts+=1;return self.attempts def finish_attempt(self,row,state,*args):row['state']=state svc.store=Store();calls=[] def send(endpoint,body,**kw):calls.append(body);return False,True,503,None,'TEST' monkeypatch.setattr(module,'post_event',send) svc.relay=RelayTransport('http://ai-control-sidecar:8080/internal/events',RELAY) row={'body':event_body()} svc._deliver(row);svc._deliver(row);svc._deliver(row) assert len(calls)==2 and calls[0] is calls[1] and row['state']=='FAILED' def test_artifact_blocked_time_never_invokes_relay(monkeypatch): module=artifact_module(monkeypatch) svc=module.ArtifactService(types.SimpleNamespace(public_base='x',endpoint='x')) svc.store=types.SimpleNamespace(prepare_delivery=lambda r,b:None) svc.relay=types.SimpleNamespace(post=lambda *a:pytest.fail('blocked time sent')) svc._deliver({'body':{'retained':'body'}}) def test_artifact_http_only_does_not_initialize_relay_or_ledger(monkeypatch,tmp_path): module=artifact_module(monkeypatch) calls=[] class Server: started=True;should_exit=False def __init__(self,c):pass def run(self):pass import uvicorn monkeypatch.setattr(uvicorn,'Server',Server) monkeypatch.setattr(module,'file_app',lambda r:FastAPI()) monkeypatch.setenv('CONTEXTRA_EVENT_TRANSPORT','invalid-mode-that-must-not-be-read') svc=module.ArtifactService(types.SimpleNamespace(root=tmp_path,enabled=False,port=1)) svc.store=types.SimpleNamespace(close=lambda:None,initialize=lambda:pytest.fail('DB touched'),recover=lambda:pytest.fail('DB recovered')) svc.start();assert svc.relay is None and not svc.workers assert svc.close() def test_existing_post_response_validation(monkeypatch): module=artifact_module(monkeypatch) requests=[] class Response: status=200 def __enter__(self):return self def __exit__(self,*a):pass def read(self,n):return b'{"result":true,"data":{"accepted":1,"duplicated":0}}' class Opener: def open(self,request,timeout):requests.append(request);assert timeout==3;return Response() monkeypatch.setattr(module.urllib.request,'build_opener',lambda *a:Opener()) assert module.post_event('http://ai-control-sidecar:8080/internal/events',event_body(),authorization='Bearer '+RELAY)[0] assert requests[0].get_header('Authorization')=='Bearer '+RELAY def test_modules_do_not_import_sqlite_or_gpu(): import ast for path in (ROOT/'control_sidecar').glob('*.py'): tree=ast.parse(path.read_text()) for node in ast.walk(tree): if isinstance(node,ast.Import):names=[n.name for n in node.names] elif isinstance(node,ast.ImportFrom):names=[node.module or ''] else:continue assert not any(n.startswith(('sqlite3','torch','gi','pyds','app.','psycopg')) for n in names) def test_invalid_crowd_skip_is_recorded_not_enabled(): m=Manager();d=start_body() next(s for s in d['scenarios'] if s['scenarioCode']=='CROWD')['enabled']=True ControlService(m,'AI-test').start_camera(Start(json.dumps(d))) assert m.group.spec['secure_control_projection']['skippedScenarios']==['CROWD'] assert not m.group.spec['crowd']['enabled'] assert 'fire' not in m.group.spec and m.group.spec['inference_profile']=={'det_model':'RFDETRMedium'} def test_secure_factory_keeps_baseline_routes_and_lifespan(monkeypatch,tmp_path): token=tmp_path/'control';token.write_text(CONTROL) monkeypatch.setenv('CONTEXTRA_CONTROL_TOKEN_FILE',str(token)) monkeypatch.setenv('AI_CONTROL_SERVER_ID','AI-test') monkeypatch.delenv('CONTEXTRA_CONTROL_TEMPLATE_FILE',raising=False) events=[] base=types.ModuleType('app.main') @asynccontextmanager async def lifespan(app): events.append('baseline-start') yield events.append('baseline-stop') def factory(): app=FastAPI(lifespan=lifespan) @app.get('/legacy-marker') def marker():return {'legacy':True} return app base.create_app=factory manager=types.ModuleType('app.stream_manager');manager.manager=Manager() monkeypatch.setitem(sys.modules,'app.main',base) monkeypatch.setitem(sys.modules,'app.stream_manager',manager) from app.secure_runtime import create_app as factory with TestClient(factory()) as client: assert client.get('/legacy-marker').json()=={'legacy':True} assert client.post('/internal/secure-control/stop',json={'cameraId':2},headers={'Authorization':'Bearer '+CONTROL}).status_code==202 assert events==['baseline-start','baseline-stop'] def test_command_owner_coalesces_conflicts_and_bounds_queue(): from app.application.camera_commands import CameraCommands, CommandUnavailable entered,release=threading.Event(),threading.Event() class Service: def start_camera(self,request):entered.set();release.wait(1);return None owner=CameraCommands(Service(),capacity=1,history=2) try: first=owner.submit(2,Start('same')) assert entered.wait(1) assert owner.submit(2,Start('same')) is first with pytest.raises(CommandUnavailable):owner.submit(2,Start('different')) with pytest.raises(CommandUnavailable):owner.submit(3,Start('other')) finally:release.set();assert owner.close(2) with pytest.raises(CommandUnavailable):owner.submit(2,Start('same'))