"""Continuous PCM WebSocket mode delegates turn decisions to backend VAD.""" from contextlib import contextmanager import pytest from fastapi.testclient import TestClient from voice_assistant.agent.mock import MockAgentAdapter from voice_assistant.audio_gateway.app import main from voice_assistant.stt.mock import MockSTTAdapter from voice_assistant.tts.mock import MockTTSAdapter from voice_assistant.vad.adapter import VADAdapter from voice_assistant.vad.models import SpeechStarted, SpeechEnded class VAD(VADAdapter): def __init__(self): self.script = [] self.tail = () self.received = [] self.closed = False async def process(self, audio): self.received.append(audio) return self.script.pop(0) if self.script else () async def flush(self): return self.tail async def reset(self): pass async def aclose(self): self.closed = True class STT(MockSTTAdapter): def __init__(self): super().__init__() self.received = [] async def send_audio(self, audio): await super().send_audio(audio) self.received.append(audio) async def stop_session(self): await self.emit_final('Pergunta de teste') await super().stop_session() @pytest.fixture def components(monkeypatch): detector = VAD() providers = [] def create_stt(**kwargs): stt = STT() providers.append(stt) return stt monkeypatch.setattr(main, 'create_vad_adapter', lambda **kwargs: detector) monkeypatch.setattr(main, 'create_stt_adapter', create_stt) monkeypatch.setattr(main, 'create_agent_adapter', lambda **kwargs: MockAgentAdapter()) monkeypatch.setattr(main, 'create_tts_adapter', lambda **kwargs: MockTTSAdapter()) return detector, providers @contextmanager def connect(mode='server'): with TestClient(main.app).websocket_connect('/ws/audio') as socket: socket.send_json({'type': 'session.start', 'turn_detection': mode}) ready = socket.receive_json() assert ready['type'] == 'session.ready' yield socket, ready def send_audio(socket, sequence, audio): socket.send_json({'type': 'audio.input', 'sequence_number': sequence}) socket.send_bytes(audio) def read_turn(socket): events, audio = [], [] for _ in range(20): message = socket.receive() if message.get('bytes') is not None: audio.append(message['bytes']) else: import json event = json.loads(message['text']) assert event['type'] != 'error', event events.append(event) if event['type'] == 'turn.ended': return events, audio raise AssertionError('Turn did not complete within expected events') def test_continuous_mode_receives_audio_before_first_turn(components): detector, providers = components with connect() as (socket, ready): assert ready['turn_detection'] == 'server' assert ready['max_audio_chunk_bytes'] == 65536 send_audio(socket, 0, bytes(512)) socket.send_json({'type': 'session.end'}) assert socket.receive_json()['type'] == 'session.ended' assert detector.received == [bytes(512)] assert providers == [] assert detector.closed def test_vad_opens_and_closes_turn_and_input_sequence_survives_next_turn(components): detector, providers = components detector.script = [ (SpeechStarted(0),), (SpeechEnded(400, 'silence'),), (), (SpeechStarted(900),), ] detector.tail = (SpeechEnded(1024, 'end_of_input'),) with connect() as (socket, ready): send_audio(socket, 0, bytes(512)) assert socket.receive_json()['type'] == 'turn.started' send_audio(socket, 1, bytes(512)) events, audio = read_turn(socket) assert events[-1]['turn_id'] == 1 assert any(event['type'] == 'stt.final' for event in events) assert b''.join(audio) == b'\x01\x02\x03\x04\x05\x06\x07\x08' send_audio(socket, 2, bytes(512)) send_audio(socket, 3, bytes(512)) assert socket.receive_json()['turn_id'] == 2 socket.send_json({'type': 'audio.input.end'}) events, audio = read_turn(socket) assert events[-1]['turn_id'] == 2 assert audio assert len(providers) == 2 assert len(detector.received) == 4 @pytest.mark.parametrize('sequence', [-1, True, 0.5, '0', 1, 2]) def test_bad_sequence_is_rejected_without_consuming_input(components, sequence): detector, _ = components with connect() as (socket, ready): socket.send_json({'type': 'audio.input', 'sequence_number': sequence}) error = socket.receive_json() assert error['code'] == 'AUDIO_SEQUENCE_ERROR' assert error['details']['expected_sequence_number'] == 0 send_audio(socket, 0, bytes(512)) socket.send_json({'type': 'session.end'}) assert socket.receive_json()['type'] == 'session.ended' assert len(detector.received) == 1 @pytest.mark.parametrize(('audio', 'code'), [ (b'', 'AUDIO_SEQUENCE_ERROR'), (bytes(3), 'AUDIO_SEQUENCE_ERROR'), (bytes(6), 'MESSAGE_TOO_LARGE'), ]) def test_bad_pcm_does_not_advance_global_sequence(components, monkeypatch, audio, code): detector, _ = components config = main.app_config.model_copy(update={'voice_input': {'max_input_chunk_bytes': 4}}) monkeypatch.setattr(main, 'app_config', config) with connect() as (socket, ready): assert ready['max_audio_chunk_bytes'] == 4 send_audio(socket, 0, audio) assert socket.receive_json()['code'] == code send_audio(socket, 0, bytes(4)) socket.send_json({'type': 'session.end'}) assert socket.receive_json()['type'] == 'session.ended' assert detector.received == [bytes(4)] def test_duplicate_metadata_does_not_overwrite_pending_pcm(components): detector, _ = components with connect() as (socket, ready): socket.send_json({'type': 'audio.input', 'sequence_number': 0}) socket.send_json({'type': 'audio.input', 'sequence_number': 1}) assert socket.receive_json()['code'] == 'AUDIO_SEQUENCE_ERROR' socket.send_bytes(bytes(512)) socket.send_json({'type': 'session.end'}) assert socket.receive_json()['type'] == 'session.ended' assert len(detector.received) == 1 def test_binary_frame_requires_metadata(components): with connect() as (socket, ready): socket.send_bytes(bytes(512)) assert socket.receive_json()['code'] == 'INVALID_AUDIO_FRAME' @pytest.mark.parametrize('event', [ {'type': 'audio.start'}, {'type': 'audio.stop', 'turn_id': 1}, {'type': 'audio.chunk', 'turn_id': 1, 'sequence_number': 0}, ]) def test_client_turn_commands_are_rejected_in_server_mode(components, event): with connect() as (socket, ready): socket.send_json(event) assert socket.receive_json()['code'] == 'TURN_CONTROLLED_BY_VAD' @pytest.mark.parametrize('event', [ {'type': 'audio.input', 'sequence_number': 0}, {'type': 'audio.input.end'}, ]) def test_manual_mode_rejects_continuous_input(components, event): with connect('client') as (socket, ready): assert ready['turn_detection'] == 'client' socket.send_json(event) assert socket.receive_json()['code'] == 'INVALID_INPUT_MODE' def test_input_end_cannot_overtake_pending_binary_frame(components): with connect() as (socket, ready): socket.send_json({'type': 'audio.input', 'sequence_number': 0}) socket.send_json({'type': 'audio.input.end'}) assert socket.receive_json()['code'] == 'INVALID_MESSAGE' def test_initialization_failure_is_reported_and_can_be_followed_by_manual_session(components, monkeypatch): def unavailable(**kwargs): raise ValueError('VAD unavailable') monkeypatch.setattr(main, 'create_vad_adapter', unavailable) with TestClient(main.app).websocket_connect('/ws/audio') as socket: socket.send_json({'type': 'session.start', 'turn_detection': 'server'}) assert socket.receive_json()['code'] == 'SESSION_START_FAILED' socket.send_json({'type': 'session.start'}) assert socket.receive_json()['type'] == 'session.ready'