File size: 4,750 Bytes
38c016b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 |
#!/usr/bin/env python3
"""
Test the new cache system to verify it works correctly.
"""
import asyncio
import sys
import tempfile
import shutil
from pathlib import Path
# Add project root to path
project_root = Path(__file__).parent.parent # Go up from test-integration to backend-py
sys.path.insert(0, str(project_root))
from src.services.word_cache import WordCacheManager
async def test_cache_system():
"""Test the cache system functionality."""
print("🧪 Testing Word Cache System\n")
# Create temporary cache directory
temp_dir = tempfile.mkdtemp()
print(f"📁 Using temporary cache directory: {temp_dir}")
try:
# Initialize cache manager
cache_manager = WordCacheManager(cache_dir=temp_dir)
await cache_manager.initialize()
# Test 1: Cache some words
print("\n🧪 Test 1: Caching words")
test_words = [
{"word": "ELEPHANT", "clue": "Large mammal with trunk", "similarity": 0.8, "source": "vector_search"},
{"word": "TIGER", "clue": "Striped big cat", "similarity": 0.7, "source": "vector_search"},
{"word": "LION", "clue": "King of jungle", "similarity": 0.75, "source": "vector_search"},
]
success = await cache_manager.cache_words("Animals", "medium", test_words)
print(f"✅ Cache operation {'succeeded' if success else 'failed'}")
# Test 2: Retrieve cached words
print("\n🧪 Test 2: Retrieving cached words")
cached_words = await cache_manager.get_cached_words("Animals", "medium", 5)
print(f"📦 Retrieved {len(cached_words)} cached words")
if cached_words:
print("📝 Cached words:")
for word in cached_words:
print(f" - {word['word']}: {word['clue']}")
# Test 3: Cache statistics
print("\n🧪 Test 3: Cache statistics")
stats = cache_manager.get_cache_stats()
print(f"📊 Cache stats: {stats}")
# Test 4: Test non-existent topic
print("\n🧪 Test 4: Non-existent topic")
empty_words = await cache_manager.get_cached_words("NonExistent", "medium", 5)
print(f"📭 Non-existent topic returned {len(empty_words)} words (expected 0)")
# Test 5: Test bootstrap warming (if static data exists)
print("\n🧪 Test 5: Bootstrap warming simulation")
static_data = {
"Technology": [
{"word": "COMPUTER", "clue": "Electronic device"},
{"word": "ROBOT", "clue": "Automated machine"},
]
}
await cache_manager.warm_cache_from_static(static_data)
tech_words = await cache_manager.get_cached_words("Technology", "medium", 5)
print(f"🔥 Bootstrap warming: Retrieved {len(tech_words)} tech words")
print("\n✅ All cache system tests completed!")
return True
except Exception as e:
print(f"\n❌ Cache system test failed: {e}")
import traceback
traceback.print_exc()
return False
finally:
# Cleanup temporary directory
shutil.rmtree(temp_dir)
print(f"🧹 Cleaned up temporary directory")
async def test_vector_integration():
"""Test integration with vector search service."""
print("\n🔗 Testing Vector Search Integration\n")
try:
from src.services.vector_search import VectorSearchService
# Create vector service (won't initialize model, just test cache integration)
vector_service = VectorSearchService()
# Test cache fallback without initialization
print("🧪 Testing cache fallback when vector search not initialized")
fallback_words = await vector_service._get_cached_fallback("Animals", "medium", 5)
print(f"📦 Fallback returned {len(fallback_words)} words")
print("✅ Vector integration test completed!")
return True
except Exception as e:
print(f"❌ Vector integration test failed: {e}")
import traceback
traceback.print_exc()
return False
async def main():
"""Run all tests."""
print("🚀 Testing Cache System Replacement\n")
cache_test = await test_cache_system()
integration_test = await test_vector_integration()
if cache_test and integration_test:
print("\n🎉 All tests passed! Cache system is working correctly.")
print("📦 Static word dependencies have been successfully replaced with caching.")
else:
print("\n❌ Some tests failed. Check the output above.")
if __name__ == "__main__":
asyncio.run(main()) |