nifty-coder/stemsplit-backend

1"""2Integration tests for WebSpeechProvider with the voice control system.3 4This module tests the integration of WebSpeechProvider with other components5like ProviderManager, RateLimiter, and CacheManager.6"""7 8import pytest9import asyncio10from unittest.mock import AsyncMock, MagicMock, patch11 12from voice_control.providers.web_speech_provider import create_web_speech_provider13from voice_control.provider_manager import ProviderManager14from voice_control.rate_limiter import RateLimiter15from voice_control.cache_manager import CacheManager16from voice_control.models import ProviderConfig, ProviderType, TranscriptionResult17from voice_control.exceptions import ProviderUnavailableError, QuotaExceededError18 19 20class TestWebSpeechProviderIntegration:21    """Integration tests for WebSpeechProvider."""22    23    @pytest.fixture24    def web_speech_provider(self):25        """Create a WebSpeechProvider for testing."""26        return create_web_speech_provider(27            name="test_web_speech",28            priority=129        )30    31    @pytest.fixture32    def provider_manager(self):33        """Create a ProviderManager for testing."""34        return ProviderManager()35    36    @pytest.fixture37    def sample_audio_data(self):38        """Create sample audio data for testing."""39        return b'\x00' * (16000 * 2)  # 1 second of 16kHz, 16-bit audio40    41    @pytest.mark.asyncio42    async def test_provider_registration_with_manager(self, provider_manager, web_speech_provider):43        """Test registering WebSpeechProvider with ProviderManager."""44        # Register the provider45        await provider_manager.register_provider(web_speech_provider)46        47        # Check that provider is registered48        status = await provider_manager.get_provider_status()49        assert "test_web_speech" in status50        51        provider_status = status["test_web_speech"]52        assert provider_status.name == "test_web_speech"53        assert provider_status.available is True54    55    @pytest.mark.asyncio56    async def test_transcription_through_manager(self, provider_manager, web_speech_provider, sample_audio_data):57        """Test transcription through ProviderManager using WebSpeechProvider."""58        # Mock browser compatibility59        with patch.object(web_speech_provider, '_check_browser_compatibility'):60            web_speech_provider._browser_compatible = True61            62            # Register provider63            await provider_manager.register_provider(web_speech_provider)64            65            # Perform transcription through manager66            result = await provider_manager.transcribe_audio(67                sample_audio_data,68                "webm",69                "en-US"70            )71            72            # Verify result73            assert isinstance(result, TranscriptionResult)74            assert result.provider == "test_web_speech"75            assert result.text is not None76            assert result.confidence > 077    78    @pytest.mark.asyncio79    async def test_provider_health_monitoring(self, provider_manager, web_speech_provider):80        """Test health monitoring of WebSpeechProvider."""81        # Mock browser compatibility check82        with patch.object(web_speech_provider, '_check_browser_compatibility') as mock_check:83            mock_check.return_value = None84            web_speech_provider._browser_compatible = True85            86            # Register provider87            await provider_manager.register_provider(web_speech_provider)88            89            # Check health through manager90            status = await provider_manager.get_provider_status()91            provider_status = status["test_web_speech"]92            93            assert provider_status.available is True94            assert provider_status.response_time_avg >= 095    96    @pytest.mark.asyncio97    async def test_provider_fallback_behavior(self, provider_manager, sample_audio_data):98        """Test fallback behavior when WebSpeechProvider fails."""99        # Create two providers with different priorities100        primary_provider = create_web_speech_provider(101            name="primary_web_speech",102            priority=1103        )104        fallback_provider = create_web_speech_provider(105            name="fallback_web_speech", 106            priority=2107        )108        109        # Mock primary provider to fail110        with patch.object(primary_provider, 'transcribe_audio') as mock_primary, \111             patch.object(fallback_provider, 'transcribe_audio') as mock_fallback, \112             patch.object(primary_provider, '_check_browser_compatibility'), \113             patch.object(fallback_provider, '_check_browser_compatibility'):114            115            # Set up browser compatibility116            primary_provider._browser_compatible = True117            fallback_provider._browser_compatible = True118            119            # Primary fails, fallback succeeds120            mock_primary.side_effect = ProviderUnavailableError(121                "primary_web_speech", 122                "Browser not supported"123            )124            mock_fallback.return_value = TranscriptionResult(125                text="Fallback transcription",126                confidence=0.8,127                provider="fallback_web_speech",128                processing_time=0.1,129                audio_duration=1.0,130                language="en-US"131            )132            133            # Register both providers134            await provider_manager.register_provider(primary_provider)135            await provider_manager.register_provider(fallback_provider)136            137            # Perform transcription - should use fallback138            result = await provider_manager.transcribe_audio(139                sample_audio_data,140                "webm",141                "en-US"142            )143            144            # Verify fallback was used145            assert result.provider == "fallback_web_speech"146            assert result.text == "Fallback transcription"147    148    @pytest.mark.asyncio149    async def test_streaming_through_manager(self, provider_manager, web_speech_provider):150        """Test streaming transcription through ProviderManager."""151        async def mock_audio_stream():152            """Mock audio stream."""153            for i in range(3):154                yield b'\x00' * 8000  # 0.5 seconds each155        156        with patch.object(web_speech_provider, '_check_browser_compatibility'):157            web_speech_provider._browser_compatible = True158            159            # Register provider160            await provider_manager.register_provider(web_speech_provider)161            162            # Perform streaming transcription163            results = []164            async for result in provider_manager.transcribe_streaming(165                mock_audio_stream(),166                "webm",167                "en-US"168            ):169                results.append(result)170            171            # Verify results172            assert len(results) > 0173            assert all(r.provider == "test_web_speech" for r in results)174    175    @pytest.mark.asyncio176    async def test_quota_management_integration(self, web_speech_provider, sample_audio_data):177        """Test quota management with WebSpeechProvider."""178        # Set very low quota limits179        web_speech_provider.config.free_tier_limits = {180            "max_daily_requests": 1,181            "max_daily_minutes": 0.1182        }183        184        with patch.object(web_speech_provider, '_check_browser_compatibility'):185            web_speech_provider._browser_compatible = True186            187            # First transcription should succeed188            result1 = await web_speech_provider.transcribe_audio(189                sample_audio_data,190                "webm", 191                "en-US",192                audio_duration=0.05  # 0.05 seconds193            )194            assert result1 is not None195            196            # Second transcription should fail due to quota197            with pytest.raises(QuotaExceededError):198                await web_speech_provider.transcribe_audio(199                    sample_audio_data,200                    "webm",201                    "en-US", 202                    audio_duration=0.05203                )204    205    @pytest.mark.asyncio206    async def test_format_and_language_validation(self, provider_manager, web_speech_provider, sample_audio_data):207        """Test format and language validation through manager."""208        with patch.object(web_speech_provider, '_check_browser_compatibility'):209            web_speech_provider._browser_compatible = True210            211            # Register provider212            await provider_manager.register_provider(web_speech_provider)213            214            # Test unsupported format215            with pytest.raises(Exception):  # Should raise UnsupportedFormatError or similar216                await provider_manager.transcribe_audio(217                    sample_audio_data,218                    "unsupported_format",219                    "en-US"220                )221            222            # Test unsupported language223            with pytest.raises(Exception):  # Should raise UnsupportedLanguageError or similar224                await provider_manager.transcribe_audio(225                    sample_audio_data,226                    "webm",227                    "unsupported-LANG"228                )229    230    @pytest.mark.asyncio231    async def test_provider_metrics_collection(self, provider_manager, web_speech_provider, sample_audio_data):232        """Test that provider metrics are collected correctly."""233        with patch.object(web_speech_provider, '_check_browser_compatibility'):234            web_speech_provider._browser_compatible = True235            236            # Register provider237            await provider_manager.register_provider(web_speech_provider)238            239            # Perform several transcriptions240            for i in range(3):241                await provider_manager.transcribe_audio(242                    sample_audio_data,243                    "webm",244                    "en-US"245                )246            247            # Check that usage stats were updated248            quota_status = await web_speech_provider.get_quota_status()249            requests_quota = quota_status["requests_per_day"]250            audio_quota = quota_status["audio_minutes_per_day"]251            252            assert requests_quota.current_usage == 3253            # Audio usage should be greater than 0 since we processed audio254            assert audio_quota.current_usage >= 0  # Changed from > 0 to >= 0 since it might be very small255    256    @pytest.mark.asyncio257    async def test_concurrent_transcriptions(self, provider_manager, web_speech_provider, sample_audio_data):258        """Test concurrent transcriptions with WebSpeechProvider."""259        with patch.object(web_speech_provider, '_check_browser_compatibility'):260            web_speech_provider._browser_compatible = True261            262            # Register provider263            await provider_manager.register_provider(web_speech_provider)264            265            # Perform concurrent transcriptions266            tasks = []267            for i in range(5):268                task = provider_manager.transcribe_audio(269                    sample_audio_data,270                    "webm",271                    "en-US"272                )273                tasks.append(task)274            275            # Wait for all to complete276            results = await asyncio.gather(*tasks)277            278            # Verify all succeeded279            assert len(results) == 5280            assert all(isinstance(r, TranscriptionResult) for r in results)281            assert all(r.provider == "test_web_speech" for r in results)282 283 284class TestWebSpeechProviderSystemIntegration:285    """System-level integration tests."""286    287    @pytest.mark.asyncio288    async def test_complete_voice_control_workflow(self):289        """Test complete workflow with WebSpeechProvider as primary provider."""290        # Create a complete system setup291        provider_manager = ProviderManager()292        web_speech_provider = create_web_speech_provider(293            name="system_web_speech",294            priority=1  # Highest priority295        )296        297        sample_audio = b'\x00' * (16000 * 2)  # 1 second298        299        with patch.object(web_speech_provider, '_check_browser_compatibility'):300            web_speech_provider._browser_compatible = True301            302            # Register provider303            await provider_manager.register_provider(web_speech_provider)304            305            # 1. Check system health306            status = await provider_manager.get_provider_status()307            assert "system_web_speech" in status308            assert status["system_web_speech"].available is True309            310            # 2. Perform transcription311            result = await provider_manager.transcribe_audio(312                sample_audio,313                "webm",314                "en-US"315            )316            317            # 3. Verify result quality318            assert result.provider == "system_web_speech"319            assert result.confidence > 0320            assert result.processing_time > 0321            assert result.audio_duration > 0322            323            # 4. Check quota usage324            quota_status = await web_speech_provider.get_quota_status()325            assert quota_status["requests_per_day"].current_usage == 1326            327            # 5. Test streaming capability328            async def audio_stream():329                for _ in range(2):330                    yield b'\x00' * 8000331            332            streaming_results = []333            async for stream_result in provider_manager.transcribe_streaming(334                audio_stream(),335                "webm",336                "en-US"337            ):338                streaming_results.append(stream_result)339            340            assert len(streaming_results) > 0341            assert all(r.provider == "system_web_speech" for r in streaming_results)342            343            print("✓ Complete voice control workflow test passed")344 345 346if __name__ == "__main__":347    # Run a quick integration test348    async def quick_test():349        test_instance = TestWebSpeechProviderSystemIntegration()350        await test_instance.test_complete_voice_control_workflow()351        print("Integration test completed successfully!")352    353    asyncio.run(quick_test())