import asyncio import json from pathlib import Path from unittest.mock import patch import pytest from fastapi.testclient import TestClient from websockets.asyncio.server import unix_serve import device_gateway.app as gateway from device_gateway.control.approvals import ApprovalRegistry from device_gateway.control.codex import CodexAdapter from device_gateway.control.models import ControlError, Project from device_gateway.control.pi_approval import PiApprovalServer from device_gateway.control.projects import ProjectRegistry from device_gateway.control.rpc import JsonUnixSocket from device_gateway.auth import DeviceCredentialStore from device_gateway.state import TargetStore def create(registry, identity='1', agent='pi', session='s', owner='runtime'): return registry.create(agent=agent, session_id=session, project_id='p', kind='command', title='Run command?', detail='touch /tmp/approval-test', owner=owner, request_identity=identity) def test_registry_decisions_fifo_identity_and_deadlines(): async def exercise(): registry = ApprovalRegistry(timeout=.03) a = create(registry) b = create(registry, '2', 'codex', 'other') assert registry.list_pending() == [a, b] with pytest.raises(ControlError): create(registry) registry.presentations['device-connection'] = a.approval_id assert not registry.resolve(b.approval_id, 'approve', device_connection='device-connection') assert not registry.resolve(a.approval_id, 'approve', device_connection='old-connection') assert not registry.resolve(a.approval_id, 'acceptForSession') assert registry.resolve(a.approval_id, 'approve', device_connection='device-connection') assert await registry.wait(a) == 'APPROVED' assert not registry.resolve(a.approval_id, 'reject') assert registry.resolve(b.approval_id, 'reject') assert await registry.wait(b) == 'REJECTED' c = create(registry, '3') waiter = asyncio.create_task(registry.wait(c)) await asyncio.sleep(0) with patch.object(asyncio.get_running_loop(), 'time', return_value=10**12): registry.expire() assert c.state == 'PENDING' and not waiter.done() assert registry.wire(c)['expires_in'] is None registry.cancel(c.approval_id) assert await waiter == 'CANCELLED' d = create(registry, '4'); registry.cancel_owner('runtime') assert await registry.wait(d) == 'CANCELLED' e = create(registry, '5'); registry.close() assert await registry.wait(e) == 'CANCELLED' assert ApprovalRegistry().list_pending() == [] assert not registry.resolve({}, 'approve') asyncio.run(exercise()) def test_codex_real_socket_nonblocking_disconnect_and_no_replay(tmp_path): async def exercise(): projects = ProjectRegistry(tmp_path/'projects.json') projects.add(Project(id='p', name='P', path=str(tmp_path))) registry = ApprovalRegistry(timeout=.5) replies, peers = [], [] async def handler(ws): peers.append(ws) async for raw in ws: m = json.loads(raw) if 'method' not in m: replies.append((len(peers), m)); continue if 'id' not in m: continue result = {} if m['method'] == 'thread/read': result = {'thread': {'id': 'thread', 'cwd':str(tmp_path), 'updatedAt':1, 'status':{'type':'active'}}} await ws.send(json.dumps({'id':m['id'], 'result':result})) async with unix_serve(handler, str(tmp_path/'codex.sock'), compression=None): adapter = CodexAdapter(projects, approvals=registry) adapter.rpc = JsonUnixSocket(str(tmp_path/'codex.sock'), on_event=adapter.event, on_disconnect=adapter.disconnected) async def request(rid, method='commandExecution'): await peers[-1].send(json.dumps({'id':rid, 'method':f'item/{method}/requestApproval', 'params':{'threadId':'thread','turnId':'turn','itemId':str(rid),'startedAtMs':1, 'command':'touch /tmp/test', 'availableDecisions':['accept','cancel']}})) for _ in range(100): if registry.list_pending(): return registry.list_pending()[0] await asyncio.sleep(.005) raise AssertionError('approval not created') try: await adapter.ready() for rid, decision, expected, kind in [(101,'approve','accept','commandExecution'), ('file-2','reject','decline','fileChange')]: if kind == 'fileChange': await peers[-1].send(json.dumps({'method':'item/started','params':{'threadId':'thread', 'item':{'type':'fileChange','id':str(rid),'changes':[{'path':'/tmp/change.txt'}]}}})) approval = await request(rid, kind) if kind == 'fileChange': assert '/tmp/change.txt' in approval.detail # reader still serves ordinary RPC while request awaits a human assert (await adapter.get_session('thread')).state == 'WAITING_APPROVAL' registry.resolve(approval.approval_id, decision) for _ in range(100): if replies and replies[-1][1]['id'] == rid: break await asyncio.sleep(.005) assert replies[-1][1] == {'id':rid, 'result':{'decision':expected}} old = await request(103) await peers[-1].close() await adapter.rpc.reader assert old.state == 'CANCELLED' assert not registry.resolve(old.approval_id, 'approve') await adapter.ready() new = await request(103) # daemon may reuse an ID on a new connection assert new.approval_id != old.approval_id assert not registry.resolve(old.approval_id,'approve') assert new.state == 'PENDING' registry.resolve(new.approval_id,'reject') await asyncio.sleep(.02) assert [m for connection,m in replies if connection == 2] == [{'id':103,'result':{'decision':'decline'}}] timeout = await request(104) with patch.object(asyncio.get_running_loop(), 'time', return_value=10**12): registry.expire() assert timeout.state == 'PENDING' registry.cancel(timeout.approval_id) await asyncio.sleep(.02) assert replies[-1][1] == {'id':104,'result':{'decision':'cancel'}} external = await request(105) await peers[-1].send(json.dumps({'method':'serverRequest/resolved','params':{'threadId':'thread','requestId':105}})) assert await registry.wait(external) == 'CANCELLED' shutdown = await request(106) await adapter.close() assert shutdown.state == 'CANCELLED' assert not any(m['id'] in (105,106) for _,m in replies) finally: await adapter.close(); registry.close() asyncio.run(exercise()) def test_pi_unix_channel_timeout_disconnect_shutdown_and_parallel_sessions(tmp_path): async def exercise(): projects = ProjectRegistry(tmp_path/'projects.json') projects.add(Project(id='p',name='P',path=str(tmp_path))) registry = ApprovalRegistry(timeout=.08) server = PiApprovalServer(registry, projects, tmp_path/'runtime/approvals.sock') await server.start() async def submit(identity, session='s'): reader, writer = await asyncio.open_unix_connection(server.path) writer.write((json.dumps(dict(session_id=session,runtime_id='runtime',tool_call_id=identity, cwd=str(tmp_path),kind='file_change',title='Edit file?',detail='/tmp/test'))+'\n').encode()) await writer.drain() for _ in range(100): found = [a for a in registry.list_pending() if json.loads(a.request_identity)[1] == identity] if found: assert json.loads(await reader.readline())['state'] == 'PENDING' return reader, writer, found[0] await asyncio.sleep(.001) raise AssertionError('Pi approval not created') try: r1,w1,a1=await submit('one'); r2,w2,a2=await submit('two','other') registry.resolve(a2.approval_id,'approve') assert json.loads(await r2.readline())['decision']=='approve' assert a1.state=='PENDING' # another session proceeds independently registry.resolve(a1.approval_id,'reject') assert json.loads(await r1.readline())['decision']=='reject' w1.close();w2.close() r,w,a=await submit('indefinite') with patch.object(asyncio.get_running_loop(), 'time', return_value=10**12): registry.expire() assert a.state == 'PENDING' registry.cancel(a.approval_id) reply = json.loads(await r.readline()) assert reply['state'] == 'CANCELLED' and reply['decision'] == 'cancel' w.close() r,w,a=await submit('disconnect');w.close();await w.wait_closed() for _ in range(100): if a.state=='CANCELLED':break await asyncio.sleep(.001) assert a.state=='CANCELLED' r,w,a=await submit('shutdown');await server.close() assert a.state=='CANCELLED';w.close() finally: await server.close();registry.close() asyncio.run(exercise()) def test_authenticated_device_fifo_admin_and_reconnect(tmp_path, monkeypatch): cred=tmp_path/'device.json';cred.write_text(json.dumps({'device_id':'test','device_token':'secret'})) monkeypatch.setattr(gateway,'credential_store',DeviceCredentialStore(cred)) monkeypatch.setattr(gateway,'target_store',TargetStore(tmp_path/'targets.json')) monkeypatch.setattr(gateway,'devices',{}) monkeypatch.setattr(gateway,'adapters',{}) monkeypatch.setattr(gateway,'project_registry',ProjectRegistry(tmp_path/'projects.json')) with TestClient(gateway.app) as client: async def add(identity): return create(gateway.approvals,identity) def receive(ws, kind): for _ in range(30): m=ws.receive_json() if m['type']==kind:return m raise AssertionError(kind) def hello(ws): ws.send_json({'type':'hello','device_id':'test','device_token':'secret','board':'m5stack-stopwatch','firmware':'0.3.0'}) receive(ws,'hello_ack') with client.websocket_connect('/device/ws') as ws: hello(ws) a=client.portal.call(add,'one'); b=client.portal.call(add,'two') assert receive(ws,'approval_request')['approval_id']==a.approval_id assert len(client.get('/api/approvals').json())==2 ws.send_json({'type':'button_event','button':'A','action':'click'}) ws.send_json({'type':'approval_response','approval_id':b.approval_id,'decision':'approve'}) assert len(client.get('/api/approvals').json())==2 ws.send_json({'type':'approval_response','approval_id':a.approval_id,'decision':'approve'}) assert receive(ws,'approval_cancelled')['state']=='APPROVED' assert receive(ws,'approval_request')['approval_id']==b.approval_id ws.send_json({'type':'approval_response','approval_id':a.approval_id,'decision':'approve'}) ws.send_json({'type':'approval_response','approval_id':{},'decision':'approve'}) assert client.get('/api/approvals').json()[0]['approval_id']==b.approval_id ws.send_json({'type':'approval_response','approval_id':b.approval_id,'decision':'reject'}) assert receive(ws,'approval_cancelled')['state']=='REJECTED' c=client.portal.call(add,'three');receive(ws,'approval_request') with client.websocket_connect('/device/ws') as ws: hello(ws) assert receive(ws,'approval_request')['approval_id']==c.approval_id for decision in ['approve','reject']: if decision=='reject': c=client.portal.call(add,'four');receive(ws,'approval_request') assert client.post(f'/api/approvals/{c.approval_id}/decision',json={'decision':decision}).status_code==200 assert receive(ws,'approval_cancelled')['state']==('APPROVED' if decision=='approve' else 'REJECTED') assert client.post(f'/api/approvals/{c.approval_id}/decision',json={'decision':decision}).status_code==409 assert client.get('/api/approvals').json()==[] assert gateway.target_store.get('test')=='server' def test_broadcast_first_decision_wins_and_cancels_other_device(monkeypatch): class Peer: def __init__(self): self.sent=[] async def send_json(self, payload): self.sent.append(payload) async def exercise(): registry=ApprovalRegistry();monkeypatch.setattr(gateway,'approvals',registry) peers=[Peer(),Peer()] devices=[gateway.Device(str(i),'simulated','0.3.0',peer) for i,peer in enumerate(peers)] monkeypatch.setattr(gateway,'devices',{d.device_id:d for d in devices}) a=create(registry) await gateway.synchronize_approvals() assert all(p.sent[-1]['approval_id']==a.approval_id for p in peers) assert registry.resolve(a.approval_id,'reject',device_connection=devices[1].connection_id) assert not registry.resolve(a.approval_id,'approve',device_connection=devices[0].connection_id) await gateway.synchronize_approvals() assert all(p.sent[-1]['type']=='approval_cancelled' and p.sent[-1]['state']=='REJECTED' for p in peers) registry.close() asyncio.run(exercise())