Spaces:
Build error
Build error
| """Unit tests for ConfigurationApplicationService""" | |
| import pytest | |
| import os | |
| import json | |
| from unittest.mock import Mock, MagicMock, patch, mock_open | |
| from src.application.services.configuration_service import ( | |
| ConfigurationApplicationService, | |
| ConfigurationException | |
| ) | |
| from src.infrastructure.config.app_config import AppConfig | |
| from src.infrastructure.config.dependency_container import DependencyContainer | |
| class TestConfigurationApplicationService: | |
| """Test cases for ConfigurationApplicationService""" | |
| def mock_container(self): | |
| """Create mock dependency container""" | |
| container = Mock(spec=DependencyContainer) | |
| return container | |
| def mock_config(self): | |
| """Create mock application config""" | |
| config = Mock(spec=AppConfig) | |
| # Mock configuration methods | |
| config.get_tts_config.return_value = { | |
| 'preferred_providers': ['kokoro', 'dia'], | |
| 'default_speed': 1.0, | |
| 'default_language': 'en', | |
| 'enable_streaming': False, | |
| 'max_text_length': 5000 | |
| } | |
| config.get_stt_config.return_value = { | |
| 'preferred_providers': ['whisper', 'parakeet'], | |
| 'default_model': 'whisper', | |
| 'chunk_length_s': 30, | |
| 'batch_size': 16, | |
| 'enable_vad': True | |
| } | |
| config.get_translation_config.return_value = { | |
| 'default_provider': 'nllb', | |
| 'model_name': 'nllb-200-3.3B', | |
| 'max_chunk_length': 512, | |
| 'batch_size': 8, | |
| 'cache_translations': True | |
| } | |
| config.get_processing_config.return_value = { | |
| 'temp_dir': '/tmp', | |
| 'cleanup_temp_files': True, | |
| 'max_file_size_mb': 100, | |
| 'supported_audio_formats': ['wav', 'mp3', 'flac'], | |
| 'processing_timeout_seconds': 300 | |
| } | |
| config.get_logging_config.return_value = { | |
| 'level': 'INFO', | |
| 'enable_file_logging': False, | |
| 'log_file_path': '/tmp/app.log', | |
| 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s' | |
| } | |
| # Mock config objects for attribute access | |
| config.tts = Mock() | |
| config.stt = Mock() | |
| config.translation = Mock() | |
| config.processing = Mock() | |
| config.logging = Mock() | |
| return config | |
| def service(self, mock_container, mock_config): | |
| """Create ConfigurationApplicationService instance""" | |
| mock_container.resolve.return_value = mock_config | |
| return ConfigurationApplicationService(mock_container, mock_config) | |
| def test_initialization(self, mock_container, mock_config): | |
| """Test service initialization""" | |
| service = ConfigurationApplicationService(mock_container, mock_config) | |
| assert service._container == mock_container | |
| assert service._config == mock_config | |
| assert service._error_mapper 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 = ConfigurationApplicationService(mock_container) | |
| assert service._container == mock_container | |
| assert service._config == mock_config | |
| mock_container.resolve.assert_called_once_with(AppConfig) | |
| def test_get_current_configuration_success(self, service, mock_config): | |
| """Test successful current configuration retrieval""" | |
| result = service.get_current_configuration() | |
| assert 'tts' in result | |
| assert 'stt' in result | |
| assert 'translation' in result | |
| assert 'processing' in result | |
| assert 'logging' in result | |
| # Verify all config methods were called | |
| mock_config.get_tts_config.assert_called_once() | |
| mock_config.get_stt_config.assert_called_once() | |
| mock_config.get_translation_config.assert_called_once() | |
| mock_config.get_processing_config.assert_called_once() | |
| mock_config.get_logging_config.assert_called_once() | |
| def test_get_current_configuration_failure(self, service, mock_config): | |
| """Test current configuration retrieval failure""" | |
| mock_config.get_tts_config.side_effect = Exception("Config error") | |
| with pytest.raises(ConfigurationException, match="Failed to retrieve configuration"): | |
| service.get_current_configuration() | |
| def test_get_tts_configuration_success(self, service, mock_config): | |
| """Test successful TTS configuration retrieval""" | |
| result = service.get_tts_configuration() | |
| assert result['preferred_providers'] == ['kokoro', 'dia'] | |
| assert result['default_speed'] == 1.0 | |
| mock_config.get_tts_config.assert_called_once() | |
| def test_get_tts_configuration_failure(self, service, mock_config): | |
| """Test TTS configuration retrieval failure""" | |
| mock_config.get_tts_config.side_effect = Exception("TTS config error") | |
| with pytest.raises(ConfigurationException, match="Failed to retrieve TTS configuration"): | |
| service.get_tts_configuration() | |
| def test_get_stt_configuration_success(self, service, mock_config): | |
| """Test successful STT configuration retrieval""" | |
| result = service.get_stt_configuration() | |
| assert result['preferred_providers'] == ['whisper', 'parakeet'] | |
| assert result['default_model'] == 'whisper' | |
| mock_config.get_stt_config.assert_called_once() | |
| def test_get_stt_configuration_failure(self, service, mock_config): | |
| """Test STT configuration retrieval failure""" | |
| mock_config.get_stt_config.side_effect = Exception("STT config error") | |
| with pytest.raises(ConfigurationException, match="Failed to retrieve STT configuration"): | |
| service.get_stt_configuration() | |
| def test_get_translation_configuration_success(self, service, mock_config): | |
| """Test successful translation configuration retrieval""" | |
| result = service.get_translation_configuration() | |
| assert result['default_provider'] == 'nllb' | |
| assert result['model_name'] == 'nllb-200-3.3B' | |
| mock_config.get_translation_config.assert_called_once() | |
| def test_get_translation_configuration_failure(self, service, mock_config): | |
| """Test translation configuration retrieval failure""" | |
| mock_config.get_translation_config.side_effect = Exception("Translation config error") | |
| with pytest.raises(ConfigurationException, match="Failed to retrieve translation configuration"): | |
| service.get_translation_configuration() | |
| def test_get_processing_configuration_success(self, service, mock_config): | |
| """Test successful processing configuration retrieval""" | |
| result = service.get_processing_configuration() | |
| assert result['temp_dir'] == '/tmp' | |
| assert result['max_file_size_mb'] == 100 | |
| mock_config.get_processing_config.assert_called_once() | |
| def test_get_processing_configuration_failure(self, service, mock_config): | |
| """Test processing configuration retrieval failure""" | |
| mock_config.get_processing_config.side_effect = Exception("Processing config error") | |
| with pytest.raises(ConfigurationException, match="Failed to retrieve processing configuration"): | |
| service.get_processing_configuration() | |
| def test_get_logging_configuration_success(self, service, mock_config): | |
| """Test successful logging configuration retrieval""" | |
| result = service.get_logging_configuration() | |
| assert result['level'] == 'INFO' | |
| assert result['enable_file_logging'] is False | |
| mock_config.get_logging_config.assert_called_once() | |
| def test_get_logging_configuration_failure(self, service, mock_config): | |
| """Test logging configuration retrieval failure""" | |
| mock_config.get_logging_config.side_effect = Exception("Logging config error") | |
| with pytest.raises(ConfigurationException, match="Failed to retrieve logging configuration"): | |
| service.get_logging_configuration() | |
| def test_update_tts_configuration_success(self, service, mock_config): | |
| """Test successful TTS configuration update""" | |
| updates = { | |
| 'default_speed': 1.5, | |
| 'enable_streaming': True | |
| } | |
| result = service.update_tts_configuration(updates) | |
| # Verify setattr was called for valid attributes | |
| assert hasattr(mock_config.tts, 'default_speed') | |
| assert hasattr(mock_config.tts, 'enable_streaming') | |
| # Verify updated config was returned | |
| mock_config.get_tts_config.assert_called() | |
| def test_update_tts_configuration_validation_error(self, service): | |
| """Test TTS configuration update with validation error""" | |
| updates = { | |
| 'default_speed': 5.0 # Invalid speed > 3.0 | |
| } | |
| with pytest.raises(ConfigurationException, match="default_speed must be between 0.1 and 3.0"): | |
| service.update_tts_configuration(updates) | |
| def test_update_stt_configuration_success(self, service, mock_config): | |
| """Test successful STT configuration update""" | |
| updates = { | |
| 'chunk_length_s': 60, | |
| 'enable_vad': False | |
| } | |
| result = service.update_stt_configuration(updates) | |
| # Verify setattr was called for valid attributes | |
| assert hasattr(mock_config.stt, 'chunk_length_s') | |
| assert hasattr(mock_config.stt, 'enable_vad') | |
| # Verify updated config was returned | |
| mock_config.get_stt_config.assert_called() | |
| def test_update_stt_configuration_validation_error(self, service): | |
| """Test STT configuration update with validation error""" | |
| updates = { | |
| 'chunk_length_s': -10 # Invalid negative value | |
| } | |
| with pytest.raises(ConfigurationException, match="chunk_length_s must be a positive integer"): | |
| service.update_stt_configuration(updates) | |
| def test_update_translation_configuration_success(self, service, mock_config): | |
| """Test successful translation configuration update""" | |
| updates = { | |
| 'max_chunk_length': 1024, | |
| 'cache_translations': False | |
| } | |
| result = service.update_translation_configuration(updates) | |
| # Verify setattr was called for valid attributes | |
| assert hasattr(mock_config.translation, 'max_chunk_length') | |
| assert hasattr(mock_config.translation, 'cache_translations') | |
| # Verify updated config was returned | |
| mock_config.get_translation_config.assert_called() | |
| def test_update_translation_configuration_validation_error(self, service): | |
| """Test translation configuration update with validation error""" | |
| updates = { | |
| 'max_chunk_length': 0 # Invalid zero value | |
| } | |
| with pytest.raises(ConfigurationException, match="max_chunk_length must be a positive integer"): | |
| service.update_translation_configuration(updates) | |
| def test_update_processing_configuration_success(self, service, mock_config): | |
| """Test successful processing configuration update""" | |
| updates = { | |
| 'max_file_size_mb': 200, | |
| 'cleanup_temp_files': False | |
| } | |
| with patch('pathlib.Path.mkdir'): | |
| result = service.update_processing_configuration(updates) | |
| # Verify setattr was called for valid attributes | |
| assert hasattr(mock_config.processing, 'max_file_size_mb') | |
| assert hasattr(mock_config.processing, 'cleanup_temp_files') | |
| # Verify updated config was returned | |
| mock_config.get_processing_config.assert_called() | |
| def test_update_processing_configuration_validation_error(self, service): | |
| """Test processing configuration update with validation error""" | |
| updates = { | |
| 'max_file_size_mb': -50 # Invalid negative value | |
| } | |
| with pytest.raises(ConfigurationException, match="max_file_size_mb must be a positive integer"): | |
| service.update_processing_configuration(updates) | |
| def test_validate_tts_updates_valid(self, service): | |
| """Test TTS updates validation with valid data""" | |
| updates = { | |
| 'preferred_providers': ['kokoro', 'dia'], | |
| 'default_speed': 1.5, | |
| 'default_language': 'es', | |
| 'enable_streaming': True, | |
| 'max_text_length': 10000 | |
| } | |
| # Should not raise exception | |
| service._validate_tts_updates(updates) | |
| def test_validate_tts_updates_invalid_provider(self, service): | |
| """Test TTS updates validation with invalid provider""" | |
| updates = { | |
| 'preferred_providers': ['invalid_provider'] | |
| } | |
| with pytest.raises(ConfigurationException, match="Invalid TTS provider"): | |
| service._validate_tts_updates(updates) | |
| def test_validate_tts_updates_invalid_speed(self, service): | |
| """Test TTS updates validation with invalid speed""" | |
| updates = { | |
| 'default_speed': 5.0 # Too high | |
| } | |
| with pytest.raises(ConfigurationException, match="default_speed must be between 0.1 and 3.0"): | |
| service._validate_tts_updates(updates) | |
| def test_validate_stt_updates_valid(self, service): | |
| """Test STT updates validation with valid data""" | |
| updates = { | |
| 'preferred_providers': ['whisper', 'parakeet'], | |
| 'default_model': 'whisper', | |
| 'chunk_length_s': 45, | |
| 'batch_size': 32, | |
| 'enable_vad': False | |
| } | |
| # Should not raise exception | |
| service._validate_stt_updates(updates) | |
| def test_validate_stt_updates_invalid_provider(self, service): | |
| """Test STT updates validation with invalid provider""" | |
| updates = { | |
| 'preferred_providers': ['invalid_stt'] | |
| } | |
| with pytest.raises(ConfigurationException, match="Invalid STT provider"): | |
| service._validate_stt_updates(updates) | |
| def test_validate_translation_updates_valid(self, service): | |
| """Test translation updates validation with valid data""" | |
| updates = { | |
| 'default_provider': 'nllb', | |
| 'model_name': 'nllb-200-1.3B', | |
| 'max_chunk_length': 256, | |
| 'batch_size': 4, | |
| 'cache_translations': False | |
| } | |
| # Should not raise exception | |
| service._validate_translation_updates(updates) | |
| def test_validate_processing_updates_valid(self, service): | |
| """Test processing updates validation with valid data""" | |
| updates = { | |
| 'temp_dir': '/tmp/test', | |
| 'cleanup_temp_files': True, | |
| 'max_file_size_mb': 150, | |
| 'supported_audio_formats': ['wav', 'mp3'], | |
| 'processing_timeout_seconds': 600 | |
| } | |
| with patch('pathlib.Path.mkdir'): | |
| # Should not raise exception | |
| service._validate_processing_updates(updates) | |
| def test_validate_processing_updates_invalid_format(self, service): | |
| """Test processing updates validation with invalid audio format""" | |
| updates = { | |
| 'supported_audio_formats': ['wav', 'invalid_format'] | |
| } | |
| with pytest.raises(ConfigurationException, match="Invalid audio format"): | |
| service._validate_processing_updates(updates) | |
| def test_save_configuration_to_file_success(self, service, mock_config): | |
| """Test successful configuration save to file""" | |
| file_path = "/tmp/config.json" | |
| service.save_configuration_to_file(file_path) | |
| mock_config.save_configuration.assert_called_once_with(file_path) | |
| def test_save_configuration_to_file_failure(self, service, mock_config): | |
| """Test configuration save to file failure""" | |
| file_path = "/tmp/config.json" | |
| mock_config.save_configuration.side_effect = Exception("Save failed") | |
| with pytest.raises(ConfigurationException, match="Failed to save configuration"): | |
| service.save_configuration_to_file(file_path) | |
| def test_load_configuration_from_file_success(self, mock_exists, service, mock_container): | |
| """Test successful configuration load from file""" | |
| file_path = "/tmp/config.json" | |
| mock_exists.return_value = True | |
| with patch('src.infrastructure.config.app_config.AppConfig') as mock_app_config: | |
| new_config = Mock() | |
| mock_app_config.return_value = new_config | |
| result = service.load_configuration_from_file(file_path) | |
| # Verify new config was created and registered | |
| mock_app_config.assert_called_once_with(config_file=file_path) | |
| mock_container.register_singleton.assert_called_once_with(AppConfig, new_config) | |
| def test_load_configuration_from_file_not_found(self, mock_exists, service): | |
| """Test configuration load from non-existent file""" | |
| file_path = "/tmp/nonexistent.json" | |
| mock_exists.return_value = False | |
| with pytest.raises(ConfigurationException, match="Configuration file not found"): | |
| service.load_configuration_from_file(file_path) | |
| def test_reload_configuration_success(self, service, mock_config): | |
| """Test successful configuration reload""" | |
| result = service.reload_configuration() | |
| mock_config.reload_configuration.assert_called_once() | |
| assert 'tts' in result | |
| def test_reload_configuration_failure(self, service, mock_config): | |
| """Test configuration reload failure""" | |
| mock_config.reload_configuration.side_effect = Exception("Reload failed") | |
| with pytest.raises(ConfigurationException, match="Failed to reload configuration"): | |
| service.reload_configuration() | |
| def test_get_provider_availability(self, service, mock_container): | |
| """Test provider availability check""" | |
| # Mock factories | |
| mock_tts_factory = Mock() | |
| mock_stt_factory = Mock() | |
| mock_translation_factory = Mock() | |
| mock_container.resolve.side_effect = [mock_tts_factory, mock_stt_factory, mock_translation_factory] | |
| # Mock successful provider creation | |
| mock_tts_factory.create_provider.return_value = Mock() | |
| mock_stt_factory.create_provider.return_value = Mock() | |
| mock_translation_factory.get_default_provider.return_value = Mock() | |
| result = service.get_provider_availability() | |
| assert 'tts' in result | |
| assert 'stt' in result | |
| assert 'translation' in result | |
| # All providers should be available | |
| assert all(result['tts'].values()) | |
| assert all(result['stt'].values()) | |
| assert result['translation']['nllb'] is True | |
| def test_get_system_info(self, service, mock_config): | |
| """Test system information retrieval""" | |
| mock_config.config_file = "/tmp/config.json" | |
| mock_config.processing.temp_dir = "/tmp" | |
| mock_config.logging.level = "INFO" | |
| mock_config.processing.supported_audio_formats = ['wav', 'mp3'] | |
| mock_config.processing.max_file_size_mb = 100 | |
| mock_config.processing.processing_timeout_seconds = 300 | |
| with patch.object(service, 'get_provider_availability', return_value={}): | |
| result = service.get_system_info() | |
| assert result['config_file'] == "/tmp/config.json" | |
| assert result['temp_directory'] == "/tmp" | |
| assert result['log_level'] == "INFO" | |
| assert 'supported_languages' in result | |
| assert 'supported_audio_formats' in result | |
| assert 'max_file_size_mb' in result | |
| def test_validate_configuration_success(self, service, mock_config): | |
| """Test successful configuration validation""" | |
| # Mock valid configuration | |
| mock_config.get_tts_config.return_value = { | |
| 'default_speed': 1.0, | |
| 'max_text_length': 5000 | |
| } | |
| mock_config.get_stt_config.return_value = { | |
| 'chunk_length_s': 30, | |
| 'batch_size': 16 | |
| } | |
| mock_config.get_processing_config.return_value = { | |
| 'temp_dir': '/tmp', | |
| 'max_file_size_mb': 100 | |
| } | |
| mock_config.get_logging_config.return_value = { | |
| 'level': 'INFO' | |
| } | |
| with patch('os.path.exists', return_value=True): | |
| result = service.validate_configuration() | |
| # Should have no issues | |
| assert all(len(issues) == 0 for issues in result.values()) | |
| def test_validate_configuration_with_issues(self, service, mock_config): | |
| """Test configuration validation with issues""" | |
| # Mock invalid configuration | |
| mock_config.get_tts_config.return_value = { | |
| 'default_speed': 5.0, # Invalid | |
| 'max_text_length': -100 # Invalid | |
| } | |
| mock_config.get_stt_config.return_value = { | |
| 'chunk_length_s': -10, # Invalid | |
| 'batch_size': 0 # Invalid | |
| } | |
| mock_config.get_processing_config.return_value = { | |
| 'temp_dir': '/nonexistent', # Invalid | |
| 'max_file_size_mb': -50 # Invalid | |
| } | |
| mock_config.get_logging_config.return_value = { | |
| 'level': 'INVALID' # Invalid | |
| } | |
| with patch('os.path.exists', return_value=False): | |
| result = service.validate_configuration() | |
| # Should have issues in each category | |
| assert len(result['tts']) > 0 | |
| assert len(result['stt']) > 0 | |
| assert len(result['processing']) > 0 | |
| assert len(result['logging']) > 0 | |
| def test_reset_to_defaults_success(self, service, mock_container): | |
| """Test successful configuration reset to defaults""" | |
| with patch('src.infrastructure.config.app_config.AppConfig') as mock_app_config: | |
| default_config = Mock() | |
| mock_app_config.return_value = default_config | |
| result = service.reset_to_defaults() | |
| # Verify new default config was created and registered | |
| mock_app_config.assert_called_once_with() | |
| mock_container.register_singleton.assert_called_once_with(AppConfig, default_config) | |
| def test_reset_to_defaults_failure(self, service): | |
| """Test configuration reset to defaults failure""" | |
| with patch('src.infrastructure.config.app_config.AppConfig', side_effect=Exception("Reset failed")): | |
| with pytest.raises(ConfigurationException, match="Failed to reset configuration"): | |
| service.reset_to_defaults() | |
| def test_cleanup(self, service): | |
| """Test service cleanup""" | |
| # Should not raise exception | |
| service.cleanup() | |
| 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() |