""" 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'])