"""Unit tests for AudioProcessingApplicationService""" import pytest import tempfile import os import time from unittest.mock import Mock, MagicMock, patch, call from contextlib import contextmanager from src.application.services.audio_processing_service import AudioProcessingApplicationService from src.application.dtos.audio_upload_dto import AudioUploadDto from src.application.dtos.processing_request_dto import ProcessingRequestDto from src.application.dtos.processing_result_dto import ProcessingResultDto from src.domain.models.audio_content import AudioContent from src.domain.models.text_content import TextContent from src.domain.models.translation_request import TranslationRequest from src.domain.models.speech_synthesis_request import SpeechSynthesisRequest from src.domain.models.voice_settings import VoiceSettings from src.domain.exceptions import ( AudioProcessingException, SpeechRecognitionException, TranslationFailedException, SpeechSynthesisException ) from src.infrastructure.config.app_config import AppConfig from src.infrastructure.config.dependency_container import DependencyContainer class TestAudioProcessingApplicationService: """Test cases for AudioProcessingApplicationService""" @pytest.fixture def mock_container(self): """Create mock dependency container""" container = Mock(spec=DependencyContainer) # Mock providers mock_stt_provider = Mock() mock_translation_provider = Mock() mock_tts_provider = Mock() container.get_stt_provider.return_value = mock_stt_provider container.get_translation_provider.return_value = mock_translation_provider container.get_tts_provider.return_value = mock_tts_provider return container @pytest.fixture def mock_config(self): """Create mock application config""" config = Mock(spec=AppConfig) # Mock configuration methods config.get_logging_config.return_value = { 'level': 'INFO', 'enable_file_logging': False, 'log_file_path': '/tmp/test.log', 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s' } config.get_processing_config.return_value = { 'max_file_size_mb': 100, 'supported_audio_formats': ['wav', 'mp3', 'flac', 'ogg', 'm4a'], 'temp_dir': '/tmp', 'cleanup_temp_files': True, 'processing_timeout_seconds': 300 } config.get_stt_config.return_value = { 'preferred_providers': ['whisper-small', 'whisper-medium'] } config.get_tts_config.return_value = { 'preferred_providers': ['kokoro', 'dia'] } return config @pytest.fixture def sample_audio_upload(self): """Create sample audio upload DTO""" return AudioUploadDto( filename="test_audio.wav", content=b"fake_audio_content_" + b"x" * 1000, # 1KB+ of fake audio content_type="audio/wav" ) @pytest.fixture def sample_processing_request(self, sample_audio_upload): """Create sample processing request DTO""" return ProcessingRequestDto( audio=sample_audio_upload, asr_model="whisper-small", target_language="es", voice="kokoro", speed=1.0, source_language="en" ) @pytest.fixture def service(self, mock_container, mock_config): """Create AudioProcessingApplicationService instance""" mock_container.resolve.return_value = mock_config return AudioProcessingApplicationService(mock_container, mock_config) def test_initialization(self, mock_container, mock_config): """Test service initialization""" service = AudioProcessingApplicationService(mock_container, mock_config) assert service._container == mock_container assert service._config == mock_config assert service._temp_files == {} assert service._error_mapper is not None assert service._recovery_manager is not None def test_initialization_without_config(self, mock_container, mock_config): """Test service initialization without explicit config""" mock_container.resolve.return_value = mock_config service = AudioProcessingApplicationService(mock_container) assert service._container == mock_container assert service._config == mock_config mock_container.resolve.assert_called_once_with(AppConfig) @patch('src.application.services.audio_processing_service.get_structured_logger') def test_setup_logging_success(self, mock_logger, service, mock_config): """Test successful logging setup""" mock_config.get_logging_config.return_value = { 'level': 'DEBUG', 'enable_file_logging': True, 'log_file_path': '/tmp/test.log', 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s' } service._setup_logging() # Verify logging configuration was retrieved mock_config.get_logging_config.assert_called_once() @patch('src.application.services.audio_processing_service.get_structured_logger') def test_setup_logging_failure(self, mock_logger, service, mock_config): """Test logging setup failure handling""" mock_config.get_logging_config.side_effect = Exception("Config error") # Should not raise exception service._setup_logging() # Warning should be logged mock_logger.return_value.warning.assert_called_once() def test_validate_request_success(self, service, sample_processing_request): """Test successful request validation""" # Should not raise exception service._validate_request(sample_processing_request) def test_validate_request_invalid_type(self, service): """Test request validation with invalid type""" with pytest.raises(ValueError, match="Request must be a ProcessingRequestDto instance"): service._validate_request("invalid_request") def test_validate_request_file_too_large(self, service, sample_processing_request, mock_config): """Test request validation with file too large""" mock_config.get_processing_config.return_value['max_file_size_mb'] = 0.001 # Very small limit with pytest.raises(ValueError, match="Audio file too large"): service._validate_request(sample_processing_request) def test_validate_request_unsupported_format(self, service, sample_processing_request, mock_config): """Test request validation with unsupported format""" sample_processing_request.audio.filename = "test.xyz" mock_config.get_processing_config.return_value['supported_audio_formats'] = ['wav', 'mp3'] with pytest.raises(ValueError, match="Unsupported audio format"): service._validate_request(sample_processing_request) @patch('os.makedirs') @patch('shutil.rmtree') def test_create_temp_directory(self, mock_rmtree, mock_makedirs, service): """Test temporary directory creation and cleanup""" correlation_id = "test-123" with service._create_temp_directory(correlation_id) as temp_dir: assert correlation_id in temp_dir mock_makedirs.assert_called_once() # Cleanup should be called mock_rmtree.assert_called_once() @patch('builtins.open', create=True) def test_convert_upload_to_audio_content_success(self, mock_open, service, sample_audio_upload): """Test successful audio upload conversion""" temp_dir = "/tmp/test" mock_file = MagicMock() mock_open.return_value.__enter__.return_value = mock_file result = service._convert_upload_to_audio_content(sample_audio_upload, temp_dir) assert isinstance(result, AudioContent) assert result.data == sample_audio_upload.content assert result.format == "wav" mock_file.write.assert_called_once_with(sample_audio_upload.content) @patch('builtins.open', side_effect=IOError("File error")) def test_convert_upload_to_audio_content_failure(self, mock_open, service, sample_audio_upload): """Test audio upload conversion failure""" temp_dir = "/tmp/test" with pytest.raises(AudioProcessingException, match="Failed to process uploaded audio"): service._convert_upload_to_audio_content(sample_audio_upload, temp_dir) def test_perform_speech_recognition_success(self, service, mock_container): """Test successful speech recognition""" audio = AudioContent(data=b"audio", format="wav", sample_rate=16000, duration=1.0) model = "whisper-small" correlation_id = "test-123" # Mock STT provider mock_stt_provider = Mock() expected_text = TextContent(text="Hello world", language="en") mock_stt_provider.transcribe.return_value = expected_text mock_container.get_stt_provider.return_value = mock_stt_provider result = service._perform_speech_recognition(audio, model, correlation_id) assert result == expected_text mock_container.get_stt_provider.assert_called_once_with(model) mock_stt_provider.transcribe.assert_called_once_with(audio, model) def test_perform_speech_recognition_failure(self, service, mock_container): """Test speech recognition failure""" audio = AudioContent(data=b"audio", format="wav", sample_rate=16000, duration=1.0) model = "whisper-small" correlation_id = "test-123" # Mock STT provider to raise exception mock_stt_provider = Mock() mock_stt_provider.transcribe.side_effect = Exception("STT failed") mock_container.get_stt_provider.return_value = mock_stt_provider with pytest.raises(SpeechRecognitionException, match="Speech recognition failed"): service._perform_speech_recognition(audio, model, correlation_id) def test_perform_translation_success(self, service, mock_container): """Test successful translation""" text = TextContent(text="Hello world", language="en") source_language = "en" target_language = "es" correlation_id = "test-123" # Mock translation provider mock_translation_provider = Mock() expected_text = TextContent(text="Hola mundo", language="es") mock_translation_provider.translate.return_value = expected_text mock_container.get_translation_provider.return_value = mock_translation_provider result = service._perform_translation(text, source_language, target_language, correlation_id) assert result == expected_text mock_container.get_translation_provider.assert_called_once() mock_translation_provider.translate.assert_called_once() def test_perform_translation_failure(self, service, mock_container): """Test translation failure""" text = TextContent(text="Hello world", language="en") source_language = "en" target_language = "es" correlation_id = "test-123" # Mock translation provider to raise exception mock_translation_provider = Mock() mock_translation_provider.translate.side_effect = Exception("Translation failed") mock_container.get_translation_provider.return_value = mock_translation_provider with pytest.raises(TranslationFailedException, match="Translation failed"): service._perform_translation(text, source_language, target_language, correlation_id) @patch('builtins.open', create=True) def test_perform_speech_synthesis_success(self, mock_open, service, mock_container): """Test successful speech synthesis""" text = TextContent(text="Hola mundo", language="es") voice = "kokoro" speed = 1.0 language = "es" temp_dir = "/tmp/test" correlation_id = "test-123" # Mock TTS provider mock_tts_provider = Mock() mock_audio = AudioContent(data=b"synthesized_audio", format="wav", sample_rate=22050, duration=2.0) mock_tts_provider.synthesize.return_value = mock_audio mock_container.get_tts_provider.return_value = mock_tts_provider # Mock file operations mock_file = MagicMock() mock_open.return_value.__enter__.return_value = mock_file result = service._perform_speech_synthesis(text, voice, speed, language, temp_dir, correlation_id) assert correlation_id in result assert result.endswith(".wav") mock_container.get_tts_provider.assert_called_once_with(voice) mock_tts_provider.synthesize.assert_called_once() mock_file.write.assert_called_once_with(mock_audio.data) def test_perform_speech_synthesis_failure(self, service, mock_container): """Test speech synthesis failure""" text = TextContent(text="Hola mundo", language="es") voice = "kokoro" speed = 1.0 language = "es" temp_dir = "/tmp/test" correlation_id = "test-123" # Mock TTS provider to raise exception mock_tts_provider = Mock() mock_tts_provider.synthesize.side_effect = Exception("TTS failed") mock_container.get_tts_provider.return_value = mock_tts_provider with pytest.raises(SpeechSynthesisException, match="Speech synthesis failed"): service._perform_speech_synthesis(text, voice, speed, language, temp_dir, correlation_id) def test_get_error_code_from_exception(self, service): """Test error code mapping from exceptions""" assert service._get_error_code_from_exception(SpeechRecognitionException("test")) == 'STT_ERROR' assert service._get_error_code_from_exception(TranslationFailedException("test")) == 'TRANSLATION_ERROR' assert service._get_error_code_from_exception(SpeechSynthesisException("test")) == 'TTS_ERROR' assert service._get_error_code_from_exception(ValueError("test")) == 'VALIDATION_ERROR' assert service._get_error_code_from_exception(Exception("test")) == 'SYSTEM_ERROR' @patch('os.path.exists') @patch('os.remove') def test_cleanup_temp_files_success(self, mock_remove, mock_exists, service): """Test successful temporary file cleanup""" service._temp_files = { "/tmp/file1.wav": "/tmp/file1.wav", "/tmp/file2.wav": "/tmp/file2.wav" } mock_exists.return_value = True service._cleanup_temp_files() assert service._temp_files == {} assert mock_remove.call_count == 2 @patch('os.path.exists') @patch('os.remove', side_effect=OSError("Permission denied")) def test_cleanup_temp_files_failure(self, mock_remove, mock_exists, service): """Test temporary file cleanup with failures""" service._temp_files = {"/tmp/file1.wav": "/tmp/file1.wav"} mock_exists.return_value = True # Should not raise exception service._cleanup_temp_files() # File should still be removed from tracking assert service._temp_files == {} def test_get_processing_status(self, service): """Test processing status retrieval""" correlation_id = "test-123" result = service.get_processing_status(correlation_id) assert result['correlation_id'] == correlation_id assert 'status' in result assert 'message' in result def test_get_supported_configurations(self, service): """Test supported configurations retrieval""" result = service.get_supported_configurations() assert 'asr_models' in result assert 'voices' in result assert 'languages' in result assert 'audio_formats' in result assert 'max_file_size_mb' in result assert 'speed_range' in result # Verify expected values assert 'whisper-small' in result['asr_models'] assert 'kokoro' in result['voices'] assert 'en' in result['languages'] def test_cleanup(self, service): """Test service cleanup""" service._temp_files = {"/tmp/test.wav": "/tmp/test.wav"} with patch.object(service, '_cleanup_temp_files') as mock_cleanup: service.cleanup() mock_cleanup.assert_called_once() def test_context_manager(self, service): """Test service as context manager""" with patch.object(service, 'cleanup') as mock_cleanup: with service as svc: assert svc == service mock_cleanup.assert_called_once() @patch('src.application.services.audio_processing_service.time.time') @patch.object(AudioProcessingApplicationService, '_create_temp_directory') @patch.object(AudioProcessingApplicationService, '_convert_upload_to_audio_content') @patch.object(AudioProcessingApplicationService, '_perform_speech_recognition_with_recovery') @patch.object(AudioProcessingApplicationService, '_perform_translation_with_recovery') @patch.object(AudioProcessingApplicationService, '_perform_speech_synthesis_with_recovery') def test_process_audio_pipeline_success(self, mock_tts, mock_translation, mock_stt, mock_convert, mock_temp_dir, mock_time, service, sample_processing_request): """Test successful audio processing pipeline""" # Setup mocks mock_time.side_effect = [0.0, 5.0] # Start and end times mock_temp_dir.return_value.__enter__.return_value = "/tmp/test" mock_temp_dir.return_value.__exit__.return_value = None mock_audio = AudioContent(data=b"audio", format="wav", sample_rate=16000, duration=1.0) mock_convert.return_value = mock_audio mock_original_text = TextContent(text="Hello world", language="en") mock_stt.return_value = mock_original_text mock_translated_text = TextContent(text="Hola mundo", language="es") mock_translation.return_value = mock_translated_text mock_tts.return_value = "/tmp/test/output_123.wav" with patch.object(service, '_validate_request'): result = service.process_audio_pipeline(sample_processing_request) # Verify result assert isinstance(result, ProcessingResultDto) assert result.success is True assert result.original_text == "Hello world" assert result.translated_text == "Hola mundo" assert result.audio_path == "/tmp/test/output_123.wav" assert result.processing_time == 5.0 @patch('src.application.services.audio_processing_service.time.time') def test_process_audio_pipeline_validation_error(self, mock_time, service, sample_processing_request): """Test audio processing pipeline with validation error""" mock_time.side_effect = [0.0, 1.0] with patch.object(service, '_validate_request', side_effect=ValueError("Invalid request")): result = service.process_audio_pipeline(sample_processing_request) # Verify error result assert isinstance(result, ProcessingResultDto) assert result.success is False assert "Invalid request" in result.error_message assert result.processing_time == 1.0 @patch('src.application.services.audio_processing_service.time.time') def test_process_audio_pipeline_domain_exception(self, mock_time, service, sample_processing_request): """Test audio processing pipeline with domain exception""" mock_time.side_effect = [0.0, 2.0] with patch.object(service, '_validate_request'): with patch.object(service, '_create_temp_directory', side_effect=SpeechRecognitionException("STT failed")): result = service.process_audio_pipeline(sample_processing_request) # Verify error result assert isinstance(result, ProcessingResultDto) assert result.success is False assert result.error_message is not None assert result.processing_time == 2.0 def test_recovery_methods_exist(self, service): """Test that recovery methods exist and are callable""" # These methods should exist for error recovery assert hasattr(service, '_perform_speech_recognition_with_recovery') assert hasattr(service, '_perform_translation_with_recovery') assert hasattr(service, '_perform_speech_synthesis_with_recovery') assert callable(service._perform_speech_recognition_with_recovery) assert callable(service._perform_translation_with_recovery) assert callable(service._perform_speech_synthesis_with_recovery)