import asyncio import base64 import io import json import wave from types import SimpleNamespace from unittest.mock import patch import pytest from fastapi.testclient import TestClient import device_gateway.app as gateway from device_gateway.auth import DeviceCredentialStore from device_gateway.control.approvals import ApprovalRegistry from device_gateway.control.models import AgentSession, ControlError, Project, Settings from device_gateway.control.projects import ProjectRegistry from device_gateway.control.settings import quick_catalog, validate_settings, settings_view from device_gateway.state import TargetStore from device_gateway.voice import VoiceManager, MAX_BYTES, MAX_CHUNKS, wav_pcm class FakeAgent: def __init__(self, agent, path): self.agent, self.path = agent, str(path) self.inputs = [] self.model = 'model-0' self.effort = 'high' if agent == 'codex' else None self.state = 'DORMANT' self.updates = 0 async def get_session(self, sid): return AgentSession(agent=self.agent,session_id=sid,project_id='p',name='Test session',cwd=self.path, updated_at='2026-10-03T00:00:00Z',model=self.model,effort=self.effort,state=self.state) async def list_sessions(self, project): return [await self.get_session('s')] async def send_input(self, sid, text): self.inputs.append((sid, text)) return dict(accepted=True, session_id=sid, resumed=self.state=='DORMANT') async def get_settings(self, sid): models = [dict(id=f'model-{i}',name=f'Model {i}',is_default=i==2, available_efforts=['low','high','ultra'] if self.agent=='codex' else [], default_effort='high' if self.agent=='codex' else None) for i in range(9)] return settings_view(self.agent,sid,self.model,self.effort,models) async def update_settings(self, sid, value, *, still_valid=lambda:True): if self.state in ('RUNNING','WAITING_APPROVAL'): raise ControlError('session busy',409) if not still_valid(): raise ControlError("cancelled",409) self.model,self.effort=validate_settings(value, await self.get_settings(sid)) self.updates+=1 return await self.get_settings(sid) async def close(self): pass class FakeSTT: def __init__(self, error=False): self.error, self.wav = error, None async def transcribe(self, data): self.wav = data if self.error: raise RuntimeError('private provider error') return '检查一下 admin 切换为什么失败' def start(vid='v'): return dict(type='voice_start',voice_id=vid,sample_rate=44100,bits=16,channels=1) def chunk(sequence=0, vid='v', pcm=b'\0\0'*1764): return dict(type='voice_chunk',voice_id=vid,sequence=sequence,audio=base64.b64encode(pcm).decode()) @pytest.mark.parametrize('agent',['pi','codex']) def test_authenticated_device_voice_and_settings_e2e(tmp_path,monkeypatch,agent): cred=tmp_path/'device.json';cred.write_text(json.dumps(dict(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',{}) projects=ProjectRegistry(tmp_path/'projects.json');projects.add(Project(id='p',name='P',path=str(tmp_path))) monkeypatch.setattr(gateway,'project_registry',projects) adapter=FakeAgent(agent,tmp_path);provider=FakeSTT() monkeypatch.setattr(gateway,'adapters',{agent:adapter}) monkeypatch.setattr(gateway.voices,'provider',provider) with TestClient(gateway.app) as client: with client.websocket_connect('/device/ws') as ws: def receive(kind): for _ in range(30): value=ws.receive_json() if value['type']==kind:return value raise AssertionError(kind) ws.send_json(dict(type='hello',device_id='test',device_token='secret',board='synthetic',firmware='0.4.0')) receive('hello_ack');receive('target_state') # Overview never guesses a session. ws.send_json(start('overview')) assert receive('voice_error')['message']=='NO AGENT SELECTED' target=f'agent:{agent}:s' assert client.post('/devices/test/target',json={'target':target}).status_code==200 assert receive('agent_state')['agent']==agent ws.send_json(start());assert receive('voice_state')['state']=='RECORDING' ws.send_json(chunk());ws.send_json(dict(type='voice_end',voice_id='v')) transcript=receive('voice_transcript') assert transcript['name']=='Test session' and transcript['agent']==agent assert adapter.inputs==[] # STT alone is never permission to send with wave.open(io.BytesIO(provider.wav)) as audio: assert (audio.getframerate(),audio.getsampwidth(),audio.getnchannels())==(44100,2,1) assert audio.getnframes()==1764 ws.send_json(dict(type='voice_decision',voice_id='stale',decision='send')) ws.send_json(dict(type='voice_decision',voice_id='v',decision='send')) assert receive('voice_state')['state']=='SENT' ws.send_json(dict(type='voice_decision',voice_id='v',decision='send')) ws.send_json(dict(type='settings_get',request_id='r')) settings=receive('settings_state') assert len(settings['available_models'])==6 and settings['available_models'][0]['id']=='model-0' assert adapter.inputs==[('s',transcript['transcript'])] assert 'audio' not in json.dumps(client.get('/devices').json()) ws.send_json(dict(type='settings_apply',request_id='r',target=target,settings={'model':'model-1'})) assert receive('settings_updated')['model']=='model-1' ws.send_json(dict(type='settings_apply',request_id='r',target=target,settings={'model':'model-2'})) assert client.get(f'/api/agents/{agent}/sessions/s/settings').json()['model']=='model-1' assert adapter.updates==1 adapter.state='RUNNING' assert client.post(f'/api/agents/{agent}/sessions/s/settings',json={'model':'model-2'}).status_code==409 adapter.state='WAITING_APPROVAL' assert client.post(f'/api/agents/{agent}/sessions/s/settings',json={'model':'model-2'}).status_code==409 adapter.state='DONE' ws.send_json(start('switch'));receive('voice_state') client.post('/devices/test/target',json={'target':'server'}) assert receive('voice_state')['state']=='CANCELLED' receive('target_state') client.post('/devices/test/target',json={'target':target});receive('agent_state') ws.send_json(start('preempt'));receive('voice_state') async def approve(): return gateway.approvals.create(agent=agent,session_id='s',project_id='p',kind='command', title='Run?',detail='pwd',owner='test',request_identity='one') approval=client.portal.call(approve) receive('approval_request') assert not gateway.voices.requests ws.send_json(dict(type='voice_decision',voice_id='preempt',decision='send')) assert len(adapter.inputs)==1 ws.send_json(dict(type='approval_response',approval_id=approval.approval_id,decision='reject')) assert receive('approval_cancelled')['state']=='REJECTED' ws.send_json(start('disconnect'));receive('voice_state') # Socket close invalidates capture memory and identity. assert not gateway.voices.requests def test_voice_limits_failures_and_async_invalidation(): async def exercise(): sent=[];target=['agent:pi:s'];approval=[False];current=[True] device=SimpleNamespace(connection_id='connection',device_id='d') adapter=FakeAgent('pi','/tmp') async def send(d,m):sent.append(m) manager=VoiceManager(FakeSTT(),lambda _:target[0],lambda _:adapter,send,lambda _:current[0],lambda:approval[0]) try: for index,bad in enumerate([chunk(1),chunk(pcm=b'x'),chunk(pcm=b'x'*4412),dict(type='voice_chunk',voice_id='v',sequence=0,audio='not base64')]): vid=f'bad{index}';await manager.handle(device,start(vid));bad['voice_id']=vid await manager.handle(device,bad);await asyncio.sleep(0) assert not manager.requests assert sent[-1]['state']=='CANCELLED' for index,field,value in [(0,'pcm',bytearray(MAX_BYTES)),(1,'sequence',MAX_CHUNKS)]: vid=f'overflow{index}';await manager.handle(device,start(vid)) req=manager.requests['connection'];setattr(req,field,value) await manager.handle(device,chunk(vid=vid,sequence=req.sequence));assert not manager.requests assert len(req.pcm)==0 await manager.handle(device,start('error'));await manager.handle(device,chunk(vid='error')) manager.provider=FakeSTT(error=True) await manager.handle(device,dict(type='voice_end',voice_id='error')) await asyncio.gather(*list(manager.tasks));assert sent[-1]['message']=='Transcription failed' assert adapter.inputs==[] # A delayed provider result cannot resurrect a cancelled transcript. class Delayed: async def transcribe(self,data): try:await asyncio.Future() except asyncio.CancelledError:return 'stale transcript' manager.provider=Delayed() for reason in ('target','approval','disconnect','cancel'): await manager.handle(device,start(reason));await manager.handle(device,chunk(vid=reason)) await manager.handle(device,dict(type='voice_end',voice_id=reason));await asyncio.sleep(0);await asyncio.sleep(0) req=manager.requests['connection'] if reason=='target':target[0]='agent:codex:other' if reason=='approval':approval[0]=True if reason=='disconnect':current[0]=False manager.cancel(device,notify=False) await asyncio.gather(*list(manager.tasks), return_exceptions=True) assert req.state=='CANCELLED' and req.transcript=='' and not req.pcm target[0]='agent:pi:s';approval[0]=False;current[0]=True # Capture deadline triggers transcription, never agent input. manager.provider=FakeSTT() await manager.handle(device,start('max'));await manager.handle(device,chunk(vid='max')) req=manager.requests['connection'];req.timer._callback() await asyncio.sleep(0);await asyncio.sleep(0) await asyncio.gather(*list(manager.tasks)) assert req.state=='AWAITING_CONFIRMATION' and adapter.inputs==[] await manager.handle(device,dict(type='voice_decision',voice_id='max',decision='cancel')) assert not manager.requests await manager.handle(device,start('max')) # replayed ID stays inert assert not manager.requests finally:await manager.close() asyncio.run(exercise()) def test_settings_catalog_priorities_and_model_effort_validation(): models=[dict(id=str(i),available_efforts=['low','experimental'],default_effort='low',is_default=i==5) for i in range(12)] settings=settings_view('codex','s','4','experimental',models) quick=quick_catalog(settings,['9','4','8','9']) assert [m['id'] for m in quick['available_models']]==['4','9','8','5','0','1'] assert validate_settings(Settings(model='5',effort='experimental'),settings)==('5','experimental') with pytest.raises(ControlError):validate_settings(Settings(model='missing'),settings) with pytest.raises(ControlError):validate_settings(Settings(model='5',effort='invented'),settings) assert len(settings['available_models'])==12 def test_openai_compatible_stt_multipart_wav_and_safe_failure(monkeypatch): import httpx from device_gateway.voice import OpenAICompatibleSTT actual_client=httpx.AsyncClient seen=[] def handle(request): body=request.read();seen.append(request) assert request.url=='https://stt.example/v1/audio/transcriptions' assert request.headers['authorization']=='Bearer test-key' assert b'RIFF' in body and b'WAVE' in body and b'voice.wav' in body and b'test-transcriber' in body return httpx.Response(200,json={'text':'public test phrase'}) def client(**kwargs): assert kwargs['timeout']==25 and kwargs['follow_redirects'] is False return actual_client(**kwargs,transport=httpx.MockTransport(handle)) monkeypatch.setattr(httpx,'AsyncClient',client) async def exercise(): provider=OpenAICompatibleSTT('https://stt.example/v1/','test-key','test-transcriber') assert await provider.transcribe(wav_pcm(b'\0\0'*1764))=='public test phrase' assert len(seen)==1 asyncio.run(exercise())