"""Tests for sound normalizer service.""" import tempfile from pathlib import Path from unittest.mock import AsyncMock, Mock, patch import pytest from sqlmodel.ext.asyncio.session import AsyncSession from app.models.sound import Sound from app.services.sound_normalizer import SoundNormalizerService class TestSoundNormalizerService: """Test sound normalizer service.""" @pytest.fixture def mock_session(self): """Create a mock session.""" return Mock(spec=AsyncSession) @pytest.fixture def normalizer_service(self, mock_session): """Create a normalizer service with mock session.""" with patch("app.services.sound_normalizer.settings") as mock_settings: mock_settings.NORMALIZED_AUDIO_FORMAT = "mp3" mock_settings.NORMALIZED_AUDIO_BITRATE = "256k" mock_settings.NORMALIZED_AUDIO_PASSES = 2 return SoundNormalizerService(mock_session) def test_init(self, normalizer_service): """Test normalizer service initialization.""" assert normalizer_service.session is not None assert normalizer_service.sound_repo is not None assert normalizer_service.output_format == "mp3" assert normalizer_service.output_bitrate == "256k" assert normalizer_service.passes == 2 assert len(normalizer_service.type_directories) == 3 assert "SDB" in normalizer_service.type_directories assert "TTS" in normalizer_service.type_directories assert "EXT" in normalizer_service.type_directories def test_get_normalized_path(self, normalizer_service): """Test normalized path generation.""" sound = Sound( id=1, type="SDB", name="Test Sound", filename="test_audio.mp3", duration=5000, size=1024, hash="test_hash", ) normalized_path = normalizer_service._get_normalized_path(sound) assert "sounds/normalized/soundboard" in str(normalized_path) assert normalized_path.name == "test_audio.mp3" def test_get_original_path(self, normalizer_service): """Test original path generation.""" sound = Sound( id=1, type="SDB", name="Test Sound", filename="test_audio.wav", duration=5000, size=1024, hash="test_hash", ) original_path = normalizer_service._get_original_path(sound) assert "sounds/originals/soundboard" in str(original_path) assert original_path.name == "test_audio.wav" def test_get_file_hash(self, normalizer_service): """Test file hash calculation.""" # Create a temporary file with tempfile.NamedTemporaryFile(mode="w", delete=False) as f: f.write("test content for hash") temp_path = Path(f.name) try: from app.utils.audio import get_file_hash hash_value = get_file_hash(temp_path) assert len(hash_value) == 64 # SHA-256 hash length assert isinstance(hash_value, str) finally: temp_path.unlink() def test_get_file_size(self, normalizer_service): """Test file size calculation.""" # Create a temporary file with tempfile.NamedTemporaryFile(mode="w", delete=False) as f: f.write("test content for size") temp_path = Path(f.name) try: from app.utils.audio import get_file_size size = get_file_size(temp_path) assert size > 0 assert isinstance(size, int) finally: temp_path.unlink() @patch("app.utils.audio.ffmpeg.probe") def test_get_audio_duration_success(self, mock_probe, normalizer_service): """Test successful audio duration extraction.""" mock_probe.return_value = {"format": {"duration": "123.456"}} temp_path = Path("/fake/path/test.mp3") from app.utils.audio import get_audio_duration duration = get_audio_duration(temp_path) assert duration == 123456 # 123.456 seconds * 1000 = 123456 ms mock_probe.assert_called_once_with(str(temp_path)) @patch("app.utils.audio.ffmpeg.probe") def test_get_audio_duration_failure(self, mock_probe, normalizer_service): """Test audio duration extraction failure.""" mock_probe.side_effect = Exception("FFmpeg error") temp_path = Path("/fake/path/test.mp3") from app.utils.audio import get_audio_duration duration = get_audio_duration(temp_path) assert duration == 0 mock_probe.assert_called_once_with(str(temp_path)) @pytest.mark.asyncio async def test_normalize_sound_already_normalized(self, normalizer_service): """Test normalizing a sound that's already normalized.""" sound = Sound( id=1, type="SDB", name="Test Sound", filename="test.mp3", duration=5000, size=1024, hash="test_hash", is_normalized=True, ) result = await normalizer_service.normalize_sound(sound) assert result["status"] == "skipped" assert result["reason"] == "already normalized" assert result["filename"] == "test.mp3" assert result["id"] == 1 @pytest.mark.asyncio async def test_normalize_sound_force_already_normalized(self, normalizer_service): """Test force normalizing a sound that's already normalized.""" sound = Sound( id=1, type="SDB", name="Test Sound", filename="test.mp3", duration=5000, size=1024, hash="test_hash", is_normalized=True, ) # Mock file operations and ffmpeg with ( patch.object(normalizer_service, "_get_original_path") as mock_orig_path, patch.object(normalizer_service, "_get_normalized_path") as mock_norm_path, patch.object( normalizer_service, "_normalize_audio_two_pass", ) as mock_normalize, patch( "app.services.sound_normalizer.get_audio_duration", return_value=6000, ), patch("app.services.sound_normalizer.get_file_size", return_value=2048), patch( "app.services.sound_normalizer.get_file_hash", return_value="new_hash", ), ): # Setup path mocks mock_orig_path.return_value = Path("/fake/original.mp3") mock_norm_path.return_value = Path("/fake/normalized.mp3") # Mock file existence with patch("pathlib.Path.exists", return_value=True): # Mock repository update normalizer_service.sound_repo.update = AsyncMock() result = await normalizer_service.normalize_sound(sound, force=True) assert result["status"] == "normalized" assert result["filename"] == "test.mp3" assert result["normalized_duration"] == 6000 assert result["normalized_size"] == 2048 assert result["normalized_hash"] == "new_hash" # Verify update was called normalizer_service.sound_repo.update.assert_called_once() @pytest.mark.asyncio async def test_normalize_sound_file_not_found(self, normalizer_service): """Test normalizing a sound where original file doesn't exist.""" sound = Sound( id=1, type="SDB", name="Test Sound", filename="missing.mp3", duration=5000, size=1024, hash="test_hash", is_normalized=False, ) with patch.object(normalizer_service, "_get_original_path") as mock_path: mock_path.return_value = Path("/fake/missing.mp3") # Mock file doesn't exist with patch("pathlib.Path.exists", return_value=False): result = await normalizer_service.normalize_sound(sound) assert result["status"] == "error" assert "Original file not found" in result["error"] assert result["filename"] == "missing.mp3" @pytest.mark.asyncio async def test_normalize_sound_one_pass(self, normalizer_service): """Test normalizing a sound using one-pass method.""" sound = Sound( id=1, type="SDB", name="Test Sound", filename="test.mp3", duration=5000, size=1024, hash="test_hash", is_normalized=False, ) with ( patch.object(normalizer_service, "_get_original_path") as mock_orig_path, patch.object(normalizer_service, "_get_normalized_path") as mock_norm_path, patch.object( normalizer_service, "_normalize_audio_one_pass", ) as mock_normalize, patch( "app.services.sound_normalizer.get_audio_duration", return_value=5500, ), patch("app.services.sound_normalizer.get_file_size", return_value=1500), patch( "app.services.sound_normalizer.get_file_hash", return_value="norm_hash", ), ): # Setup path mocks mock_orig_path.return_value = Path("/fake/original.mp3") mock_norm_path.return_value = Path("/fake/normalized.mp3") # Mock file existence with patch("pathlib.Path.exists", return_value=True): # Mock repository update normalizer_service.sound_repo.update = AsyncMock() result = await normalizer_service.normalize_sound(sound, one_pass=True) assert result["status"] == "normalized" assert result["normalized_duration"] == 5500 assert result["normalized_size"] == 1500 assert result["normalized_hash"] == "norm_hash" # Verify one-pass was used mock_normalize.assert_called_once() @pytest.mark.asyncio async def test_normalize_sound_normalization_error(self, normalizer_service): """Test handling normalization errors.""" sound = Sound( id=1, type="SDB", name="Test Sound", filename="test.mp3", duration=5000, size=1024, hash="test_hash", is_normalized=False, ) with ( patch.object(normalizer_service, "_get_original_path") as mock_orig_path, patch.object(normalizer_service, "_get_normalized_path") as mock_norm_path, ): # Setup path mocks mock_orig_path.return_value = Path("/fake/original.mp3") mock_norm_path.return_value = Path("/fake/normalized.mp3") # Mock file existence but normalization fails with ( patch("pathlib.Path.exists", return_value=True), patch.object( normalizer_service, "_normalize_audio_two_pass", ) as mock_normalize, ): mock_normalize.side_effect = Exception("Normalization failed") result = await normalizer_service.normalize_sound(sound) assert result["status"] == "error" assert "Normalization failed" in result["error"] assert result["filename"] == "test.mp3" @pytest.mark.asyncio async def test_normalize_all_sounds(self, normalizer_service): """Test normalizing all unnormalized sounds.""" sounds = [ Sound( id=1, type="SDB", name="Sound 1", filename="sound1.mp3", duration=5000, size=1024, hash="hash1", is_normalized=False, ), Sound( id=2, type="TTS", name="Sound 2", filename="sound2.wav", duration=3000, size=512, hash="hash2", is_normalized=False, ), ] # Mock repository calls normalizer_service.sound_repo.get_unnormalized_sounds = AsyncMock( return_value=sounds, ) # Mock individual normalization with patch.object(normalizer_service, "normalize_sound") as mock_normalize: mock_normalize.side_effect = [ { "filename": "sound1.mp3", "status": "normalized", "reason": None, "original_path": "/fake/sound1.mp3", "normalized_path": "/fake/sound1_normalized.mp3", "normalized_filename": "sound1_normalized.mp3", "normalized_duration": 5000, "normalized_size": 1024, "normalized_hash": "norm_hash1", "id": 1, "error": None, }, { "filename": "sound2.wav", "status": "normalized", "reason": None, "original_path": "/fake/sound2.wav", "normalized_path": "/fake/sound2_normalized.mp3", "normalized_filename": "sound2_normalized.mp3", "normalized_duration": 3000, "normalized_size": 512, "normalized_hash": "norm_hash2", "id": 2, "error": None, }, ] results = await normalizer_service.normalize_all_sounds() assert results["processed"] == 2 assert results["normalized"] == 2 assert results["skipped"] == 0 assert results["errors"] == 0 assert len(results["files"]) == 2 @pytest.mark.asyncio async def test_normalize_sounds_by_type(self, normalizer_service): """Test normalizing sounds by type.""" sdb_sounds = [ Sound( id=1, type="SDB", name="SDB Sound", filename="sdb.mp3", duration=5000, size=1024, hash="sdb_hash", is_normalized=False, ), ] # Mock repository calls normalizer_service.sound_repo.get_unnormalized_sounds_by_type = AsyncMock( return_value=sdb_sounds, ) # Mock individual normalization with patch.object(normalizer_service, "normalize_sound") as mock_normalize: mock_normalize.return_value = { "filename": "sdb.mp3", "status": "normalized", "reason": None, "original_path": "/fake/sdb.mp3", "normalized_path": "/fake/sdb_normalized.mp3", "normalized_filename": "sdb_normalized.mp3", "normalized_duration": 5000, "normalized_size": 1024, "normalized_hash": "sdb_norm_hash", "id": 1, "error": None, } results = await normalizer_service.normalize_sounds_by_type("SDB") assert results["processed"] == 1 assert results["normalized"] == 1 assert results["skipped"] == 0 assert results["errors"] == 0 assert len(results["files"]) == 1 # Verify correct repository method was called normalizer_service.sound_repo.get_unnormalized_sounds_by_type.assert_called_once_with( "SDB", ) @pytest.mark.asyncio async def test_normalize_sounds_with_errors(self, normalizer_service): """Test normalizing sounds with some errors.""" sounds = [ Sound( id=1, type="SDB", name="Good Sound", filename="good.mp3", duration=5000, size=1024, hash="good_hash", is_normalized=False, ), Sound( id=2, type="SDB", name="Bad Sound", filename="bad.mp3", duration=3000, size=512, hash="bad_hash", is_normalized=False, ), ] # Mock repository calls normalizer_service.sound_repo.get_unnormalized_sounds = AsyncMock( return_value=sounds, ) # Mock individual normalization with one success and one error with patch.object(normalizer_service, "normalize_sound") as mock_normalize: mock_normalize.side_effect = [ { "filename": "good.mp3", "status": "normalized", "reason": None, "original_path": "/fake/good.mp3", "normalized_path": "/fake/good_normalized.mp3", "normalized_filename": "good_normalized.mp3", "normalized_duration": 5000, "normalized_size": 1024, "normalized_hash": "good_norm_hash", "id": 1, "error": None, }, { "filename": "bad.mp3", "status": "error", "reason": None, "original_path": "/fake/bad.mp3", "normalized_path": None, "normalized_filename": None, "normalized_duration": None, "normalized_size": None, "normalized_hash": None, "id": 2, "error": "File processing failed", }, ] results = await normalizer_service.normalize_all_sounds() assert results["processed"] == 2 assert results["normalized"] == 1 assert results["skipped"] == 0 assert results["errors"] == 1 assert len(results["files"]) == 2 # Check error file details error_file = next(f for f in results["files"] if f["status"] == "error") assert error_file["filename"] == "bad.mp3" assert error_file["error"] == "File processing failed" @pytest.mark.asyncio @patch("app.services.sound_normalizer.ffmpeg") async def test_normalize_audio_one_pass_mp3( self, mock_ffmpeg, normalizer_service, ): """Test one-pass audio normalization for MP3.""" input_path = Path("/fake/input.wav") output_path = Path("/fake/output.mp3") # Mock ffmpeg chain mock_stream = Mock() mock_ffmpeg.input.return_value = mock_stream mock_ffmpeg.filter.return_value = mock_stream mock_ffmpeg.output.return_value = mock_stream mock_ffmpeg.overwrite_output.return_value = mock_stream await normalizer_service._normalize_audio_one_pass(input_path, output_path) # Verify ffmpeg chain was called correctly mock_ffmpeg.input.assert_called_once_with(str(input_path)) mock_ffmpeg.filter.assert_called_once_with( mock_stream, "loudnorm", I=-23, TP=-2, LRA=7, ) mock_ffmpeg.output.assert_called_once() mock_ffmpeg.run.assert_called_once() # Check output arguments include MP3 codec and bitrate output_call_args = mock_ffmpeg.output.call_args assert output_call_args[0][1] == str(output_path) # output path output_kwargs = output_call_args[1] assert output_kwargs["acodec"] == "libmp3lame" assert output_kwargs["audio_bitrate"] == "256k" @pytest.mark.asyncio @patch("app.services.sound_normalizer.ffmpeg") async def test_normalize_audio_two_pass_analysis( self, mock_ffmpeg, normalizer_service, ): """Test two-pass audio normalization analysis phase.""" input_path = Path("/fake/input.wav") output_path = Path("/fake/output.mp3") # Mock ffmpeg chain mock_stream = Mock() mock_ffmpeg.input.return_value = mock_stream mock_ffmpeg.filter.return_value = mock_stream mock_ffmpeg.output.return_value = mock_stream mock_ffmpeg.overwrite_output.return_value = mock_stream # Mock analysis output with valid JSON analysis_json = """{ "input_i": "-23.0", "input_lra": "11.0", "input_tp": "-2.0", "input_thresh": "-33.0", "target_offset": "0.0" }""" mock_ffmpeg.run.side_effect = [ (None, analysis_json.encode("utf-8")), # First pass analysis None, # Second pass normalization ] await normalizer_service._normalize_audio_two_pass(input_path, output_path) # Verify two ffmpeg runs occurred assert mock_ffmpeg.run.call_count == 2 # Verify analysis pass used print_format=json first_filter_call = mock_ffmpeg.filter.call_args_list[0] assert "print_format" in first_filter_call[1] assert first_filter_call[1]["print_format"] == "json" # Verify second pass used measured values second_filter_call = mock_ffmpeg.filter.call_args_list[1] measured_args = second_filter_call[1] assert "measured_I" in measured_args assert "measured_LRA" in measured_args assert "measured_TP" in measured_args assert "measured_thresh" in measured_args assert "offset" in measured_args