"""弱网回归：使用真实 VoiceManager 状态机，不 mock 状态迁移。""" import asyncio from types import SimpleNamespace import pytest import json from fastapi.testclient import TestClient import device_gateway.app as gateway from device_gateway.auth import DeviceCredentialStore from device_gateway.state import TargetStore from device_gateway.voice import VoiceManager from test_voice_settings import FakeAgent, FakeSTT, start, chunk from test_navigator import Agent, registry from device_gateway.control.navigator import Navigator from device_gateway.control.operations import Operations from device_gateway.control.user_input import UserInputs @pytest.mark.parametrize('agent', ['pi', 'codex']) @pytest.mark.parametrize('offline_complete', [False, True]) @pytest.mark.parametrize('invalidation', [None, 'aba', 'approval', 'native_pending']) def test_stt_reconnect_with_real_navigator_context(tmp_path, agent, offline_complete, invalidation): async def run(): old = SimpleNamespace(device_id='d', connection_id='old', capabilities={'target-select'}) new = SimpleNamespace(device_id='d', connection_id='new', capabilities={'target-select'}) current = [old] approval = [False] targets = TargetStore(tmp_path / 'targets.json') targets.set('d', f'agent:{agent}:s0', 'p') adapter = Agent(agent, tmp_path) inputs = UserInputs() messages = [] ready, release = asyncio.Event(), asyncio.Event() async def send(device, event): messages.append((device.connection_id, event)) async def payload(target): return {} navigator = Navigator(registry(tmp_path), {agent: adapter}, targets, Operations(tmp_path / 'operations.db'), inputs, send, lambda d: current[0] is d, payload, lambda d: None) class Provider(FakeSTT): calls = 0 async def transcribe(self, data): self.calls += 1 ready.set() await release.wait() return await super().transcribe(data) provider = Provider() manager = VoiceManager(provider, targets.get, lambda _: adapter, send, lambda d: current[0] is d, lambda: approval[0], targets.generation, context_prepare=navigator.voice_prepare, context_valid=navigator.voice_valid, context_submit=navigator.voice_submit) try: await manager.handle(old, dict(start(), context=dict(kind='message', target=targets.get('d'), expected_target_generation=targets.generation('d')))) await manager.handle(old, chunk()) await manager.handle(old, dict(type='voice_end', voice_id='v', total_bytes=3528)) await ready.wait() req = manager.requests['d'] current[0] = None if not offline_complete: current[0] = new await manager.handle(new, dict(type='voice_status', voice_id='v')) assert req.state == 'TRANSCRIBING' release.set() await asyncio.gather(*list(manager.tasks)) assert req.state == 'AWAITING_CONFIRMATION' assert provider.calls == 1 and not req.pcm and adapter.inputs == [] if offline_complete: assert not any(event['type']=='voice_transcript' for _,event in messages) if invalidation == 'aba': targets.set('d', f'agent:{agent}:s1'); targets.set('d', f'agent:{agent}:s0') elif invalidation == 'approval': approval[0] = True elif invalidation == 'native_pending': inputs.create(agent, 's0', 'p', 'native', [dict(id='q', header='Q', question='Explain')], 'test-owner') current[0] = new messages.clear() await manager.handle(new, dict(type='voice_status', voice_id='v')) if invalidation: assert 'd' not in manager.requests assert not any(event['type']=='voice_transcript' for _,event in messages) else: assert any(event['type']=='voice_end_ack' for _,event in messages) assert any(event['type']=='voice_transcript' for _,event in messages) await manager.handle(new, dict(type='voice_decision', voice_id='v', decision='send')) await manager.handle(new, dict(type='voice_decision', voice_id='v', decision='send')) assert len(adapter.inputs) == (0 if invalidation else 1) assert provider.calls == 1 finally: release.set() await manager.close() asyncio.run(run()) def test_authenticated_replacement_websocket_preserves_upload(tmp_path, monkeypatch): credential = tmp_path / 'credential.json' credential.write_text(json.dumps({'device_id':'test','device_token':'secret'})) monkeypatch.setattr(gateway, 'credential_store', DeviceCredentialStore(credential)) targets = TargetStore(tmp_path / 'targets.json') targets.set('test', 'agent:pi:s') monkeypatch.setattr(gateway, 'target_store', targets) monkeypatch.setattr(gateway, 'devices', {}) adapter = FakeAgent('pi', tmp_path) monkeypatch.setattr(gateway, 'adapters', {'pi':adapter}) provider = FakeSTT() monkeypatch.setattr(gateway.voices, 'provider', provider) # 仅替换业务上下文准备，认证、连接替换、finally 和语音状态机均走生产代码。 monkeypatch.setattr(gateway.voices, 'context_prepare', None) def receive(ws, kind): for _ in range(30): event = ws.receive_json() if event['type'] == kind: return event raise AssertionError(kind) hello = dict(type='hello',device_id='test',device_token='secret',board='synthetic',firmware='0.4.0') with TestClient(gateway.app) as client: with client.websocket_connect('/device/ws') as old: old.send_json(hello); receive(old, 'hello_ack'); receive(old, 'agent_state') old.send_json(start()); receive(old, 'voice_state') old.send_json(chunk()); assert receive(old, 'voice_ack')['next_sequence'] == 1 req = gateway.voices.requests['test'] with client.websocket_connect('/device/ws') as new: new.send_json(hello); receive(new, 'hello_ack'); receive(new, 'agent_state') assert gateway.voices.requests['test'] is req and len(req.pcm) == 3528 new.send_json(dict(type='voice_status',voice_id='v')) assert receive(new, 'voice_ack')['state'] == 'RECORDING' new.send_json(chunk()); assert receive(new, 'voice_ack')['next_sequence'] == 1 new.send_json(chunk(1)); assert receive(new, 'voice_ack')['next_sequence'] == 2 new.send_json(dict(type='voice_end',voice_id='v',total_bytes=7056)) assert receive(new, 'voice_transcript')['voice_id'] == 'v' assert adapter.inputs == [] @pytest.mark.parametrize('during_stt', [False, True, 'offline_result']) def test_replacement_resume_and_stt_delivery(during_stt): async def run(): old = SimpleNamespace(device_id='d', connection_id='old') new = SimpleNamespace(device_id='d', connection_id='new') current = [old] messages = [] ready = asyncio.Event() release = asyncio.Event() class Provider(FakeSTT): calls = 0 async def transcribe(self, data): self.calls += 1 ready.set() await release.wait() return await super().transcribe(data) provider = Provider() adapter = FakeAgent('pi', '/tmp') async def send(device, event): messages.append((device.connection_id, event)) manager = VoiceManager(provider, lambda _: 'agent:pi:s', lambda _: adapter, send, lambda d: current[0] is d, lambda: False) try: await manager.handle(old, start()) await manager.handle(old, chunk()) req = manager.requests['d'] if during_stt: await manager.handle(old, {'type':'voice_end', 'voice_id':'v', 'total_bytes':3528}) await ready.wait() manager.suspend(old, reason='replacement') if during_stt == 'offline_result': current[0] = None release.set() await asyncio.gather(*list(manager.tasks)) assert req.state == 'AWAITING_CONFIRMATION' assert not any(m['type'] == 'voice_transcript' for _,m in messages) current[0] = new await manager.handle(new, {'type':'voice_status', 'voice_id':'v'}) manager.suspend(old, reason='disconnect') assert req.state == ('AWAITING_CONFIRMATION' if during_stt == 'offline_result' else 'TRANSCRIBING' if during_stt else 'RECORDING') assert req.connection == 'new' if not during_stt: await manager.handle(new, chunk()) assert len(req.pcm) == 3528 and req.sequence == 1 await manager.handle(new, chunk(1)) await manager.handle(new, {'type':'voice_end', 'voice_id':'v', 'total_bytes':7056}) await manager.handle(new, {'type':'voice_end', 'voice_id':'v'}) release.set() await asyncio.gather(*list(manager.tasks)) assert provider.calls == 1 and req.state == 'AWAITING_CONFIRMATION' assert any(d == 'new' and m['type'] == 'voice_transcript' for d,m in messages) await manager.handle(new, {'type':'voice_status', 'voice_id':'v'}) assert messages[-1][1]['type'] == 'voice_transcript' assert adapter.inputs == [] finally: await manager.close() asyncio.run(run()) @pytest.mark.parametrize('invalidation', ['cancel', 'approval', 'target', 'generation']) def test_resume_invalidation(invalidation): async def run(): old = SimpleNamespace(device_id='d', connection_id='old') new = SimpleNamespace(device_id='d', connection_id='new') current = [old]; target = ['agent:pi:s']; approval = [False]; generation = [0] provider = FakeSTT(); adapter = FakeAgent('pi', '/tmp') async def send(d,m): pass manager = VoiceManager(provider, lambda _:target[0], lambda _:adapter, send, lambda d:current[0] is d, lambda:approval[0], lambda _:generation[0]) try: await manager.handle(old,start()); await manager.handle(old,chunk()) req = manager.requests['d'] manager.suspend(old); current[0] = new if invalidation == 'approval': approval[0] = True if invalidation == 'target': target[0] = 'agent:pi:other' if invalidation == 'generation': generation[0] += 1 await manager.handle(new, {'type':'voice_cancel' if invalidation == 'cancel' else 'voice_status','voice_id':'v'}) assert not manager.requests and not req.pcm assert provider.wav is None and not adapter.inputs count = len(manager.tasks) await manager.handle(new, {'type':'voice_end','voice_id':'v'}) await manager.handle(new, {'type':'voice_status','voice_id':'v'}) assert len(manager.tasks) == count finally: await manager.close() asyncio.run(run()) @pytest.mark.parametrize('total', [0, 3526, 3530, True, '3528']) def test_total_bytes_mismatch_never_transcribes(total): async def run(): device = SimpleNamespace(device_id='d',connection_id='c') provider = FakeSTT(); adapter = FakeAgent('pi','/tmp') async def send(d,m): pass manager = VoiceManager(provider,lambda _:'agent:pi:s',lambda _:adapter,send,lambda _:True,lambda:False) try: await manager.handle(device,start()); await manager.handle(device,chunk()) await manager.handle(device,{'type':'voice_end','voice_id':'v','total_bytes':total}) await asyncio.gather(*list(manager.tasks)) assert not manager.requests and provider.wav is None and not adapter.inputs finally: await manager.close() asyncio.run(run())