"""Tests pour le service principal.""" import pytest from unittest.mock import Mock, patch, MagicMock import threading import time from voice.service import VoiceService from voice.voice_types import TTSRequest, MP3Request, JawConfig, PlaybackState class TestVoiceService: """Tests pour VoiceService.""" @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_initialization(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test initialisation service.""" service = VoiceService() assert service.jaw_config is not None assert service.jaw_processor is not None assert service.state is not None assert service.capabilities is not None assert not service._running @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_start_success(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test démarrage réussi.""" # Mock MQTT connect mock_mqtt_instance = Mock() mock_mqtt_instance.connect.return_value = True mock_mqtt.return_value = mock_mqtt_instance service = VoiceService() result = service.start() assert result is True assert service._running is True mock_mqtt_instance.connect.assert_called_once() mock_mqtt_instance.publish_capabilities.assert_called_once() mock_mqtt_instance.publish_state.assert_called_once() @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_start_mqtt_failure(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test échec démarrage MQTT.""" # Mock MQTT connect failure mock_mqtt_instance = Mock() mock_mqtt_instance.connect.return_value = False mock_mqtt.return_value = mock_mqtt_instance service = VoiceService() result = service.start() assert result is False assert service._running is False @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_stop(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test arrêt service.""" mock_mp3_instance = Mock() mock_mqtt_instance = Mock() mock_mp3.return_value = mock_mp3_instance mock_mqtt.return_value = mock_mqtt_instance service = VoiceService() service._running = True service.stop() assert service._running is False mock_mp3_instance.stop.assert_called_once() mock_mqtt_instance.disconnect.assert_called_once() @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_handle_tts_request(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test gestion requête TTS.""" # Setup mocks mock_tts_instance = Mock() mock_tts_instance.synthesize_with_phonemes.return_value = ( b"fake wav data", ["s", "a", "l", "u", "t"], 500, ) mock_tts_instance.generate_visemes.return_value = [] mock_tts.return_value = mock_tts_instance mock_audio_instance = Mock() mock_audio_instance.play_wav_data.return_value = True mock_audio.return_value = mock_audio_instance mock_mp3_instance = Mock() mock_mp3_instance.get_state.return_value = PlaybackState.IDLE mock_mp3.return_value = mock_mp3_instance mock_mqtt_instance = Mock() mock_mqtt.return_value = mock_mqtt_instance service = VoiceService() request = TTSRequest(text="Salut !") service.handle_tts_request(request) # Vérifie synthèse TTS mock_tts_instance.synthesize_with_phonemes.assert_called_once_with(request) # Vérifie lecture audio mock_audio_instance.play_wav_data.assert_called_once() # Vérifie mise à jour état assert mock_mqtt_instance.publish_state.call_count >= 2 # Début et fin @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_handle_mp3_play(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test gestion lecture MP3.""" mock_mp3_instance = Mock() mock_mp3_instance.play.return_value = True mock_mp3.return_value = mock_mp3_instance mock_mqtt_instance = Mock() mock_mqtt.return_value = mock_mqtt_instance service = VoiceService() request = MP3Request(file="/test.mp3") service.handle_mp3_play(request) mock_mp3_instance.play.assert_called_once_with(request) mock_mqtt_instance.publish_state.assert_called() @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_handle_mp3_controls(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test contrôles MP3 (stop, pause, resume).""" mock_mp3_instance = Mock() mock_mp3.return_value = mock_mp3_instance mock_mqtt_instance = Mock() mock_mqtt.return_value = mock_mqtt_instance service = VoiceService() # Test stop service.handle_mp3_stop() mock_mp3_instance.stop.assert_called() # Test pause service.handle_mp3_pause() mock_mp3_instance.pause.assert_called() # Test resume service.handle_mp3_resume() mock_mp3_instance.resume.assert_called() @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_handle_jaw_config(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test mise à jour configuration jaw sync.""" service = VoiceService() new_config = JawConfig(gain=2.0, smoothing=0.5, latency_ms=200) service.handle_jaw_config(new_config) assert service.jaw_config.gain == 2.0 assert service.jaw_config.smoothing == 0.5 assert service.jaw_config.latency_ms == 200 @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_rms_data_handling(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test gestion données RMS.""" mock_mqtt_instance = Mock() mock_mqtt.return_value = mock_mqtt_instance service = VoiceService() from voice.voice_types import RMSData rms_data = RMSData(ts_ms=1000, rms=0.5) service.handle_rms_data(rms_data) mock_mqtt_instance.publish_rms.assert_called_once_with(rms_data.to_dict()) @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_get_status(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test récupération statut.""" mock_mqtt_instance = Mock() mock_mqtt_instance.is_connected.return_value = True mock_mqtt.return_value = mock_mqtt_instance mock_mp3_instance = Mock() mock_mp3_instance.get_state.return_value = PlaybackState.PLAYING mock_mp3.return_value = mock_mp3_instance mock_audio_instance = Mock() mock_audio_instance.audio_system = "pulse" mock_audio.return_value = mock_audio_instance service = VoiceService() service._running = True status = service.get_status() assert status["running"] is True assert status["mqtt_connected"] is True assert status["mp3_state"] == "playing" assert status["audio_system"] == "pulse" assert "state" in status assert "capabilities" in status assert "jaw_config" in status @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_threading_safety(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test sécurité threading.""" service = VoiceService() # Test accès concurrent à l'état def update_state(): for i in range(10): service.state.playing = not service.state.playing time.sleep(0.001) threads = [] for i in range(3): thread = threading.Thread(target=update_state) threads.append(thread) thread.start() for thread in threads: thread.join() # Le test passe s'il n'y a pas de race condition assert True @patch("voice.service.MQTTHandler") @patch("voice.service.TTSEngine") @patch("voice.service.MP3Player") @patch("voice.service.AudioOutput") def test_callback_setup(self, mock_audio, mock_mp3, mock_tts, mock_mqtt): """Test configuration callbacks.""" mock_mqtt_instance = Mock() mock_mqtt.return_value = mock_mqtt_instance mock_mp3_instance = Mock() mock_mp3.return_value = mock_mp3_instance service = VoiceService() # Vérifie que les callbacks MQTT sont configurés expected_callbacks = [ ("tts", "handle_tts_request"), ("mp3_play", "handle_mp3_play"), ("mp3_stop", "handle_mp3_stop"), ("mp3_pause", "handle_mp3_pause"), ("mp3_resume", "handle_mp3_resume"), ("jaw_config", "handle_jaw_config"), ] for topic, method_name in expected_callbacks: mock_mqtt_instance.set_callback.assert_any_call(topic, getattr(service, method_name)) # Vérifie callback RMS du MP3Player mock_mp3_instance.set_rms_callback.assert_called_once_with(service.handle_rms_data)