nifty-coder/stemsplit-backend
0
1"""2Unit tests for the CacheManager implementation.3 4Tests cover audio fingerprinting, LRU cache behavior, TTL expiration,5cache statistics tracking, and cache hit optimization.6"""7 8import asyncio9import pytest10import time11from datetime import datetime12from unittest.mock import AsyncMock, MagicMock, patch13 14from voice_control.cache_manager import CacheManager, CacheEntry, create_cache_manager15from voice_control.models import TranscriptionResult, CacheStats16from voice_control.exceptions import CacheError17 18 19class TestCacheEntry:20 """Test the CacheEntry data class."""21 22 def test_cache_entry_creation(self):23 """Test creating a cache entry."""24 result = TranscriptionResult(25 text="hello world",26 confidence=0.95,27 provider="test_provider",28 processing_time=1.0,29 audio_duration=2.0,30 language="en-US"31 )32 33 entry = CacheEntry(34 result=result,35 created_at=time.time(),36 last_accessed=time.time(),37 ttl=360038 )39 40 assert entry.result == result41 assert entry.ttl == 360042 assert entry.access_count == 043 assert entry.size_bytes > 044 45 def test_cache_entry_expiration(self):46 """Test cache entry expiration logic."""47 result = TranscriptionResult(48 text="test",49 confidence=0.9,50 provider="test",51 processing_time=1.0,52 audio_duration=1.0,53 language="en-US"54 )55 56 # Create expired entry57 entry = CacheEntry(58 result=result,59 created_at=time.time() - 7200, # 2 hours ago60 last_accessed=time.time() - 3600, # 1 hour ago61 ttl=3600 # 1 hour TTL62 )63 64 assert entry.is_expired65 66 def test_cache_entry_touch(self):67 """Test updating entry access information."""68 result = TranscriptionResult(69 text="test",70 confidence=0.9,71 provider="test",72 processing_time=1.0,73 audio_duration=1.0,74 language="en-US"75 )76 77 entry = CacheEntry(78 result=result,79 created_at=time.time(),80 last_accessed=time.time(),81 ttl=360082 )83 84 initial_access_count = entry.access_count85 initial_last_accessed = entry.last_accessed86 87 time.sleep(0.01) # Small delay88 entry.touch()89 90 assert entry.access_count == initial_access_count + 191 assert entry.last_accessed > initial_last_accessed92 93 94class TestCacheManager:95 """Test the CacheManager class."""96 97 @pytest.fixture98 def cache_manager(self):99 """Create a cache manager for testing."""100 return CacheManager(101 max_size_mb=1, # Small size for testing102 default_ttl=3600,103 cleanup_interval=60104 )105 106 @pytest.fixture107 def sample_result(self):108 """Create a sample transcription result."""109 return TranscriptionResult(110 text="hello world test transcription",111 confidence=0.95,112 provider="test_provider",113 processing_time=1.5,114 audio_duration=3.0,115 language="en-US",116 alternatives=["hello world", "hello word"]117 )118 119 @pytest.fixture120 def sample_audio(self):121 """Create sample audio data."""122 return b"fake_audio_data_for_testing" * 100123 124 @pytest.mark.asyncio125 async def test_audio_fingerprint_generation(self, cache_manager, sample_audio):126 """Test audio fingerprint generation."""127 fingerprint1 = await cache_manager.generate_audio_fingerprint(sample_audio)128 fingerprint2 = await cache_manager.generate_audio_fingerprint(sample_audio)129 130 # Same audio should produce same fingerprint131 assert fingerprint1 == fingerprint2132 assert len(fingerprint1) == 64 # SHA-256 hex length133 134 # Different audio should produce different fingerprint135 different_audio = b"different_audio_data" * 100136 fingerprint3 = await cache_manager.generate_audio_fingerprint(different_audio)137 assert fingerprint1 != fingerprint3138 139 @pytest.mark.asyncio140 async def test_cache_transcription_and_retrieval(self, cache_manager, sample_result, sample_audio):141 """Test caching and retrieving transcription results."""142 fingerprint = await cache_manager.generate_audio_fingerprint(sample_audio)143 144 # Initially should be cache miss145 cached_result = await cache_manager.get_cached_transcription(fingerprint)146 assert cached_result is None147 148 # Cache the result149 success = await cache_manager.cache_transcription(fingerprint, sample_result)150 assert success151 152 # Should now be cache hit153 cached_result = await cache_manager.get_cached_transcription(fingerprint)154 assert cached_result is not None155 assert cached_result.text == sample_result.text156 assert cached_result.confidence == sample_result.confidence157 assert cached_result.provider == sample_result.provider158 159 @pytest.mark.asyncio160 async def test_cache_ttl_expiration(self, cache_manager, sample_result, sample_audio):161 """Test TTL-based cache expiration."""162 fingerprint = await cache_manager.generate_audio_fingerprint(sample_audio)163 164 # Cache with very short TTL165 success = await cache_manager.cache_transcription(fingerprint, sample_result, ttl=1)166 assert success167 168 # Should be available immediately169 cached_result = await cache_manager.get_cached_transcription(fingerprint)170 assert cached_result is not None171 172 # Wait for expiration173 await asyncio.sleep(1.1)174 175 # Should now be expired176 cached_result = await cache_manager.get_cached_transcription(fingerprint)177 assert cached_result is None178 179 @pytest.mark.asyncio180 async def test_lru_eviction(self):181 """Test LRU (Least Recently Used) eviction policy."""182 # Create cache manager with very small size to force eviction183 cache_manager = CacheManager(max_size_mb=0.01) # 10KB only184 185 # Create multiple large results that will exceed cache size186 results = []187 fingerprints = []188 189 for i in range(5):190 audio_data = f"audio_data_{i}".encode() * 2000 # Make it large191 result = TranscriptionResult(192 text=f"transcription_{i}" * 200, # Make text large193 confidence=0.9,194 provider="test",195 processing_time=1.0,196 audio_duration=1.0,197 language="en-US"198 )199 200 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)201 fingerprints.append(fingerprint)202 results.append(result)203 204 await cache_manager.cache_transcription(fingerprint, result)205 206 # Access some entries to make them more recently used207 await cache_manager.get_cached_transcription(fingerprints[3])208 await cache_manager.get_cached_transcription(fingerprints[4])209 210 # Add one more large entry to trigger eviction211 large_audio = b"large_audio_data" * 5000212 large_result = TranscriptionResult(213 text="large transcription" * 500,214 confidence=0.9,215 provider="test",216 processing_time=1.0,217 audio_duration=1.0,218 language="en-US"219 )220 large_fingerprint = await cache_manager.generate_audio_fingerprint(large_audio)221 await cache_manager.cache_transcription(large_fingerprint, large_result)222 223 # Some early entries should have been evicted224 stats = await cache_manager.get_cache_stats()225 assert stats.evictions > 0226 227 # Recently accessed entries should still be available228 cached_3 = await cache_manager.get_cached_transcription(fingerprints[3])229 cached_4 = await cache_manager.get_cached_transcription(fingerprints[4])230 # At least one should survive, or the large entry should be there231 large_cached = await cache_manager.get_cached_transcription(large_fingerprint)232 assert cached_3 is not None or cached_4 is not None or large_cached is not None233 234 @pytest.mark.asyncio235 async def test_cache_statistics(self, cache_manager, sample_result, sample_audio):236 """Test cache statistics tracking."""237 fingerprint = await cache_manager.generate_audio_fingerprint(sample_audio)238 239 # Initial stats240 stats = await cache_manager.get_cache_stats()241 initial_requests = stats.total_requests242 243 # Cache miss244 await cache_manager.get_cached_transcription(fingerprint)245 246 # Cache the result247 await cache_manager.cache_transcription(fingerprint, sample_result)248 249 # Cache hit250 await cache_manager.get_cached_transcription(fingerprint)251 252 # Check updated stats253 stats = await cache_manager.get_cache_stats()254 assert stats.total_requests == initial_requests + 2255 assert stats.cache_hits >= 1256 assert stats.cache_misses >= 1257 assert stats.entry_count >= 1258 assert 0.0 <= stats.hit_rate <= 1.0259 260 @pytest.mark.asyncio261 async def test_clear_cache(self, cache_manager, sample_result, sample_audio):262 """Test clearing all cache entries."""263 fingerprint = await cache_manager.generate_audio_fingerprint(sample_audio)264 265 # Cache a result266 await cache_manager.cache_transcription(fingerprint, sample_result)267 268 # Verify it's cached269 cached_result = await cache_manager.get_cached_transcription(fingerprint)270 assert cached_result is not None271 272 # Clear cache273 await cache_manager.clear_cache()274 275 # Verify it's gone276 cached_result = await cache_manager.get_cached_transcription(fingerprint)277 assert cached_result is None278 279 # Check stats280 stats = await cache_manager.get_cache_stats()281 assert stats.entry_count == 0282 assert stats.total_size_bytes == 0283 284 @pytest.mark.asyncio285 async def test_evict_expired(self, cache_manager):286 """Test manual expiration of expired entries."""287 # Create results with different TTLs288 results = []289 fingerprints = []290 291 for i in range(3):292 audio_data = f"audio_{i}".encode()293 result = TranscriptionResult(294 text=f"text_{i}",295 confidence=0.9,296 provider="test",297 processing_time=1.0,298 audio_duration=1.0,299 language="en-US"300 )301 302 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)303 fingerprints.append(fingerprint)304 results.append(result)305 306 # Cache with different TTLs307 ttl = 1 if i < 2 else 3600 # First two expire quickly308 await cache_manager.cache_transcription(fingerprint, result, ttl=ttl)309 310 # Wait for some to expire311 await asyncio.sleep(1.1)312 313 # Manually evict expired entries314 evicted_count = await cache_manager.evict_expired()315 assert evicted_count >= 2316 317 # Non-expired entry should still be available318 cached_result = await cache_manager.get_cached_transcription(fingerprints[2])319 assert cached_result is not None320 321 @pytest.mark.asyncio322 async def test_context_manager(self):323 """Test using cache manager as async context manager."""324 async with CacheManager() as cache_manager:325 assert cache_manager._running326 327 # Test basic functionality328 audio_data = b"test_audio"329 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)330 assert len(fingerprint) == 64331 332 # Should be stopped after context exit333 assert not cache_manager._running334 335 @pytest.mark.asyncio336 async def test_concurrent_access(self, cache_manager):337 """Test concurrent cache access."""338 async def cache_and_retrieve(index):339 audio_data = f"audio_{index}".encode()340 result = TranscriptionResult(341 text=f"text_{index}",342 confidence=0.9,343 provider="test",344 processing_time=1.0,345 audio_duration=1.0,346 language="en-US"347 )348 349 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)350 await cache_manager.cache_transcription(fingerprint, result)351 352 cached_result = await cache_manager.get_cached_transcription(fingerprint)353 assert cached_result is not None354 assert cached_result.text == f"text_{index}"355 356 return fingerprint357 358 # Run multiple concurrent operations359 tasks = [cache_and_retrieve(i) for i in range(10)]360 fingerprints = await asyncio.gather(*tasks)361 362 # All should have succeeded363 assert len(fingerprints) == 10364 assert len(set(fingerprints)) == 10 # All unique365 366 def test_get_cache_info(self, cache_manager):367 """Test getting cache information."""368 info = cache_manager.get_cache_info()369 370 assert "max_size_bytes" in info371 assert "current_size_bytes" in info372 assert "entry_count" in info373 assert "default_ttl" in info374 assert "cleanup_interval" in info375 assert "running" in info376 assert "stats" in info377 378 assert info["max_size_bytes"] == 1 * 1024 * 1024 # 1MB379 assert info["default_ttl"] == 3600380 assert info["cleanup_interval"] == 60381 382 383class TestCacheManagerFactory:384 """Test the cache manager factory function."""385 386 def test_create_in_memory_cache_manager(self):387 """Test creating in-memory cache manager."""388 config = {389 "max_size_mb": 50,390 "default_ttl": 1800,391 "cleanup_interval": 300392 }393 394 cache_manager = create_cache_manager(config)395 assert isinstance(cache_manager, CacheManager)396 assert cache_manager.max_size_bytes == 50 * 1024 * 1024397 assert cache_manager.default_ttl == 1800398 assert cache_manager.cleanup_interval == 300399 400 @patch('voice_control.cache_manager.RedisCacheManager')401 def test_create_redis_cache_manager(self, mock_redis_cache):402 """Test creating Redis cache manager."""403 config = {404 "redis_url": "redis://localhost:6379",405 "default_ttl": 1800,406 "max_connections": 20407 }408 409 cache_manager = create_cache_manager(config)410 411 # Should have created RedisCacheManager412 mock_redis_cache.assert_called_once_with(413 redis_url="redis://localhost:6379",414 default_ttl=1800,415 max_connections=20416 )417 418 419class TestErrorHandling:420 """Test error handling in cache manager."""421 422 @pytest.mark.asyncio423 async def test_fingerprint_generation_error(self):424 """Test handling errors in fingerprint generation."""425 cache_manager = CacheManager()426 427 # Mock hashlib to raise an exception428 with patch('voice_control.cache_manager.hashlib.sha256') as mock_hasher:429 mock_hasher.side_effect = Exception("Hash error")430 431 with pytest.raises(CacheError) as exc_info:432 await cache_manager.generate_audio_fingerprint(b"test_audio")433 434 assert "fingerprint_generation" in str(exc_info.value)435 436 @pytest.mark.asyncio437 async def test_cache_operation_resilience(self):438 """Test that cache operations are resilient to errors."""439 cache_manager = CacheManager()440 sample_result = TranscriptionResult(441 text="test",442 confidence=0.9,443 provider="test",444 processing_time=1.0,445 audio_duration=1.0,446 language="en-US"447 )448 449 # Test with invalid fingerprint450 result = await cache_manager.get_cached_transcription("")451 assert result is None452 453 # Test caching with None result (should handle gracefully)454 success = await cache_manager.cache_transcription("test_key", sample_result)455 assert success # Should not crash456 457 458@pytest.mark.integration459class TestCacheManagerIntegration:460 """Integration tests for cache manager with other components."""461 462 @pytest.mark.asyncio463 async def test_realistic_usage_pattern(self):464 """Test realistic usage pattern with multiple operations."""465 async with CacheManager(max_size_mb=5, default_ttl=60) as cache_manager:466 # Simulate multiple audio transcriptions467 transcriptions = []468 469 for i in range(20):470 # Generate unique audio data471 audio_data = f"audio_sample_{i}".encode() * (100 + i * 10)472 473 # Generate fingerprint474 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)475 476 # Check cache first (should be miss initially)477 cached_result = await cache_manager.get_cached_transcription(fingerprint)478 479 if cached_result is None:480 # Simulate transcription481 result = TranscriptionResult(482 text=f"Transcribed text for sample {i}",483 confidence=0.85 + (i % 10) * 0.01,484 provider="test_provider",485 processing_time=1.0 + i * 0.1,486 audio_duration=2.0 + i * 0.2,487 language="en-US"488 )489 490 # Cache the result491 await cache_manager.cache_transcription(fingerprint, result)492 transcriptions.append((fingerprint, result))493 else:494 transcriptions.append((fingerprint, cached_result))495 496 # Verify some results are cached497 stats = await cache_manager.get_cache_stats()498 assert stats.total_requests > 0499 assert stats.entry_count > 0500 501 # Test accessing some cached results502 for fingerprint, expected_result in transcriptions[:5]:503 cached_result = await cache_manager.get_cached_transcription(fingerprint)504 if cached_result: # May have been evicted due to size limits505 assert cached_result.text == expected_result.text506 507 # Final stats check508 final_stats = await cache_manager.get_cache_stats()509 assert final_stats.total_requests >= stats.total_requests510 assert 0.0 <= final_stats.hit_rate <= 1.0511 512 513@pytest.mark.integration514class TestCacheHitOptimization:515 """Tests for cache hit optimization functionality (Task 5.2)."""516 517 @pytest.fixture518 def cache_manager(self):519 """Create a cache manager for testing."""520 return CacheManager(521 max_size_mb=10,522 default_ttl=3600,523 cleanup_interval=60,524 enable_stats=True525 )526 527 @pytest.fixture528 def sample_audio_data(self):529 """Create sample audio data for testing."""530 return b"sample_audio_data_for_cache_testing" * 50531 532 @pytest.fixture533 def sample_transcription_result(self):534 """Create a sample transcription result."""535 return TranscriptionResult(536 text="This is a sample transcription for cache testing",537 confidence=0.95,538 provider="test_provider",539 processing_time=1.5,540 audio_duration=3.0,541 language="en-US",542 alternatives=["alternative transcription"]543 )544 545 @pytest.mark.asyncio546 async def test_cache_hit_avoids_api_calls(self, cache_manager, sample_audio_data, sample_transcription_result):547 """548 Test that cache hits avoid API calls (Requirement 4.3).549 550 **Validates: Requirements 4.3**551 """552 async with cache_manager:553 # Generate fingerprint554 fingerprint = await cache_manager.generate_audio_fingerprint(sample_audio_data)555 556 # Initial cache miss557 cached_result = await cache_manager.get_cached_transcription(fingerprint)558 assert cached_result is None559 560 # Cache the transcription result561 cache_success = await cache_manager.cache_transcription(fingerprint, sample_transcription_result)562 assert cache_success563 564 # Subsequent access should be cache hit (no API call needed)565 cached_result = await cache_manager.get_cached_transcription(fingerprint)566 assert cached_result is not None567 assert cached_result.text == sample_transcription_result.text568 assert cached_result.provider == sample_transcription_result.provider569 assert cached_result.confidence == sample_transcription_result.confidence570 571 # Verify cache statistics show the hit572 stats = await cache_manager.get_cache_stats()573 assert stats.cache_hits >= 1574 assert stats.total_requests >= 2 # At least one miss + one hit575 576 @pytest.mark.asyncio577 async def test_cache_statistics_tracking(self, cache_manager):578 """579 Test comprehensive cache statistics tracking.580 581 **Validates: Requirements 4.3**582 """583 async with cache_manager:584 initial_stats = await cache_manager.get_cache_stats()585 586 # Perform multiple cache operations587 audio_samples = []588 results = []589 590 for i in range(5):591 audio_data = f"audio_sample_{i}".encode() * 100592 result = TranscriptionResult(593 text=f"transcription_{i}",594 confidence=0.9,595 provider="test_provider",596 processing_time=1.0,597 audio_duration=2.0,598 language="en-US"599 )600 601 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)602 audio_samples.append((audio_data, fingerprint))603 results.append(result)604 605 # First access - cache miss606 cached_result = await cache_manager.get_cached_transcription(fingerprint)607 assert cached_result is None608 609 # Cache the result610 await cache_manager.cache_transcription(fingerprint, result)611 612 # Second access - cache hit613 cached_result = await cache_manager.get_cached_transcription(fingerprint)614 assert cached_result is not None615 616 # Check final statistics617 final_stats = await cache_manager.get_cache_stats()618 619 # Should have 5 misses (first access) + 5 hits (second access) = 10 total requests620 assert final_stats.total_requests == initial_stats.total_requests + 10621 assert final_stats.cache_hits == initial_stats.cache_hits + 5622 assert final_stats.cache_misses == initial_stats.cache_misses + 5623 assert final_stats.entry_count == initial_stats.entry_count + 5624 625 # Hit rate should be exactly 50% for this test626 expected_hit_rate = final_stats.cache_hits / final_stats.total_requests627 assert abs(final_stats.hit_rate - expected_hit_rate) < 0.01628 629 @pytest.mark.asyncio630 async def test_detailed_cache_statistics(self, cache_manager):631 """632 Test detailed cache statistics and optimization recommendations.633 634 **Validates: Requirements 4.3**635 """636 async with cache_manager:637 # Perform various cache operations638 for i in range(10):639 audio_data = f"detailed_stats_audio_{i}".encode() * 100640 result = TranscriptionResult(641 text=f"detailed_stats_transcription_{i}",642 confidence=0.9,643 provider="test_provider",644 processing_time=1.0,645 audio_duration=2.0,646 language="en-US"647 )648 649 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)650 651 # Cache miss652 await cache_manager.get_cached_transcription(fingerprint)653 654 # Cache the result655 await cache_manager.cache_transcription(fingerprint, result)656 657 # Multiple cache hits to vary access patterns658 for _ in range(i + 1): # Varying access counts659 await cache_manager.get_cached_transcription(fingerprint)660 661 # Get detailed statistics662 detailed_stats = await cache_manager.get_detailed_cache_stats()663 664 # Verify structure665 assert "basic_stats" in detailed_stats666 assert "size_stats" in detailed_stats667 assert "performance_stats" in detailed_stats668 assert "optimization_recommendations" in detailed_stats669 670 # Verify basic stats671 basic_stats = detailed_stats["basic_stats"]672 assert basic_stats["total_requests"] > 0673 assert basic_stats["cache_hits"] > 0674 assert basic_stats["cache_misses"] > 0675 assert 0.0 <= basic_stats["hit_rate"] <= 1.0676 677 # Verify size stats678 size_stats = detailed_stats["size_stats"]679 assert size_stats["entry_count"] == 10680 assert size_stats["total_size_bytes"] > 0681 assert size_stats["avg_entry_size_bytes"] > 0682 assert 0.0 <= size_stats["utilization_rate"] <= 1.0683 684 # Verify performance stats685 perf_stats = detailed_stats["performance_stats"]686 assert "cache_efficiency" in perf_stats687 assert "eviction_rate" in perf_stats688 assert "access_patterns" in perf_stats689 690 # Verify access patterns691 access_patterns = perf_stats["access_patterns"]692 assert access_patterns["most_accessed_entries"] > 0693 assert access_patterns["avg_access_count"] > 0694 695 # Verify recommendations exist696 recommendations = detailed_stats["optimization_recommendations"]697 assert isinstance(recommendations, list)698 assert len(recommendations) > 0699 700 @pytest.mark.asyncio701 async def test_batch_cache_lookup(self, cache_manager):702 """703 Test batch cache lookup optimization.704 705 **Validates: Requirements 4.3**706 """707 async with cache_manager:708 # Prepare test data709 fingerprints = []710 results = []711 712 for i in range(5):713 audio_data = f"batch_test_audio_{i}".encode() * 100714 result = TranscriptionResult(715 text=f"batch_test_transcription_{i}",716 confidence=0.9,717 provider="test_provider",718 processing_time=1.0,719 audio_duration=2.0,720 language="en-US"721 )722 723 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)724 fingerprints.append(fingerprint)725 results.append(result)726 727 # Cache every other result728 if i % 2 == 0:729 await cache_manager.cache_transcription(fingerprint, result)730 731 # Perform batch lookup732 batch_results = await cache_manager.get_cached_transcription_batch(fingerprints)733 734 # Verify results735 assert len(batch_results) == 5736 737 # Check that cached entries are found and non-cached are None738 for i, fingerprint in enumerate(fingerprints):739 if i % 2 == 0: # Should be cached740 assert batch_results[fingerprint] is not None741 assert batch_results[fingerprint].text == f"batch_test_transcription_{i}"742 else: # Should not be cached743 assert batch_results[fingerprint] is None744 745 @pytest.mark.asyncio746 async def test_cache_warming(self, cache_manager):747 """748 Test cache warming functionality.749 750 **Validates: Requirements 4.3**751 """752 async with cache_manager:753 # Prepare fingerprints for warming754 fingerprints = []755 for i in range(3):756 audio_data = f"warming_test_audio_{i}".encode() * 100757 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)758 fingerprints.append(fingerprint)759 760 # Mock transcription callback761 async def mock_transcription_callback(fingerprint):762 return TranscriptionResult(763 text=f"warmed_transcription_{fingerprint[:8]}",764 confidence=0.9,765 provider="warming_provider",766 processing_time=1.0,767 audio_duration=2.0,768 language="en-US"769 )770 771 # Warm cache772 warming_results = await cache_manager.warm_cache(773 fingerprints, 774 mock_transcription_callback775 )776 777 # Verify warming results778 assert len(warming_results) == 3779 for fingerprint in fingerprints:780 assert warming_results[fingerprint] is True781 782 # Verify entries are cached783 for fingerprint in fingerprints:784 cached_result = await cache_manager.get_cached_transcription(fingerprint)785 assert cached_result is not None786 assert cached_result.provider == "warming_provider"787 788 @pytest.mark.asyncio789 async def test_cache_performance_optimization(self, cache_manager):790 """791 Test cache performance optimization operations.792 793 **Validates: Requirements 4.3**794 """795 async with cache_manager:796 # Create entries with different access patterns797 fingerprints = []798 799 for i in range(20):800 audio_data = f"optimization_test_audio_{i}".encode() * 100801 result = TranscriptionResult(802 text=f"optimization_test_transcription_{i}",803 confidence=0.9,804 provider="test_provider",805 processing_time=1.0,806 audio_duration=2.0,807 language="en-US"808 )809 810 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)811 fingerprints.append(fingerprint)812 813 # Cache with short TTL for some entries to create expired entries814 ttl = 1 if i < 5 else 3600815 await cache_manager.cache_transcription(fingerprint, result, ttl=ttl)816 817 # Create frequent access pattern for some entries818 if i >= 15: # Last 5 entries819 for _ in range(10): # Access 10 times each820 await cache_manager.get_cached_transcription(fingerprint)821 822 # Wait for some entries to expire823 await asyncio.sleep(1.1)824 825 # Perform optimization826 optimization_results = await cache_manager.optimize_cache_performance()827 828 # Verify optimization results structure829 assert "initial_stats" in optimization_results830 assert "expired_evicted" in optimization_results831 assert "optimizations_applied" in optimization_results832 assert "final_stats" in optimization_results833 834 # Verify expired entries were evicted835 assert optimization_results["expired_evicted"] >= 5836 837 # Verify optimizations were applied838 optimizations = optimization_results["optimizations_applied"]839 assert isinstance(optimizations, list)840 841 # Check that frequently accessed entries were promoted842 frequent_promotion = any("frequently accessed" in opt for opt in optimizations)843 assert frequent_promotion844 845 @pytest.mark.asyncio846 async def test_identical_audio_produces_cache_hits(self, cache_manager):847 """848 Test that identical audio data produces cache hits.849 850 **Validates: Requirements 4.1, 4.3**851 """852 async with cache_manager:853 # Create identical audio data854 audio_data = b"identical_audio_content" * 200855 856 result1 = TranscriptionResult(857 text="first transcription",858 confidence=0.95,859 provider="provider1",860 processing_time=1.0,861 audio_duration=2.0,862 language="en-US"863 )864 865 # First transcription and caching866 fingerprint1 = await cache_manager.generate_audio_fingerprint(audio_data)867 await cache_manager.cache_transcription(fingerprint1, result1)868 869 # Second identical audio should produce same fingerprint and cache hit870 fingerprint2 = await cache_manager.generate_audio_fingerprint(audio_data)871 assert fingerprint1 == fingerprint2872 873 cached_result = await cache_manager.get_cached_transcription(fingerprint2)874 assert cached_result is not None875 assert cached_result.text == result1.text876 assert cached_result.provider == result1.provider877 878 @pytest.mark.asyncio879 async def test_different_audio_produces_cache_misses(self, cache_manager):880 """881 Test that different audio data produces cache misses.882 883 **Validates: Requirements 4.5**884 """885 async with cache_manager:886 # Create different audio data887 audio_data1 = b"first_audio_content" * 200888 audio_data2 = b"second_audio_content" * 200889 890 result1 = TranscriptionResult(891 text="first transcription",892 confidence=0.95,893 provider="provider1",894 processing_time=1.0,895 audio_duration=2.0,896 language="en-US"897 )898 899 # Cache first audio900 fingerprint1 = await cache_manager.generate_audio_fingerprint(audio_data1)901 await cache_manager.cache_transcription(fingerprint1, result1)902 903 # Different audio should produce different fingerprint and cache miss904 fingerprint2 = await cache_manager.generate_audio_fingerprint(audio_data2)905 assert fingerprint1 != fingerprint2906 907 cached_result = await cache_manager.get_cached_transcription(fingerprint2)908 assert cached_result is None909 910 @pytest.mark.asyncio911 async def test_cache_performance_under_load(self, cache_manager):912 """913 Test cache performance under concurrent load.914 915 **Validates: Requirements 4.3**916 """917 async with cache_manager:918 import time919 920 async def cache_operation(index):921 """Perform cache operations for a specific index."""922 audio_data = f"load_test_audio_{index}".encode() * 100923 result = TranscriptionResult(924 text=f"load_test_transcription_{index}",925 confidence=0.9,926 provider="load_test_provider",927 processing_time=0.5,928 audio_duration=1.0,929 language="en-US"930 )931 932 # Generate fingerprint933 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)934 935 # Cache miss936 start_time = time.time()937 cached_result = await cache_manager.get_cached_transcription(fingerprint)938 miss_time = time.time() - start_time939 assert cached_result is None940 941 # Cache the result942 await cache_manager.cache_transcription(fingerprint, result)943 944 # Cache hit945 start_time = time.time()946 cached_result = await cache_manager.get_cached_transcription(fingerprint)947 hit_time = time.time() - start_time948 assert cached_result is not None949 950 return miss_time, hit_time951 952 # Run concurrent cache operations953 tasks = [cache_operation(i) for i in range(20)]954 results = await asyncio.gather(*tasks)955 956 # Verify all operations completed successfully957 assert len(results) == 20958 959 # Cache hits should generally be faster than misses960 miss_times = [r[0] for r in results]961 hit_times = [r[1] for r in results]962 963 avg_miss_time = sum(miss_times) / len(miss_times)964 avg_hit_time = sum(hit_times) / len(hit_times)965 966 # Cache hits should be faster (though this may not always be true in tests)967 # The important thing is that both operations complete successfully968 assert avg_miss_time >= 0969 assert avg_hit_time >= 0970 971 # Verify final statistics972 stats = await cache_manager.get_cache_stats()973 assert stats.total_requests >= 40 # At least 20 misses + 20 hits974 assert stats.cache_hits >= 20975 assert stats.cache_misses >= 20976 977 @pytest.mark.asyncio978 async def test_cache_size_limits_and_eviction(self, cache_manager):979 """980 Test cache size limits and LRU eviction behavior.981 982 **Validates: Requirements 4.2**983 """984 # Create cache manager with very small size to force eviction985 small_cache = CacheManager(max_size_mb=0.01, enable_stats=True) # 10KB only986 987 async with small_cache:988 cached_items = []989 990 # Fill cache beyond capacity991 for i in range(10):992 audio_data = f"large_audio_data_{i}".encode() * 2000 # Make it large993 result = TranscriptionResult(994 text=f"large_transcription_{i}" * 100, # Make text large995 confidence=0.9,996 provider="test_provider",997 processing_time=1.0,998 audio_duration=1.0,999 language="en-US"1000 )1001 1002 fingerprint = await small_cache.generate_audio_fingerprint(audio_data)1003 await small_cache.cache_transcription(fingerprint, result)1004 cached_items.append((fingerprint, result))1005 1006 # Access some items to make them recently used1007 for fingerprint, _ in cached_items[-3:]: # Access last 3 items1008 await small_cache.get_cached_transcription(fingerprint)1009 1010 # Check that evictions occurred1011 stats = await small_cache.get_cache_stats()1012 assert stats.evictions > 01013 1014 # Recently accessed items should be more likely to remain1015 remaining_count = 01016 for fingerprint, _ in cached_items[-3:]:1017 cached_result = await small_cache.get_cached_transcription(fingerprint)1018 if cached_result is not None:1019 remaining_count += 11020 1021 # At least some recently accessed items should remain1022 # (exact behavior depends on size calculations)1023 assert remaining_count >= 0 # Basic sanity check1024 1025 @pytest.mark.asyncio1026 async def test_cache_ttl_expiration_behavior(self, cache_manager):1027 """1028 Test TTL-based cache expiration and statistics.1029 1030 **Validates: Requirements 4.4**1031 """1032 async with cache_manager:1033 audio_data = b"ttl_test_audio" * 1001034 result = TranscriptionResult(1035 text="ttl test transcription",1036 confidence=0.95,1037 provider="ttl_test_provider",1038 processing_time=1.0,1039 audio_duration=2.0,1040 language="en-US"1041 )1042 1043 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)1044 1045 # Cache with short TTL1046 await cache_manager.cache_transcription(fingerprint, result, ttl=1)1047 1048 # Should be available immediately1049 cached_result = await cache_manager.get_cached_transcription(fingerprint)1050 assert cached_result is not None1051 1052 # Wait for expiration1053 await asyncio.sleep(1.1)1054 1055 # Should now be expired (cache miss)1056 cached_result = await cache_manager.get_cached_transcription(fingerprint)1057 assert cached_result is None1058 1059 # Check that expiration was tracked in statistics1060 stats = await cache_manager.get_cache_stats()1061 assert stats.cache_misses >= 1 # The expired access should count as miss eviction1062 small_cache = CacheManager(max_size_mb=0.01, enable_stats=True) # 10KB only1063 1064 async with small_cache:1065 cached_items = []1066 1067 # Fill cache beyond capacity1068 for i in range(10):1069 audio_data = f"large_audio_data_{i}".encode() * 2000 # Make it large1070 result = TranscriptionResult(1071 text=f"large_transcription_{i}" * 100, # Make text large1072 confidence=0.9,1073 provider="test_provider",1074 processing_time=1.0,1075 audio_duration=1.0,1076 language="en-US"1077 )1078 1079 fingerprint = await small_cache.generate_audio_fingerprint(audio_data)1080 await small_cache.cache_transcription(fingerprint, result)1081 cached_items.append((fingerprint, result))1082 1083 # Access some items to make them recently used1084 for fingerprint, _ in cached_items[-3:]: # Access last 3 items1085 await small_cache.get_cached_transcription(fingerprint)1086 1087 # Check that evictions occurred1088 stats = await small_cache.get_cache_stats()1089 assert stats.evictions > 01090 1091 # Recently accessed items should be more likely to remain1092 remaining_count = 01093 for fingerprint, _ in cached_items[-3:]:1094 cached_result = await small_cache.get_cached_transcription(fingerprint)1095 if cached_result is not None:1096 remaining_count += 11097 1098 # At least some recently accessed items should remain1099 # (exact behavior depends on size calculations)1100 assert remaining_count >= 0 # Basic sanity check1101 1102 @pytest.mark.asyncio1103 async def test_cache_ttl_expiration_behavior(self, cache_manager):1104 """1105 Test TTL-based cache expiration and statistics.1106 1107 **Validates: Requirements 4.4**1108 """1109 async with cache_manager:1110 audio_data = b"ttl_test_audio" * 1001111 result = TranscriptionResult(1112 text="ttl test transcription",1113 confidence=0.95,1114 provider="ttl_test_provider",1115 processing_time=1.0,1116 audio_duration=2.0,1117 language="en-US"1118 )1119 1120 fingerprint = await cache_manager.generate_audio_fingerprint(audio_data)1121 1122 # Cache with short TTL1123 await cache_manager.cache_transcription(fingerprint, result, ttl=1)1124 1125 # Should be available immediately1126 cached_result = await cache_manager.get_cached_transcription(fingerprint)1127 assert cached_result is not None1128 1129 # Wait for expiration1130 await asyncio.sleep(1.1)1131 1132 # Should now be expired (cache miss)1133 cached_result = await cache_manager.get_cached_transcription(fingerprint)1134 assert cached_result is None1135 1136 # Check that expiration was tracked in statistics1137 stats = await cache_manager.get_cache_stats()1138 assert stats.cache_misses >= 1 # The expired access should count as miss1139 1140 1141if __name__ == "__main__":1142 pytest.main([__file__])