nifty-coder/stemsplit-backend
0
1"""2Unit tests for GoogleSpeechProvider.3 4This module contains comprehensive tests for the Google Speech-to-Text provider5implementation, including quota management, error handling, and transcription6functionality.7"""8 9import pytest10import asyncio11from unittest.mock import Mock, AsyncMock, patch, MagicMock12from datetime import datetime, timezone, timedelta13import json14 15from voice_control.providers.google_speech_provider import (16 GoogleSpeechProvider, 17 create_google_speech_provider,18 GOOGLE_SPEECH_AVAILABLE19)20from voice_control.models import (21 ProviderConfig, 22 ProviderType, 23 TranscriptionResult,24 QuotaStatus,25 QuotaType,26 WordTimestamp27)28from voice_control.exceptions import (29 ProviderError,30 ProviderUnavailableError,31 ProviderAuthenticationError,32 UnsupportedFormatError,33 UnsupportedLanguageError,34 QuotaExceededError,35 RateLimitExceededError,36 TranscriptionError37)38 39 40class TestGoogleSpeechProvider:41 """Test cases for GoogleSpeechProvider class."""42 43 @pytest.fixture44 def mock_google_speech(self):45 """Mock Google Cloud Speech modules."""46 with patch('voice_control.providers.google_speech_provider.GOOGLE_SPEECH_AVAILABLE', True):47 with patch('voice_control.providers.google_speech_provider.speech') as mock_speech:48 with patch('voice_control.providers.google_speech_provider.service_account') as mock_sa:49 with patch('voice_control.providers.google_speech_provider.google_exceptions') as mock_exc:50 # Setup mock speech client51 mock_client = Mock()52 mock_speech.SpeechClient.return_value = mock_client53 54 # Setup mock recognition config55 mock_config = Mock()56 mock_speech.RecognitionConfig.return_value = mock_config57 mock_speech.RecognitionConfig.AudioEncoding = Mock()58 mock_speech.RecognitionConfig.AudioEncoding.LINEAR16 = "LINEAR16"59 mock_speech.RecognitionConfig.AudioEncoding.FLAC = "FLAC"60 mock_speech.RecognitionConfig.AudioEncoding.MP3 = "MP3"61 mock_speech.RecognitionConfig.AudioEncoding.WEBM_OPUS = "WEBM_OPUS"62 63 # Setup mock audio64 mock_audio = Mock()65 mock_speech.RecognitionAudio.return_value = mock_audio66 67 # Setup mock streaming config68 mock_streaming_config = Mock()69 mock_speech.StreamingRecognitionConfig.return_value = mock_streaming_config70 71 # Setup mock streaming request72 mock_streaming_request = Mock()73 mock_speech.StreamingRecognizeRequest.return_value = mock_streaming_request74 75 yield {76 'speech': mock_speech,77 'service_account': mock_sa,78 'exceptions': mock_exc,79 'client': mock_client80 }81 82 @pytest.fixture83 def provider_config(self):84 """Create a test provider configuration."""85 return ProviderConfig(86 name="test_google_speech",87 provider_type=ProviderType.GOOGLE_SPEECH,88 enabled=True,89 priority=2,90 free_tier_limits={91 "audio_minutes_per_month": 60.0,92 "requests_per_minute": 1000,93 "requests_per_day": 5000094 },95 rate_limits={96 "requests_per_minute": 1000,97 "requests_per_day": 5000098 },99 supported_formats=["wav", "flac", "mp3", "webm"],100 supported_languages=["en-US", "es-ES", "fr-FR"],101 cost_per_minute=0.006,102 api_credentials={103 "service_account_key": '{"type": "service_account", "project_id": "test"}'104 },105 timeout_seconds=60,106 max_retries=3107 )108 109 @pytest.fixture110 def sample_audio_data(self):111 """Create sample audio data for testing."""112 return b'\x00\x01' * 8000 # 16KB of sample audio data113 114 def test_provider_initialization_without_library(self):115 """Test provider initialization when Google Speech library is not available."""116 with patch('voice_control.providers.google_speech_provider.GOOGLE_SPEECH_AVAILABLE', False):117 config = ProviderConfig(118 name="test",119 provider_type=ProviderType.GOOGLE_SPEECH,120 enabled=True,121 priority=1,122 free_tier_limits={},123 rate_limits={},124 supported_formats=["wav"],125 supported_languages=["en-US"],126 cost_per_minute=0.0,127 api_credentials={}128 )129 130 with pytest.raises(ImportError, match="Google Cloud Speech library not available"):131 GoogleSpeechProvider(config)132 133 def test_provider_initialization_invalid_type(self, mock_google_speech):134 """Test provider initialization with invalid provider type."""135 config = ProviderConfig(136 name="invalid",137 provider_type=ProviderType.WEB_SPEECH_API, # Wrong type138 enabled=True,139 priority=1,140 free_tier_limits={},141 rate_limits={},142 supported_formats=["wav"],143 supported_languages=["en-US"],144 cost_per_minute=0.0,145 api_credentials={}146 )147 148 with pytest.raises(ValueError, match="Invalid provider type"):149 GoogleSpeechProvider(config)150 151 @pytest.mark.asyncio152 async def test_provider_initialization_success(self, mock_google_speech, provider_config):153 """Test successful provider initialization."""154 provider = GoogleSpeechProvider(provider_config)155 156 # Wait for initialization to complete157 await asyncio.sleep(0.1)158 159 assert provider.name == "test_google_speech"160 assert provider.provider_type == ProviderType.GOOGLE_SPEECH161 assert provider.config == provider_config162 163 def test_supports_format(self, mock_google_speech, provider_config):164 """Test format support checking."""165 provider = GoogleSpeechProvider(provider_config)166 167 # Test supported formats168 assert provider.supports_format("wav")169 assert provider.supports_format("WAV") # Case insensitive170 assert provider.supports_format("flac")171 assert provider.supports_format("mp3")172 assert provider.supports_format("webm")173 174 # Test unsupported formats175 assert not provider.supports_format("aac")176 assert not provider.supports_format("m4a")177 assert not provider.supports_format("unknown")178 179 def test_supports_language(self, mock_google_speech, provider_config):180 """Test language support checking."""181 provider = GoogleSpeechProvider(provider_config)182 183 # Test supported languages184 assert provider.supports_language("en-US")185 assert provider.supports_language("es-ES")186 assert provider.supports_language("fr-FR")187 assert provider.supports_language("de-DE")188 assert provider.supports_language("ja-JP")189 190 # Test unsupported languages191 assert not provider.supports_language("xx-XX")192 assert not provider.supports_language("invalid")193 194 @pytest.mark.asyncio195 async def test_estimate_cost_free_tier(self, mock_google_speech, provider_config):196 """Test cost estimation within free tier."""197 provider = GoogleSpeechProvider(provider_config)198 199 # Test within free tier (60 minutes/month)200 cost = await provider.estimate_cost(1800) # 30 minutes201 assert cost == 0.0202 203 # Test at free tier limit204 cost = await provider.estimate_cost(3600) # 60 minutes205 assert cost == 0.0206 207 @pytest.mark.asyncio208 async def test_estimate_cost_exceeding_free_tier(self, mock_google_speech, provider_config):209 """Test cost estimation exceeding free tier."""210 provider = GoogleSpeechProvider(provider_config)211 212 # Simulate some usage213 provider._usage_stats["audio_minutes_this_month"] = 50.0 # 50 minutes used214 215 # Test exceeding free tier216 cost = await provider.estimate_cost(1800) # 30 minutes (20 minutes over limit)217 expected_cost = 20.0 * 0.006 # 20 minutes at $0.006/minute218 assert abs(cost - expected_cost) < 0.001219 220 @pytest.mark.asyncio221 async def test_check_health_success(self, mock_google_speech, provider_config):222 """Test successful health check."""223 provider = GoogleSpeechProvider(provider_config)224 225 # Mock successful recognition226 mock_response = Mock()227 mock_google_speech['client'].recognize.return_value = mock_response228 229 health = await provider.check_health()230 assert health is True231 232 @pytest.mark.asyncio233 async def test_check_health_failure(self, mock_google_speech, provider_config):234 """Test health check failure."""235 provider = GoogleSpeechProvider(provider_config)236 237 # Mock recognition failure238 mock_google_speech['client'].recognize.side_effect = Exception("API Error")239 240 health = await provider.check_health()241 assert health is False242 243 @pytest.mark.asyncio244 async def test_get_quota_status(self, mock_google_speech, provider_config):245 """Test quota status retrieval."""246 provider = GoogleSpeechProvider(provider_config)247 248 # Set some usage249 provider._usage_stats["requests_this_minute"] = 10250 provider._usage_stats["requests_today"] = 100251 provider._usage_stats["audio_minutes_this_month"] = 15.5252 253 quota_status = await provider.get_quota_status()254 255 assert "requests_per_minute" in quota_status256 assert "requests_per_day" in quota_status257 assert "audio_minutes_per_month" in quota_status258 259 # Check requests per minute260 rpm_status = quota_status["requests_per_minute"]261 assert rpm_status.provider == provider.name262 assert rpm_status.quota_type == QuotaType.REQUESTS_PER_MINUTE263 assert rpm_status.current_usage == 10264 assert rpm_status.limit == 1000265 266 # Check audio minutes per month267 ampm_status = quota_status["audio_minutes_per_month"]268 assert ampm_status.current_usage == 15.5269 assert ampm_status.limit == 60.0270 assert abs(ampm_status.percentage_used - (15.5 / 60.0)) < 0.001271 272 @pytest.mark.asyncio273 async def test_transcribe_audio_unsupported_format(self, mock_google_speech, provider_config, sample_audio_data):274 """Test transcription with unsupported format."""275 provider = GoogleSpeechProvider(provider_config)276 277 with pytest.raises(UnsupportedFormatError):278 await provider.transcribe_audio(sample_audio_data, "aac")279 280 @pytest.mark.asyncio281 async def test_transcribe_audio_unsupported_language(self, mock_google_speech, provider_config, sample_audio_data):282 """Test transcription with unsupported language."""283 provider = GoogleSpeechProvider(provider_config)284 285 with pytest.raises(UnsupportedLanguageError):286 await provider.transcribe_audio(sample_audio_data, "wav", "xx-XX")287 288 @pytest.mark.asyncio289 async def test_transcribe_audio_quota_exceeded(self, mock_google_speech, provider_config, sample_audio_data):290 """Test transcription when quota is exceeded."""291 provider = GoogleSpeechProvider(provider_config)292 293 # Set usage to exceed monthly limit294 provider._usage_stats["audio_minutes_this_month"] = 59.5 # Close to 60 minute limit295 296 with pytest.raises(QuotaExceededError):297 await provider.transcribe_audio(sample_audio_data, "wav", audio_duration=120) # 2 minutes298 299 @pytest.mark.asyncio300 async def test_transcribe_audio_rate_limit_exceeded(self, mock_google_speech, provider_config, sample_audio_data):301 """Test transcription when rate limit is exceeded."""302 provider = GoogleSpeechProvider(provider_config)303 304 # Set usage to exceed per-minute limit305 provider._usage_stats["requests_this_minute"] = 1000306 307 with pytest.raises(RateLimitExceededError):308 await provider.transcribe_audio(sample_audio_data, "wav")309 310 @pytest.mark.asyncio311 async def test_transcribe_audio_success(self, mock_google_speech, provider_config, sample_audio_data):312 """Test successful audio transcription."""313 provider = GoogleSpeechProvider(provider_config)314 315 # Mock successful recognition response316 mock_alternative = Mock()317 mock_alternative.transcript = "Hello world"318 mock_alternative.confidence = 0.95319 mock_alternative.words = []320 321 mock_result = Mock()322 mock_result.alternatives = [mock_alternative]323 324 mock_response = Mock()325 mock_response.results = [mock_result]326 327 mock_google_speech['client'].recognize.return_value = mock_response328 329 result = await provider.transcribe_audio(sample_audio_data, "wav", "en-US")330 331 assert isinstance(result, TranscriptionResult)332 assert result.text == "Hello world"333 assert result.confidence == 0.95334 assert result.provider == provider.name335 assert result.language == "en-US"336 assert result.is_final is True337 338 @pytest.mark.asyncio339 async def test_transcribe_audio_with_word_timestamps(self, mock_google_speech, provider_config, sample_audio_data):340 """Test transcription with word timestamps."""341 provider = GoogleSpeechProvider(provider_config)342 343 # Mock word info344 mock_word1 = Mock()345 mock_word1.word = "Hello"346 mock_word1.start_time = Mock()347 mock_word1.start_time.total_seconds.return_value = 0.0348 mock_word1.end_time = Mock()349 mock_word1.end_time.total_seconds.return_value = 0.5350 mock_word1.confidence = 0.98351 352 mock_word2 = Mock()353 mock_word2.word = "world"354 mock_word2.start_time = Mock()355 mock_word2.start_time.total_seconds.return_value = 0.6356 mock_word2.end_time = Mock()357 mock_word2.end_time.total_seconds.return_value = 1.0358 mock_word2.confidence = 0.92359 360 mock_alternative = Mock()361 mock_alternative.transcript = "Hello world"362 mock_alternative.confidence = 0.95363 mock_alternative.words = [mock_word1, mock_word2]364 365 mock_result = Mock()366 mock_result.alternatives = [mock_alternative]367 368 mock_response = Mock()369 mock_response.results = [mock_result]370 371 mock_google_speech['client'].recognize.return_value = mock_response372 373 result = await provider.transcribe_audio(374 sample_audio_data, 375 "wav", 376 "en-US",377 enable_word_time_offsets=True378 )379 380 assert result.word_timestamps is not None381 assert len(result.word_timestamps) == 2382 383 word1 = result.word_timestamps[0]384 assert word1.word == "Hello"385 assert word1.start_time == 0.0386 assert word1.end_time == 0.5387 assert word1.confidence == 0.98388 389 word2 = result.word_timestamps[1]390 assert word2.word == "world"391 assert word2.start_time == 0.6392 assert word2.end_time == 1.0393 assert word2.confidence == 0.92394 395 @pytest.mark.asyncio396 async def test_transcribe_audio_empty_response(self, mock_google_speech, provider_config, sample_audio_data):397 """Test transcription with empty response."""398 provider = GoogleSpeechProvider(provider_config)399 400 # Mock empty response401 mock_response = Mock()402 mock_response.results = []403 404 mock_google_speech['client'].recognize.return_value = mock_response405 406 result = await provider.transcribe_audio(sample_audio_data, "wav", "en-US")407 408 assert result.text == ""409 assert result.confidence == 0.0410 assert result.provider == provider.name411 412 @pytest.mark.asyncio413 async def test_transcribe_audio_google_api_error(self, mock_google_speech, provider_config, sample_audio_data):414 """Test transcription with Google API error."""415 provider = GoogleSpeechProvider(provider_config)416 417 # Mock Google API error418 mock_error = Mock()419 mock_error.code = 401420 mock_google_speech['client'].recognize.side_effect = mock_error421 422 with pytest.raises(ProviderAuthenticationError):423 await provider.transcribe_audio(sample_audio_data, "wav", "en-US")424 425 @pytest.mark.asyncio426 async def test_transcribe_streaming_unsupported_format(self, mock_google_speech, provider_config):427 """Test streaming transcription with unsupported format."""428 provider = GoogleSpeechProvider(provider_config)429 430 async def mock_audio_stream():431 yield b'audio_chunk'432 433 with pytest.raises(UnsupportedFormatError):434 async for _ in provider.transcribe_streaming(mock_audio_stream(), "aac"):435 pass436 437 @pytest.mark.asyncio438 async def test_transcribe_streaming_success(self, mock_google_speech, provider_config):439 """Test successful streaming transcription."""440 provider = GoogleSpeechProvider(provider_config)441 442 # Mock streaming response443 mock_alternative = Mock()444 mock_alternative.transcript = "Streaming text"445 mock_alternative.confidence = 0.85446 447 mock_result = Mock()448 mock_result.alternatives = [mock_alternative]449 mock_result.is_final = True450 451 mock_response = Mock()452 mock_response.results = [mock_result]453 454 mock_google_speech['client'].streaming_recognize.return_value = [mock_response]455 456 async def mock_audio_stream():457 yield b'audio_chunk_1'458 yield b'audio_chunk_2'459 460 results = []461 async for result in provider.transcribe_streaming(mock_audio_stream(), "wav", "en-US"):462 results.append(result)463 464 assert len(results) == 1465 assert results[0].text == "Streaming text"466 assert results[0].confidence == 0.85467 assert results[0].is_final is True468 469 @pytest.mark.asyncio470 async def test_usage_stats_reset(self, mock_google_speech, provider_config):471 """Test usage statistics reset functionality."""472 provider = GoogleSpeechProvider(provider_config)473 474 # Set usage stats to old values475 old_time = datetime.now(timezone.utc) - timedelta(days=2)476 provider._usage_stats["last_daily_reset"] = old_time.date()477 provider._usage_stats["requests_today"] = 100478 479 # Trigger reset by calling _reset_counters_if_needed480 await provider._reset_counters_if_needed()481 482 # Check that daily counter was reset483 assert provider._usage_stats["requests_today"] == 0484 assert provider._usage_stats["last_daily_reset"] == datetime.now(timezone.utc).date()485 486 def test_create_google_speech_provider_factory(self, mock_google_speech):487 """Test the factory function for creating GoogleSpeechProvider."""488 provider = create_google_speech_provider(489 name="custom_google",490 priority=5,491 api_credentials={"service_account_key": "test_key"}492 )493 494 assert provider.name == "custom_google"495 assert provider.config.priority == 5496 assert provider.config.api_credentials["service_account_key"] == "test_key"497 assert provider.config.provider_type == ProviderType.GOOGLE_SPEECH498 499 500class TestGoogleSpeechProviderIntegration:501 """Integration tests for GoogleSpeechProvider."""502 503 @pytest.fixture504 def mock_google_speech_integration(self):505 """Mock Google Cloud Speech modules for integration tests."""506 with patch('voice_control.providers.google_speech_provider.GOOGLE_SPEECH_AVAILABLE', True):507 with patch('voice_control.providers.google_speech_provider.speech') as mock_speech:508 with patch('voice_control.providers.google_speech_provider.service_account') as mock_sa:509 with patch('voice_control.providers.google_speech_provider.google_exceptions') as mock_exc:510 # Setup mock speech client511 mock_client = Mock()512 mock_speech.SpeechClient.return_value = mock_client513 514 # Setup mock recognition config515 mock_config = Mock()516 mock_speech.RecognitionConfig.return_value = mock_config517 mock_speech.RecognitionConfig.AudioEncoding = Mock()518 mock_speech.RecognitionConfig.AudioEncoding.LINEAR16 = "LINEAR16"519 520 # Setup mock audio521 mock_audio = Mock()522 mock_speech.RecognitionAudio.return_value = mock_audio523 524 yield {525 'speech': mock_speech,526 'service_account': mock_sa,527 'exceptions': mock_exc,528 'client': mock_client529 }530 531 @pytest.mark.asyncio532 async def test_full_transcription_workflow(self, mock_google_speech_integration):533 """Test complete transcription workflow."""534 # Create provider with realistic config535 config = ProviderConfig(536 name="integration_test",537 provider_type=ProviderType.GOOGLE_SPEECH,538 enabled=True,539 priority=2,540 free_tier_limits={541 "audio_minutes_per_month": 60.0,542 "requests_per_minute": 1000,543 "requests_per_day": 50000544 },545 rate_limits={546 "requests_per_minute": 1000,547 "requests_per_day": 50000548 },549 supported_formats=["wav", "mp3", "flac"],550 supported_languages=["en-US", "es-ES"],551 cost_per_minute=0.006,552 api_credentials={553 "service_account_key": '{"type": "service_account", "project_id": "test"}'554 }555 )556 557 provider = GoogleSpeechProvider(config)558 559 # Mock successful transcription560 mock_alternative = Mock()561 mock_alternative.transcript = "Integration test successful"562 mock_alternative.confidence = 0.92563 mock_alternative.words = []564 565 mock_result = Mock()566 mock_result.alternatives = [mock_alternative]567 568 mock_response = Mock()569 mock_response.results = [mock_result]570 571 mock_google_speech['client'].recognize.return_value = mock_response572 573 # Test transcription574 audio_data = b'\x00\x01' * 4000 # 8KB sample575 result = await provider.transcribe_audio(audio_data, "wav", "en-US")576 577 # Verify result578 assert result.text == "Integration test successful"579 assert result.confidence == 0.92580 assert result.provider == "integration_test"581 assert result.language == "en-US"582 583 # Check that usage was tracked584 quota_status = await provider.get_quota_status()585 assert quota_status["requests_per_minute"].current_usage == 1586 assert quota_status["audio_minutes_per_month"].current_usage > 0587 588 @pytest.mark.asyncio589 async def test_quota_management_workflow(self, mock_google_speech):590 """Test quota management across multiple requests."""591 provider = create_google_speech_provider(592 api_credentials={"service_account_key": "test"}593 )594 595 # Mock successful responses596 mock_alternative = Mock()597 mock_alternative.transcript = "Test"598 mock_alternative.confidence = 0.9599 mock_alternative.words = []600 601 mock_result = Mock()602 mock_result.alternatives = [mock_alternative]603 604 mock_response = Mock()605 mock_response.results = [mock_result]606 607 mock_google_speech['client'].recognize.return_value = mock_response608 609 # Make multiple requests610 audio_data = b'\x00\x01' * 1000611 for i in range(5):612 result = await provider.transcribe_audio(audio_data, "wav", "en-US")613 assert result.text == "Test"614 615 # Check quota tracking616 quota_status = await provider.get_quota_status()617 assert quota_status["requests_per_minute"].current_usage == 5618 619 # Verify cost estimation620 cost = await provider.estimate_cost(300) # 5 minutes621 assert cost == 0.0 # Should be free within tier