quaik8/sdb2-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement sound normalization service and API endpoints

Browse Source 
- Added SoundNormalizerService for normalizing audio files with support for one-pass and two-pass normalization methods.
- Introduced API endpoints for normalizing all sounds and specific sounds by ID, including support for force normalization and handling of already normalized sounds.
- Created comprehensive test suite for the sound normalizer service and its API endpoints, covering various scenarios including success, errors, and edge cases.
- Refactored sound scanning service to utilize SHA-256 for file hashing instead of MD5 for improved security.
- Enhanced logging and error handling throughout the sound normalization process.
This commit is contained in:
JSC
2025-07-28 09:18:18 +02:00
parent 36949a1f1c
commit 0fffce53b4
8 changed files with 2031 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

613
tests/api/v1/test_sound_normalizer_endpoints.py Normal file
View File

@@ -0,0 +1,613 @@
"""Tests for sound normalizer API endpoints."""
from unittest.mock import patch
import pytest
from httpx import ASGITransport, AsyncClient
from app.models.user import User
from app.services.sound_normalizer import NormalizationResults
class TestSoundNormalizerEndpoints:
"""Test sound normalizer API endpoints."""
@pytest.mark.asyncio
async def test_normalize_all_sounds_success(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test successful normalization of all sounds."""
mock_results: NormalizationResults = {
"processed": 3,
"normalized": 2,
"skipped": 1,
"errors": 0,
"files": [
{
"filename": "test1.mp3",
"status": "normalized",
"reason": None,
"original_path": "/fake/test1.mp3",
"normalized_path": "/fake/test1_normalized.mp3",
"normalized_filename": "test1_normalized.mp3",
"normalized_duration": 5000,
"normalized_size": 1024,
"normalized_hash": "norm_hash1",
"id": 1,
"error": None,
},
{
"filename": "test2.wav",
"status": "normalized",
"reason": None,
"original_path": "/fake/test2.wav",
"normalized_path": "/fake/test2_normalized.mp3",
"normalized_filename": "test2_normalized.mp3",
"normalized_duration": 7000,
"normalized_size": 2048,
"normalized_hash": "norm_hash2",
"id": 2,
"error": None,
},
{
"filename": "test3.mp3",
"status": "skipped",
"reason": "already normalized",
"original_path": None,
"normalized_path": None,
"normalized_filename": None,
"normalized_duration": None,
"normalized_size": None,
"normalized_hash": None,
"id": 3,
"error": None,
},
],
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds"
) as mock_normalize:
mock_normalize.return_value = mock_results
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/all"
)
assert response.status_code == 200
data = response.json()
assert "message" in data
assert "Sound normalization completed" in data["message"]
assert "results" in data
results = data["results"]
assert results["processed"] == 3
assert results["normalized"] == 2
assert results["skipped"] == 1
assert results["errors"] == 0
assert len(results["files"]) == 3
@pytest.mark.asyncio
async def test_normalize_all_sounds_with_force(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test normalization with force parameter."""
mock_results: NormalizationResults = {
"processed": 1,
"normalized": 1,
"skipped": 0,
"errors": 0,
"files": [],
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds"
) as mock_normalize:
mock_normalize.return_value = mock_results
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/all", params={"force": True}
)
assert response.status_code == 200
# Verify force parameter was passed
mock_normalize.assert_called_once_with(force=True, one_pass=None)
@pytest.mark.asyncio
async def test_normalize_all_sounds_with_one_pass(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test normalization with one_pass parameter."""
mock_results: NormalizationResults = {
"processed": 1,
"normalized": 1,
"skipped": 0,
"errors": 0,
"files": [],
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds"
) as mock_normalize:
mock_normalize.return_value = mock_results
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/all", params={"one_pass": True}
)
assert response.status_code == 200
# Verify one_pass parameter was passed
mock_normalize.assert_called_once_with(force=False, one_pass=True)
@pytest.mark.asyncio
async def test_normalize_all_sounds_unauthenticated(self, client: AsyncClient):
"""Test normalizing sounds without authentication."""
response = await client.post("/api/v1/sounds/normalize/all")
assert response.status_code == 401
data = response.json()
assert "Could not validate credentials" in data["detail"]
@pytest.mark.asyncio
async def test_normalize_all_sounds_non_admin(
self,
test_app,
test_user: User,
):
"""Test normalizing sounds with non-admin user."""
from app.core.dependencies import get_current_active_user_flexible
# Override the dependency to return regular user
async def override_get_current_user():
test_user.role = "user"
return test_user
test_app.dependency_overrides[get_current_active_user_flexible] = (
override_get_current_user
)
headers = {"API-TOKEN": "test_api_token"}
async with AsyncClient(
transport=ASGITransport(app=test_app),
base_url="http://test",
) as client:
response = await client.post(
"/api/v1/sounds/normalize/all", headers=headers
)
assert response.status_code == 403
data = response.json()
assert "Only administrators can normalize sounds" in data["detail"]
# Clean up override
test_app.dependency_overrides.pop(get_current_active_user_flexible, None)
@pytest.mark.asyncio
async def test_normalize_all_sounds_service_error(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test normalization when service raises an error."""
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_all_sounds"
) as mock_normalize:
mock_normalize.side_effect = Exception("Normalization service failed")
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/all"
)
assert response.status_code == 500
data = response.json()
assert "Failed to normalize sounds" in data["detail"]
assert "Normalization service failed" in data["detail"]
@pytest.mark.asyncio
async def test_normalize_sounds_by_type_success(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test successful normalization by sound type."""
mock_results: NormalizationResults = {
"processed": 2,
"normalized": 2,
"skipped": 0,
"errors": 0,
"files": [
{
"filename": "sdb1.mp3",
"status": "normalized",
"reason": None,
"original_path": "/fake/sdb1.mp3",
"normalized_path": "/fake/sdb1_normalized.mp3",
"normalized_filename": "sdb1_normalized.mp3",
"normalized_duration": 4000,
"normalized_size": 800,
"normalized_hash": "sdb_hash1",
"id": 10,
"error": None,
},
{
"filename": "sdb2.wav",
"status": "normalized",
"reason": None,
"original_path": "/fake/sdb2.wav",
"normalized_path": "/fake/sdb2_normalized.mp3",
"normalized_filename": "sdb2_normalized.mp3",
"normalized_duration": 6000,
"normalized_size": 1200,
"normalized_hash": "sdb_hash2",
"id": 11,
"error": None,
},
],
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_sounds_by_type"
) as mock_normalize:
mock_normalize.return_value = mock_results
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/type/SDB"
)
assert response.status_code == 200
data = response.json()
assert "Normalization of SDB sounds completed" in data["message"]
assert "results" in data
results = data["results"]
assert results["processed"] == 2
assert results["normalized"] == 2
assert len(results["files"]) == 2
# Verify the service was called with correct type
mock_normalize.assert_called_once_with(
sound_type="SDB", force=False, one_pass=None
)
@pytest.mark.asyncio
async def test_normalize_sounds_by_type_invalid_type(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test normalization with invalid sound type."""
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/type/INVALID"
)
assert response.status_code == 400
data = response.json()
assert "Invalid sound type" in data["detail"]
assert "Must be one of: SDB, TTS, EXT" in data["detail"]
@pytest.mark.asyncio
async def test_normalize_sounds_by_type_with_params(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test normalization by type with force and one_pass parameters."""
mock_results: NormalizationResults = {
"processed": 1,
"normalized": 1,
"skipped": 0,
"errors": 0,
"files": [],
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_sounds_by_type"
) as mock_normalize:
mock_normalize.return_value = mock_results
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/type/TTS",
params={"force": True, "one_pass": False},
)
assert response.status_code == 200
# Verify parameters were passed correctly
mock_normalize.assert_called_once_with(
sound_type="TTS", force=True, one_pass=False
)
@pytest.mark.asyncio
async def test_normalize_sound_by_id_success(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test successful normalization of a specific sound."""
# Mock the sound
mock_sound = type(
"Sound",
(),
{
"id": 42,
"filename": "specific_sound.mp3",
"type": "SDB",
"name": "Specific Sound",
},
)()
# Mock normalization result
mock_result = {
"filename": "specific_sound.mp3",
"status": "normalized",
"reason": None,
"original_path": "/fake/specific_sound.mp3",
"normalized_path": "/fake/specific_sound_normalized.mp3",
"normalized_filename": "specific_sound_normalized.mp3",
"normalized_duration": 8000,
"normalized_size": 1600,
"normalized_hash": "specific_hash",
"id": 42,
"error": None,
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_sound"
) as mock_normalize_sound, patch(
"app.repositories.sound.SoundRepository.get_by_id"
) as mock_get_sound:
mock_get_sound.return_value = mock_sound
mock_normalize_sound.return_value = mock_result
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/42"
)
assert response.status_code == 200
data = response.json()
assert "Sound normalization normalized" in data["message"]
assert "specific_sound.mp3" in data["message"]
assert data["status"] == "normalized"
assert data["normalized_filename"] == "specific_sound_normalized.mp3"
# Verify sound was retrieved and normalized
mock_get_sound.assert_called_once_with(42)
mock_normalize_sound.assert_called_once()
@pytest.mark.asyncio
async def test_normalize_sound_by_id_not_found(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test normalization of non-existent sound."""
with patch(
"app.repositories.sound.SoundRepository.get_by_id"
) as mock_get_sound:
mock_get_sound.return_value = None
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/999"
)
assert response.status_code == 404
data = response.json()
assert "Sound with ID 999 not found" in data["detail"]
@pytest.mark.asyncio
async def test_normalize_sound_by_id_normalization_error(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test normalization when the sound normalization fails."""
# Mock the sound
mock_sound = type(
"Sound",
(),
{
"id": 42,
"filename": "error_sound.mp3",
"type": "SDB",
"name": "Error Sound",
},
)()
# Mock normalization error result
mock_result = {
"filename": "error_sound.mp3",
"status": "error",
"reason": None,
"original_path": "/fake/error_sound.mp3",
"normalized_path": None,
"normalized_filename": None,
"normalized_duration": None,
"normalized_size": None,
"normalized_hash": None,
"id": 42,
"error": "File format not supported",
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_sound"
) as mock_normalize_sound, patch(
"app.repositories.sound.SoundRepository.get_by_id"
) as mock_get_sound:
mock_get_sound.return_value = mock_sound
mock_normalize_sound.return_value = mock_result
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/42"
)
assert response.status_code == 500
data = response.json()
assert "Failed to normalize sound" in data["detail"]
assert "File format not supported" in data["detail"]
@pytest.mark.asyncio
async def test_normalize_sound_by_id_with_params(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test sound normalization with force and one_pass parameters."""
# Mock the sound
mock_sound = type(
"Sound",
(),
{
"id": 42,
"filename": "param_sound.mp3",
"type": "SDB",
"name": "Param Sound",
},
)()
# Mock normalization result
mock_result = {
"filename": "param_sound.mp3",
"status": "normalized",
"reason": None,
"original_path": "/fake/param_sound.mp3",
"normalized_path": "/fake/param_sound_normalized.mp3",
"normalized_filename": "param_sound_normalized.mp3",
"normalized_duration": 5000,
"normalized_size": 1000,
"normalized_hash": "param_hash",
"id": 42,
"error": None,
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_sound"
) as mock_normalize_sound, patch(
"app.repositories.sound.SoundRepository.get_by_id"
) as mock_get_sound:
mock_get_sound.return_value = mock_sound
mock_normalize_sound.return_value = mock_result
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/42",
params={"force": True, "one_pass": True},
)
assert response.status_code == 200
# Verify parameters were passed to normalize_sound
call_args = mock_normalize_sound.call_args
assert call_args[1]["force"] == True
assert call_args[1]["one_pass"] == True
@pytest.mark.asyncio
async def test_normalize_sound_by_id_skipped(
self,
authenticated_admin_client: AsyncClient,
admin_user: User,
):
"""Test normalization when sound is already normalized and not forced."""
# Mock the sound
mock_sound = type(
"Sound",
(),
{
"id": 42,
"filename": "already_normalized.mp3",
"type": "SDB",
"name": "Already Normalized",
},
)()
# Mock skipped result
mock_result = {
"filename": "already_normalized.mp3",
"status": "skipped",
"reason": "already normalized",
"original_path": None,
"normalized_path": None,
"normalized_filename": None,
"normalized_duration": None,
"normalized_size": None,
"normalized_hash": None,
"id": 42,
"error": None,
}
with patch(
"app.services.sound_normalizer.SoundNormalizerService.normalize_sound"
) as mock_normalize_sound, patch(
"app.repositories.sound.SoundRepository.get_by_id"
) as mock_get_sound:
mock_get_sound.return_value = mock_sound
mock_normalize_sound.return_value = mock_result
response = await authenticated_admin_client.post(
"/api/v1/sounds/normalize/42"
)
assert response.status_code == 200
data = response.json()
assert "Sound normalization skipped" in data["message"]
assert data["status"] == "skipped"
assert data["reason"] == "already normalized"
@pytest.mark.asyncio
async def test_normalize_sound_by_id_unauthenticated(self, client: AsyncClient):
"""Test normalizing a specific sound without authentication."""
response = await client.post("/api/v1/sounds/normalize/42")
assert response.status_code == 401
data = response.json()
assert "Could not validate credentials" in data["detail"]
@pytest.mark.asyncio
async def test_normalize_sound_by_id_non_admin(
self,
test_app,
test_user: User,
):
"""Test normalizing a specific sound with non-admin user."""
from app.core.dependencies import get_current_active_user_flexible
# Override the dependency to return regular user
async def override_get_current_user():
test_user.role = "user"
return test_user
test_app.dependency_overrides[get_current_active_user_flexible] = (
override_get_current_user
)
headers = {"API-TOKEN": "test_api_token"}
async with AsyncClient(
transport=ASGITransport(app=test_app),
base_url="http://test",
) as client:
response = await client.post(
"/api/v1/sounds/normalize/42", headers=headers
)
assert response.status_code == 403
data = response.json()
assert "Only administrators can normalize sounds" in data["detail"]
# Clean up override
test_app.dependency_overrides.pop(get_current_active_user_flexible, None)

559
tests/services/test_sound_normalizer.py Normal file
View File

@@ -0,0 +1,559 @@
"""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 "test_audio.mp3" == normalized_path.name
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 "test_audio.wav" == original_path.name
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:
hash_value = normalizer_service._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:
size = normalizer_service._get_file_size(temp_path)
assert size > 0
assert isinstance(size, int)
finally:
temp_path.unlink()
@patch("app.services.sound_normalizer.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")
duration = normalizer_service._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.services.sound_normalizer.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")
duration = normalizer_service._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.object(normalizer_service, "_get_audio_duration", return_value=6000), \
patch.object(normalizer_service, "_get_file_size", return_value=2048), \
patch.object(normalizer_service, "_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.object(normalizer_service, "_get_audio_duration", return_value=5500), \
patch.object(normalizer_service, "_get_file_size", return_value=1500), \
patch.object(normalizer_service, "_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