""" Tests d'intégration pour le service ASR complet """ import asyncio import json import pytest import tempfile from pathlib import Path from unittest.mock import Mock, patch, AsyncMock from asr.service import ASRService class TestASRIntegration: """Tests d'intégration complets.""" @pytest.fixture def temp_wav_file(self): """Crée un fichier WAV temporaire pour les tests.""" # En-tête WAV minimal pour un fichier 16kHz mono wav_header = bytes( [ # RIFF chunk 0x52, 0x49, 0x46, 0x46, # "RIFF" 0x24, 0x08, 0x00, 0x00, # Taille fichier - 8 0x57, 0x41, 0x56, 0x45, # "WAVE" # fmt chunk 0x66, 0x6D, 0x74, 0x20, # "fmt " 0x10, 0x00, 0x00, 0x00, # Taille chunk fmt 0x01, 0x00, # Format PCM 0x01, 0x00, # 1 canal 0x80, 0x3E, 0x00, 0x00, # 16000 Hz 0x00, 0x7D, 0x00, 0x00, # Byte rate 0x02, 0x00, # Block align 0x10, 0x00, # 16 bits par échantillon # data chunk 0x64, 0x61, 0x74, 0x61, # "data" 0x00, 0x08, 0x00, 0x00, # Taille data ] ) # Données audio simulées (2048 échantillons de silence) audio_data = b"\x00\x00" * 1024 with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: f.write(wav_header + audio_data) return Path(f.name) @pytest.fixture def mock_complete_environment(self): """Mock complet de l'environnement ASR.""" mocks = {} # Mock MQTT mocks["mqtt_client"] = Mock() mocks["mqtt_client"].connect.return_value = 0 # Mock Vosk mocks["vosk_model"] = Mock() mocks["vosk_recognizer"] = Mock() mocks["vosk_recognizer"].AcceptWaveform.return_value = True mocks["vosk_recognizer"].Result.return_value = json.dumps( {"text": "bonjour skull"} ) mocks["vosk_recognizer"].FinalResult.return_value = json.dumps( {"text": "bonjour skull"} ) # Mock PyAudio mocks["pyaudio"] = Mock() mocks["pyaudio_instance"] = Mock() mocks["audio_stream"] = Mock() mocks["pyaudio"].PyAudio.return_value = mocks["pyaudio_instance"] mocks["pyaudio_instance"].get_default_input_device_info.return_value = { "name": "Test Mic" } mocks["pyaudio_instance"].open.return_value = mocks["audio_stream"] mocks["pyaudio"].paInt16 = 8 mocks["pyaudio"].paContinue = 0 return mocks @pytest.mark.asyncio async def test_complete_workflow(self, mock_complete_environment): """Test le workflow complet de reconnaissance.""" mocks = mock_complete_environment with patch("asr.service.mqtt.Client", return_value=mocks["mqtt_client"]): with patch("asr.audio_capture.pyaudio", mocks["pyaudio"]): with patch("asr.vosk_engine.vosk") as mock_vosk: with patch.object(Path, "exists", return_value=True): # Configuration du mock Vosk mock_vosk.Model.return_value = mocks["vosk_model"] mock_vosk.KaldiRecognizer.return_value = mocks[ "vosk_recognizer" ] # Créer et démarrer le service service = ASRService() # Mock des méthodes async pour éviter l'attente infinie service._run_processing_loop = AsyncMock() await service.start() # Vérifications assert service.vosk_engine is not None assert service.audio_capture.is_running is True mocks["mqtt_client"].connect.assert_called_once() # Simuler une séquence VAD complète service._handle_vad_message({"active": True}) assert service.vad_active is True assert service.session_active is True # Simuler fin de parole service._handle_vad_message({"end_of_speech": True}) # Nettoyer await service.stop() @pytest.mark.asyncio async def test_config_update_workflow(self, mock_complete_environment): """Test la mise à jour de configuration à chaud.""" mocks = mock_complete_environment with patch("asr.service.mqtt.Client", return_value=mocks["mqtt_client"]): with patch("asr.audio_capture.pyaudio", mocks["pyaudio"]): with patch("asr.vosk_engine.vosk") as mock_vosk: with patch.object(Path, "exists", return_value=True): mock_vosk.Model.return_value = mocks["vosk_model"] mock_vosk.KaldiRecognizer.return_value = mocks[ "vosk_recognizer" ] service = ASRService() service._run_processing_loop = AsyncMock() await service.start() # Test mise à jour EOS original_eos = service.config.eos_ms service._handle_config_message({"eos_ms": 1500}) assert service.config.eos_ms != original_eos assert service.config.eos_ms == 1500 # Test mise à jour langue (doit recharger le moteur) with patch.object( service, "_reload_engine", new=AsyncMock() ) as mock_reload: service._handle_config_message({"lang": "en"}) assert service.config.lang == "en" mock_reload.assert_called_once() await service.stop() @pytest.mark.asyncio async def test_silence_handling(self, mock_complete_environment): """Test que le silence ne produit pas de transcription.""" mocks = mock_complete_environment # Configurer le mock pour retourner du texte vide mocks["vosk_recognizer"].FinalResult.return_value = json.dumps({"text": ""}) with patch("asr.service.mqtt.Client", return_value=mocks["mqtt_client"]): with patch("asr.audio_capture.pyaudio", mocks["pyaudio"]): with patch("asr.vosk_engine.vosk") as mock_vosk: with patch.object(Path, "exists", return_value=True): mock_vosk.Model.return_value = mocks["vosk_model"] mock_vosk.KaldiRecognizer.return_value = mocks[ "vosk_recognizer" ] service = ASRService() service._run_processing_loop = AsyncMock() await service.start() # Simuler une session avec silence service._start_asr_session() await service._finalize_transcription() # Vérifier qu'aucun texte n'a été publié publish_calls = [ call for call in mocks["mqtt_client"].publish.call_args_list if call[0][0] == "asr/text" ] assert len(publish_calls) == 0 await service.stop() @pytest.mark.asyncio async def test_capabilities_retained(self, mock_complete_environment): """Test que les capacités sont publiées en retained.""" mocks = mock_complete_environment with patch("asr.service.mqtt.Client", return_value=mocks["mqtt_client"]): with patch("asr.audio_capture.pyaudio", mocks["pyaudio"]): with patch("asr.vosk_engine.vosk") as mock_vosk: with patch.object(Path, "exists", return_value=True): mock_vosk.Model.return_value = mocks["vosk_model"] mock_vosk.KaldiRecognizer.return_value = mocks[ "vosk_recognizer" ] service = ASRService() service._run_processing_loop = AsyncMock() await service.start() # Vérifier la publication des capacités capabilities_calls = [ call for call in mocks["mqtt_client"].publish.call_args_list if call[0][0] == "asr/capabilities" ] assert len(capabilities_calls) > 0 # Vérifier que retain=True for call in capabilities_calls: assert call[1].get("retain") is True # Vérifier le contenu payload = json.loads(capabilities_calls[0][0][1]) assert payload["engine"] == "vosk" assert payload["lang"] == "fr" assert payload["sample_rate"] == 16000 await service.stop() @pytest.mark.asyncio async def test_error_recovery(self, mock_complete_environment): """Test la récupération d'erreurs.""" mocks = mock_complete_environment with patch("asr.service.mqtt.Client", return_value=mocks["mqtt_client"]): with patch("asr.audio_capture.pyaudio", mocks["pyaudio"]): with patch("asr.vosk_engine.vosk") as mock_vosk: with patch.object(Path, "exists", return_value=True): mock_vosk.Model.return_value = mocks["vosk_model"] mock_vosk.KaldiRecognizer.return_value = mocks[ "vosk_recognizer" ] service = ASRService() service._run_processing_loop = AsyncMock() await service.start() # Simuler une erreur dans le traitement Vosk mocks["vosk_recognizer"].FinalResult.side_effect = Exception( "Test error" ) # Le service doit survivre à l'erreur service._start_asr_session() await service._finalize_transcription() # Ne doit pas lever d'exception assert ( service.session_active is False ) # Session fermée après erreur await service.stop()