nifty-coder/stemsplit-backend

1"""2Integration tests for GoogleSpeechProvider with the provider manager system.3 4This module tests the integration of GoogleSpeechProvider with other components5in the voice control system.6"""7 8import pytest9from unittest.mock import Mock, patch10import asyncio11 12from voice_control.providers.google_speech_provider import (13    GoogleSpeechProvider, 14    create_google_speech_provider15)16from voice_control.provider_manager import ProviderManager17from voice_control.models import (18    ProviderConfig, 19    ProviderType, 20    TranscriptionResult21)22 23 24class TestGoogleSpeechProviderIntegration:25    """Integration tests for GoogleSpeechProvider."""26    27    @pytest.fixture28    def mock_google_speech(self):29        """Mock Google Cloud Speech modules."""30        with patch('voice_control.providers.google_speech_provider.GOOGLE_SPEECH_AVAILABLE', True):31            with patch('voice_control.providers.google_speech_provider.speech') as mock_speech:32                with patch('voice_control.providers.google_speech_provider.service_account') as mock_sa:33                    # Setup mock speech client34                    mock_client = Mock()35                    mock_speech.SpeechClient.return_value = mock_client36                    37                    # Setup mock recognition config38                    mock_config = Mock()39                    mock_speech.RecognitionConfig.return_value = mock_config40                    mock_speech.RecognitionConfig.AudioEncoding = Mock()41                    mock_speech.RecognitionConfig.AudioEncoding.LINEAR16 = "LINEAR16"42                    43                    # Setup mock audio44                    mock_audio = Mock()45                    mock_speech.RecognitionAudio.return_value = mock_audio46                    47                    yield {48                        'speech': mock_speech,49                        'service_account': mock_sa,50                        'client': mock_client51                    }52    53    @pytest.mark.asyncio54    async def test_provider_manager_integration(self, mock_google_speech):55        """Test GoogleSpeechProvider integration with ProviderManager."""56        # Create Google Speech provider57        provider = create_google_speech_provider(58            name="test_google_speech",59            priority=2,60            api_credentials={"service_account_key": "test_key"}61        )62        63        # Create provider manager64        manager = ProviderManager()65        66        # Register the provider67        await manager.register_provider(provider)68        69        # Mock successful transcription70        mock_alternative = Mock()71        mock_alternative.transcript = "Hello from Google Speech"72        mock_alternative.confidence = 0.9573        mock_alternative.words = []74        75        mock_result = Mock()76        mock_result.alternatives = [mock_alternative]77        78        mock_response = Mock()79        mock_response.results = [mock_result]80        81        mock_google_speech['client'].recognize.return_value = mock_response82        83        # Test transcription through provider manager84        audio_data = b'\x00\x01' * 100085        result = await manager.transcribe_audio(audio_data, "wav", "en-US")86        87        # Verify result88        assert isinstance(result, TranscriptionResult)89        assert result.text == "Hello from Google Speech"90        assert result.confidence == 0.9591        assert result.provider == "test_google_speech"92        assert result.language == "en-US"93    94    @pytest.mark.asyncio95    async def test_quota_management_integration(self, mock_google_speech):96        """Test quota management integration."""97        provider = create_google_speech_provider(98            api_credentials={"service_account_key": "test"}99        )100        101        # Mock successful responses102        mock_alternative = Mock()103        mock_alternative.transcript = "Test"104        mock_alternative.confidence = 0.9105        mock_alternative.words = []106        107        mock_result = Mock()108        mock_result.alternatives = [mock_alternative]109        110        mock_response = Mock()111        mock_response.results = [mock_result]112        113        mock_google_speech['client'].recognize.return_value = mock_response114        115        # Make multiple requests to test quota tracking116        audio_data = b'\x00\x01' * 500117        for i in range(3):118            result = await provider.transcribe_audio(audio_data, "wav", "en-US")119            assert result.text == "Test"120        121        # Check quota status122        quota_status = await provider.get_quota_status()123        124        # Verify quota tracking125        assert quota_status["requests_per_minute"].current_usage == 3126        assert quota_status["audio_minutes_per_month"].current_usage > 0127        128        # Verify cost estimation129        cost = await provider.estimate_cost(300)  # 5 minutes130        assert cost == 0.0  # Should be free within tier131    132    @pytest.mark.asyncio133    async def test_error_handling_integration(self, mock_google_speech):134        """Test error handling integration."""135        provider = create_google_speech_provider(136            api_credentials={"service_account_key": "test"}137        )138        139        # Test unsupported format140        with pytest.raises(Exception):  # Should raise UnsupportedFormatError141            await provider.transcribe_audio(b'test', "unsupported_format")142        143        # Test unsupported language144        with pytest.raises(Exception):  # Should raise UnsupportedLanguageError145            await provider.transcribe_audio(b'test', "wav", "xx-XX")146    147    @pytest.mark.asyncio148    async def test_health_check_integration(self, mock_google_speech):149        """Test health check integration."""150        provider = create_google_speech_provider(151            api_credentials={"service_account_key": "test"}152        )153        154        # Mock successful health check155        mock_response = Mock()156        mock_google_speech['client'].recognize.return_value = mock_response157        158        health = await provider.check_health()159        assert health is True160        161        # Mock failed health check162        mock_google_speech['client'].recognize.side_effect = Exception("API Error")163        164        health = await provider.check_health()165        assert health is False166    167    def test_factory_function(self, mock_google_speech):168        """Test the factory function creates provider correctly."""169        provider = create_google_speech_provider(170            name="custom_google",171            priority=5,172            api_credentials={"service_account_key": "test_key"}173        )174        175        assert provider.name == "custom_google"176        assert provider.config.priority == 5177        assert provider.config.api_credentials["service_account_key"] == "test_key"178        assert provider.config.provider_type == ProviderType.GOOGLE_SPEECH179        assert provider.supports_format("wav")180        assert provider.supports_language("en-US")