"""Tests pour le module MP3.""" import pytest import time import threading from unittest.mock import Mock, patch, MagicMock import numpy as np from voice.mp3 import MP3Player from voice.voice_types import MP3Request, PlaybackState, RMSData class TestMP3Player: """Tests pour MP3Player.""" @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") def test_initialization(self, mock_init, mock_pre_init, jaw_processor): """Test initialisation MP3Player.""" player = MP3Player(jaw_processor) assert player.jaw_processor == jaw_processor assert player.state == PlaybackState.IDLE assert player.current_file is None assert not player.loop_enabled # Vérifie init pygame mock_pre_init.assert_called_once() mock_init.assert_called_once() @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") @patch("os.path.exists") def test_play_nonexistent_file(self, mock_exists, mock_init, mock_pre_init, jaw_processor): """Test lecture fichier inexistant.""" mock_exists.return_value = False player = MP3Player(jaw_processor) request = MP3Request(file="/path/to/nonexistent.mp3") result = player.play(request) assert result is False assert player.state == PlaybackState.IDLE @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") @patch("os.path.exists") @patch("pydub.AudioSegment.from_mp3") def test_play_success( self, mock_from_mp3, mock_exists, mock_init, mock_pre_init, jaw_processor ): """Test lecture réussie.""" mock_exists.return_value = True # Mock AudioSegment mock_segment = Mock() mock_segment.__len__ = Mock(return_value=1000) # 1 seconde mock_segment.frame_rate = 22050 mock_segment.channels = 1 mock_segment.sample_width = 2 mock_segment.raw_data = b"\x00" * 1000 mock_from_mp3.return_value = mock_segment player = MP3Player(jaw_processor) request = MP3Request(file="/path/to/test.mp3") with patch.object(player, "_play_loop"): result = player.play(request) assert result is True assert player.state == PlaybackState.PLAYING assert player.current_file == "/path/to/test.mp3" @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") @patch("pygame.mixer.music") def test_pause_resume(self, mock_music, mock_init, mock_pre_init, jaw_processor): """Test pause/reprise.""" player = MP3Player(jaw_processor) player.state = PlaybackState.PLAYING # Test pause player.pause() assert player.state == PlaybackState.PAUSED mock_music.pause.assert_called_once() # Test reprise player.resume() assert player.state == PlaybackState.PLAYING mock_music.unpause.assert_called_once() @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") @patch("pygame.mixer.music") def test_stop(self, mock_music, mock_init, mock_pre_init, jaw_processor): """Test arrêt.""" player = MP3Player(jaw_processor) player.state = PlaybackState.PLAYING player.current_file = "/test.mp3" player.stop() assert player.state == PlaybackState.IDLE assert player.current_file is None mock_music.stop.assert_called_once() @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") def test_rms_callback(self, mock_init, mock_pre_init, jaw_processor): """Test callback RMS.""" player = MP3Player(jaw_processor) callback_data = [] def test_callback(rms_data): callback_data.append(rms_data) player.set_rms_callback(test_callback) # Simule données RMS rms_data = RMSData(ts_ms=1000, rms=0.5) player.rms_callback(rms_data) assert len(callback_data) == 1 assert callback_data[0] == rms_data @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") def test_segment_to_numpy_mono(self, mock_init, mock_pre_init, jaw_processor): """Test conversion AudioSegment mono vers numpy.""" player = MP3Player(jaw_processor) # Mock AudioSegment mono 16-bit mock_segment = Mock() mock_segment.channels = 1 mock_segment.sample_width = 2 mock_segment.raw_data = (np.array([100, -100, 200, -200], dtype=np.int16)).tobytes() audio_array = player._segment_to_numpy(mock_segment) assert isinstance(audio_array, np.ndarray) assert audio_array.dtype == np.float32 assert len(audio_array) == 4 # Vérifie normalisation [-1, 1] assert np.all(audio_array >= -1.0) and np.all(audio_array <= 1.0) @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") def test_segment_to_numpy_stereo(self, mock_init, mock_pre_init, jaw_processor): """Test conversion AudioSegment stéréo vers mono.""" player = MP3Player(jaw_processor) # Mock AudioSegment stéréo mock_segment = Mock() mock_segment.channels = 2 mock_segment.sample_width = 2 mock_segment.raw_data = b"\x00\x01\x00\x02" # 4 bytes = 2 samples stéréo # Mock set_channels pour conversion mono mock_mono = Mock() mock_mono.channels = 1 mock_mono.sample_width = 2 mock_mono.raw_data = b"\x00\x01" # 2 bytes = 1 sample mono mock_segment.set_channels.return_value = mock_mono audio_array = player._segment_to_numpy(mock_segment) # Vérifie que set_channels(1) a été appelé mock_segment.set_channels.assert_called_once_with(1) assert isinstance(audio_array, np.ndarray) @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") def test_generate_rms_stream_timing(self, mock_init, mock_pre_init, jaw_processor): """Test timing génération RMS.""" player = MP3Player(jaw_processor) # Mock processeur jaw sync mock_rms_data = [ RMSData(ts_ms=0, rms=0.1), RMSData(ts_ms=20, rms=0.2), RMSData(ts_ms=40, rms=0.3), ] with patch.object(jaw_processor, "process_audio_frames", return_value=mock_rms_data): callback_times = [] def timing_callback(rms_data): callback_times.append(time.time()) player.set_rms_callback(timing_callback) # Audio test court audio_data = np.array([0.1, 0.2, 0.3]) start_time = time.time() player._generate_rms_stream(audio_data, 22050) # Vérifie que les callbacks ont été appelés assert len(callback_times) == len(mock_rms_data) @patch("pygame.mixer.pre_init") @patch("pygame.mixer.init") def test_play_loop_with_stop_event(self, mock_init, mock_pre_init, jaw_processor): """Test boucle lecture avec événement stop.""" player = MP3Player(jaw_processor) # Mock AudioSegment mock_segment = Mock() mock_segment.__len__ = Mock(return_value=100) mock_segment.frame_rate = 22050 with patch("pydub.AudioSegment.from_mp3", return_value=mock_segment): with patch.object(player, "_play_segment") as mock_play: # Déclenche stop après court délai def stop_after_delay(): time.sleep(0.1) player._stop_event.set() threading.Thread(target=stop_after_delay, daemon=True).start() player._play_loop("/test.mp3") # Vérifie que _play_segment a été appelé au moins une fois mock_play.assert_called()