490 lines
17 KiB
Python
490 lines
17 KiB
Python
"""
|
|
Unit tests for AudioProcessor class.
|
|
"""
|
|
|
|
import pytest
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
|
|
from src.core.audio_processor import (
|
|
AudioProcessor,
|
|
CensorshipMethod,
|
|
ProcessingOptions,
|
|
ProcessingResult
|
|
)
|
|
|
|
|
|
class TestCensorshipMethod:
|
|
"""Test CensorshipMethod enum."""
|
|
|
|
def test_enum_values(self):
|
|
"""Test enum values are correct."""
|
|
assert CensorshipMethod.SILENCE.value == "silence"
|
|
assert CensorshipMethod.BEEP.value == "beep"
|
|
assert CensorshipMethod.WHITE_NOISE.value == "white_noise"
|
|
assert CensorshipMethod.FADE.value == "fade"
|
|
|
|
def test_enum_membership(self):
|
|
"""Test enum membership."""
|
|
methods = list(CensorshipMethod)
|
|
assert len(methods) == 4
|
|
assert CensorshipMethod.SILENCE in methods
|
|
assert CensorshipMethod.BEEP in methods
|
|
assert CensorshipMethod.WHITE_NOISE in methods
|
|
assert CensorshipMethod.FADE in methods
|
|
|
|
|
|
class TestProcessingOptions:
|
|
"""Test ProcessingOptions dataclass."""
|
|
|
|
def test_default_values(self):
|
|
"""Test default option values."""
|
|
options = ProcessingOptions()
|
|
|
|
assert options.censorship_method == CensorshipMethod.SILENCE
|
|
assert options.beep_frequency == 1000
|
|
assert options.beep_volume == -20
|
|
assert options.noise_volume == -30
|
|
assert options.fade_duration == 10
|
|
assert options.normalize_output is True
|
|
assert options.target_dBFS == -20.0
|
|
assert options.preserve_format is True
|
|
assert options.chunk_duration == 1800
|
|
|
|
def test_custom_values(self):
|
|
"""Test custom option values."""
|
|
options = ProcessingOptions(
|
|
censorship_method=CensorshipMethod.BEEP,
|
|
beep_frequency=800,
|
|
normalize_output=False,
|
|
target_dBFS=-10.0
|
|
)
|
|
|
|
assert options.censorship_method == CensorshipMethod.BEEP
|
|
assert options.beep_frequency == 800
|
|
assert options.normalize_output is False
|
|
assert options.target_dBFS == -10.0
|
|
|
|
|
|
class TestProcessingResult:
|
|
"""Test ProcessingResult dataclass."""
|
|
|
|
def test_default_result(self):
|
|
"""Test default result values."""
|
|
result = ProcessingResult(success=True)
|
|
|
|
assert result.success is True
|
|
assert result.output_path is None
|
|
assert result.duration is None
|
|
assert result.segments_censored == 0
|
|
assert result.processing_time is None
|
|
assert result.error is None
|
|
assert result.warnings == []
|
|
|
|
def test_result_with_values(self):
|
|
"""Test result with custom values."""
|
|
result = ProcessingResult(
|
|
success=True,
|
|
output_path="/path/to/output.mp3",
|
|
duration=30.5,
|
|
segments_censored=3,
|
|
processing_time=2.1,
|
|
warnings=["Warning 1", "Warning 2"]
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.output_path == "/path/to/output.mp3"
|
|
assert result.duration == 30.5
|
|
assert result.segments_censored == 3
|
|
assert result.processing_time == 2.1
|
|
assert len(result.warnings) == 2
|
|
|
|
def test_warnings_initialization(self):
|
|
"""Test warnings list is properly initialized."""
|
|
result = ProcessingResult(success=False)
|
|
assert result.warnings == []
|
|
assert isinstance(result.warnings, list)
|
|
|
|
|
|
class TestAudioProcessor:
|
|
"""Test AudioProcessor class."""
|
|
|
|
def test_initialization_default(self):
|
|
"""Test processor initialization with defaults."""
|
|
processor = AudioProcessor()
|
|
|
|
assert processor.audio_utils is not None
|
|
|
|
def test_initialization_with_utils(self):
|
|
"""Test processor initialization with custom utils."""
|
|
mock_utils = Mock()
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
assert processor.audio_utils == mock_utils
|
|
|
|
def test_process_audio_success(self, temp_dir):
|
|
"""Test successful audio processing."""
|
|
# Setup mock audio utils
|
|
mock_utils = Mock()
|
|
mock_utils.validate_audio_file.return_value = {
|
|
"valid": True,
|
|
"duration": 30.0,
|
|
"warnings": []
|
|
}
|
|
mock_utils.load_audio.return_value = Mock() # Mock audio data
|
|
mock_utils.apply_censorship.return_value = Mock() # Mock processed audio
|
|
mock_utils.normalize_audio.return_value = Mock() # Mock normalized audio
|
|
mock_utils.save_audio.return_value = True
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
# Setup test data
|
|
input_path = str(temp_dir / "input.mp3")
|
|
output_path = str(temp_dir / "output.mp3")
|
|
segments = [(5.0, 6.0), (10.0, 11.5)]
|
|
|
|
# Process audio
|
|
result = processor.process_audio(input_path, output_path, segments)
|
|
|
|
# Verify result
|
|
assert result.success is True
|
|
assert result.output_path == output_path
|
|
assert result.duration == 30.0
|
|
assert result.segments_censored == 2
|
|
assert result.processing_time is not None
|
|
assert result.error is None
|
|
|
|
# Verify method calls
|
|
mock_utils.validate_audio_file.assert_called_once_with(input_path)
|
|
mock_utils.load_audio.assert_called_once_with(input_path)
|
|
mock_utils.apply_censorship.assert_called_once()
|
|
mock_utils.normalize_audio.assert_called_once()
|
|
mock_utils.save_audio.assert_called_once()
|
|
|
|
def test_process_audio_invalid_file(self):
|
|
"""Test processing with invalid audio file."""
|
|
mock_utils = Mock()
|
|
mock_utils.validate_audio_file.return_value = {
|
|
"valid": False,
|
|
"errors": ["Invalid format", "Corrupted file"]
|
|
}
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
result = processor.process_audio("invalid.mp3", "output.mp3", [])
|
|
|
|
assert result.success is False
|
|
assert "Invalid audio file" in result.error
|
|
assert "Invalid format" in result.error
|
|
assert "Corrupted file" in result.error
|
|
|
|
def test_process_audio_no_segments(self, temp_dir):
|
|
"""Test processing with no censorship segments."""
|
|
mock_utils = Mock()
|
|
mock_utils.validate_audio_file.return_value = {
|
|
"valid": True,
|
|
"duration": 20.0,
|
|
"warnings": []
|
|
}
|
|
mock_utils.load_audio.return_value = Mock()
|
|
mock_utils.normalize_audio.return_value = Mock()
|
|
mock_utils.save_audio.return_value = True
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
result = processor.process_audio(
|
|
str(temp_dir / "input.mp3"),
|
|
str(temp_dir / "output.mp3"),
|
|
[] # No segments
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.segments_censored == 0
|
|
|
|
# Should not call apply_censorship
|
|
mock_utils.apply_censorship.assert_not_called()
|
|
|
|
def test_process_audio_with_beep(self, temp_dir):
|
|
"""Test processing with beep censorship method."""
|
|
mock_utils = Mock()
|
|
mock_utils.validate_audio_file.return_value = {
|
|
"valid": True,
|
|
"duration": 15.0,
|
|
"warnings": []
|
|
}
|
|
mock_utils.load_audio.return_value = Mock()
|
|
mock_utils.apply_censorship.return_value = Mock()
|
|
mock_utils.normalize_audio.return_value = Mock()
|
|
mock_utils.save_audio.return_value = True
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
options = ProcessingOptions(
|
|
censorship_method=CensorshipMethod.BEEP,
|
|
beep_frequency=800
|
|
)
|
|
|
|
result = processor.process_audio(
|
|
str(temp_dir / "input.mp3"),
|
|
str(temp_dir / "output.mp3"),
|
|
[(1.0, 2.0)],
|
|
options
|
|
)
|
|
|
|
assert result.success is True
|
|
|
|
# Verify apply_censorship was called with beep parameters
|
|
call_args = mock_utils.apply_censorship.call_args
|
|
assert call_args[0][2] == "beep" # method
|
|
assert call_args[1]["frequency"] == 800
|
|
|
|
def test_process_audio_progress_callback(self, temp_dir):
|
|
"""Test processing with progress callback."""
|
|
mock_utils = Mock()
|
|
mock_utils.validate_audio_file.return_value = {
|
|
"valid": True,
|
|
"duration": 10.0,
|
|
"warnings": []
|
|
}
|
|
mock_utils.load_audio.return_value = Mock()
|
|
mock_utils.apply_censorship.return_value = Mock()
|
|
mock_utils.normalize_audio.return_value = Mock()
|
|
mock_utils.save_audio.return_value = True
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
# Track progress calls
|
|
progress_calls = []
|
|
def progress_callback(message, percent):
|
|
progress_calls.append((message, percent))
|
|
|
|
result = processor.process_audio(
|
|
str(temp_dir / "input.mp3"),
|
|
str(temp_dir / "output.mp3"),
|
|
[(1.0, 2.0)],
|
|
progress_callback=progress_callback
|
|
)
|
|
|
|
assert result.success is True
|
|
assert len(progress_calls) >= 5 # Should have multiple progress updates
|
|
|
|
# Check progress percentages increase
|
|
percentages = [call[1] for call in progress_calls]
|
|
assert percentages == sorted(percentages)
|
|
assert percentages[0] == 0
|
|
assert percentages[-1] == 100
|
|
|
|
def test_process_audio_save_failure(self, temp_dir):
|
|
"""Test handling of save failure."""
|
|
mock_utils = Mock()
|
|
mock_utils.validate_audio_file.return_value = {
|
|
"valid": True,
|
|
"duration": 10.0,
|
|
"warnings": []
|
|
}
|
|
mock_utils.load_audio.return_value = Mock()
|
|
mock_utils.normalize_audio.return_value = Mock()
|
|
mock_utils.save_audio.return_value = False # Save fails
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
result = processor.process_audio(
|
|
str(temp_dir / "input.mp3"),
|
|
str(temp_dir / "output.mp3"),
|
|
[]
|
|
)
|
|
|
|
assert result.success is False
|
|
assert "Failed to save processed audio" in result.error
|
|
|
|
def test_process_audio_exception(self, temp_dir):
|
|
"""Test handling of processing exception."""
|
|
mock_utils = Mock()
|
|
mock_utils.validate_audio_file.side_effect = Exception("Test error")
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
result = processor.process_audio(
|
|
str(temp_dir / "input.mp3"),
|
|
str(temp_dir / "output.mp3"),
|
|
[]
|
|
)
|
|
|
|
assert result.success is False
|
|
assert "Test error" in result.error
|
|
|
|
def test_process_batch(self, temp_dir):
|
|
"""Test batch processing multiple files."""
|
|
mock_utils = Mock()
|
|
mock_utils.validate_audio_file.return_value = {
|
|
"valid": True,
|
|
"duration": 20.0,
|
|
"warnings": []
|
|
}
|
|
mock_utils.load_audio.return_value = Mock()
|
|
mock_utils.normalize_audio.return_value = Mock()
|
|
mock_utils.save_audio.return_value = True
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
# Setup batch mapping
|
|
file_mappings = [
|
|
(str(temp_dir / "input1.mp3"), str(temp_dir / "output1.mp3"), [(1.0, 2.0)]),
|
|
(str(temp_dir / "input2.mp3"), str(temp_dir / "output2.mp3"), [(3.0, 4.0)]),
|
|
(str(temp_dir / "input3.mp3"), str(temp_dir / "output3.mp3"), [])
|
|
]
|
|
|
|
# Track progress
|
|
progress_calls = []
|
|
def progress_callback(message, percent):
|
|
progress_calls.append((message, percent))
|
|
|
|
results = processor.process_batch(
|
|
file_mappings,
|
|
progress_callback=progress_callback
|
|
)
|
|
|
|
assert len(results) == 3
|
|
assert all(result.success for result in results)
|
|
|
|
# Check progress includes file numbers
|
|
file_messages = [call[0] for call in progress_calls if "File" in call[0]]
|
|
assert len(file_messages) > 0
|
|
assert "File 1/3" in file_messages[0]
|
|
|
|
def test_validate_segments_valid(self):
|
|
"""Test validation of valid segments."""
|
|
processor = AudioProcessor()
|
|
|
|
segments = [(1.0, 3.0), (5.0, 7.0), (10.0, 12.0)]
|
|
duration = 15.0
|
|
|
|
cleaned, warnings = processor.validate_segments(segments, duration)
|
|
|
|
assert len(cleaned) == 3
|
|
assert cleaned == segments
|
|
assert len(warnings) == 0
|
|
|
|
def test_validate_segments_invalid_order(self):
|
|
"""Test validation with invalid segment order."""
|
|
processor = AudioProcessor()
|
|
|
|
segments = [(3.0, 1.0), (5.0, 7.0)] # First segment has start > end
|
|
duration = 10.0
|
|
|
|
cleaned, warnings = processor.validate_segments(segments, duration)
|
|
|
|
assert len(cleaned) == 1 # Only valid segment
|
|
assert cleaned[0] == (5.0, 7.0)
|
|
assert len(warnings) == 1
|
|
assert "Invalid segment" in warnings[0]
|
|
|
|
def test_validate_segments_beyond_duration(self):
|
|
"""Test validation with segments beyond audio duration."""
|
|
processor = AudioProcessor()
|
|
|
|
segments = [(1.0, 3.0), (8.0, 12.0), (15.0, 20.0)] # Last two beyond duration
|
|
duration = 10.0
|
|
|
|
cleaned, warnings = processor.validate_segments(segments, duration)
|
|
|
|
assert len(cleaned) == 2
|
|
assert cleaned[0] == (1.0, 3.0)
|
|
assert cleaned[1] == (8.0, 10.0) # Clipped to duration
|
|
assert len(warnings) >= 1
|
|
|
|
def test_validate_segments_overlapping(self):
|
|
"""Test validation with overlapping segments."""
|
|
processor = AudioProcessor()
|
|
|
|
segments = [(1.0, 4.0), (3.0, 6.0), (8.0, 10.0)] # First two overlap
|
|
duration = 15.0
|
|
|
|
cleaned, warnings = processor.validate_segments(segments, duration)
|
|
|
|
assert len(cleaned) == 2 # Overlapping segment removed
|
|
assert (1.0, 4.0) in cleaned
|
|
assert (8.0, 10.0) in cleaned
|
|
assert len(warnings) >= 1
|
|
assert "Overlapping segments" in warnings[0]
|
|
|
|
def test_validate_segments_sorting(self):
|
|
"""Test that segments are sorted by start time."""
|
|
processor = AudioProcessor()
|
|
|
|
segments = [(8.0, 10.0), (1.0, 3.0), (5.0, 7.0)] # Unsorted
|
|
duration = 15.0
|
|
|
|
cleaned, warnings = processor.validate_segments(segments, duration)
|
|
|
|
assert len(cleaned) == 3
|
|
assert cleaned == [(1.0, 3.0), (5.0, 7.0), (8.0, 10.0)] # Sorted
|
|
|
|
def test_estimate_processing_time(self, temp_dir):
|
|
"""Test processing time estimation."""
|
|
mock_utils = Mock()
|
|
mock_utils.get_duration.return_value = 120.0 # 2 minutes
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
estimate = processor.estimate_processing_time(
|
|
str(temp_dir / "test.mp3"),
|
|
num_segments=5
|
|
)
|
|
|
|
# Should be base_time + segment_time + overhead
|
|
# (120/60 * 0.1) + (5 * 0.05) + 2.0 = 0.2 + 0.25 + 2.0 = 2.45
|
|
assert estimate == pytest.approx(2.45, rel=0.1)
|
|
|
|
def test_estimate_processing_time_error(self):
|
|
"""Test processing time estimation with error."""
|
|
mock_utils = Mock()
|
|
mock_utils.get_duration.side_effect = Exception("File not found")
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
estimate = processor.estimate_processing_time("nonexistent.mp3", 3)
|
|
|
|
assert estimate == 10.0 # Default estimate
|
|
|
|
def test_get_supported_formats(self):
|
|
"""Test getting supported formats."""
|
|
mock_utils = Mock()
|
|
mock_utils.SUPPORTED_FORMATS = {'mp3', 'wav', 'flac', 'm4a'}
|
|
|
|
processor = AudioProcessor(audio_utils=mock_utils)
|
|
|
|
formats = processor.get_supported_formats()
|
|
|
|
assert formats == {'mp3', 'wav', 'flac', 'm4a'}
|
|
|
|
@patch('src.core.audio_processor.which')
|
|
def test_check_dependencies(self, mock_which):
|
|
"""Test dependency checking."""
|
|
mock_which.return_value = "/usr/bin/ffmpeg" # ffmpeg found
|
|
|
|
processor = AudioProcessor()
|
|
|
|
deps = processor.check_dependencies()
|
|
|
|
assert deps['ffmpeg'] is True
|
|
assert deps['pydub'] is True
|
|
assert deps['librosa'] is True
|
|
assert deps['numpy'] is True
|
|
|
|
@patch('src.core.audio_processor.which')
|
|
def test_check_dependencies_missing_ffmpeg(self, mock_which):
|
|
"""Test dependency checking with missing ffmpeg."""
|
|
mock_which.return_value = None # ffmpeg not found
|
|
|
|
processor = AudioProcessor()
|
|
|
|
deps = processor.check_dependencies()
|
|
|
|
assert deps['ffmpeg'] is False
|
|
assert deps['pydub'] is True # Others still available
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v']) |