8847131f24
- Removed unnecessary blank lines and adjusted formatting in test files. - Ensured consistent use of commas in function calls and assertions across various test cases. - Updated import statements for better organization and clarity. - Enhanced mock setups in tests for better isolation and reliability. - Improved assertions to follow a consistent style for better readability.
594 lines
21 KiB
Python
594 lines
21 KiB
Python
"""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
|