import asyncio import json import tempfile import os from typing import Dict, Any from context.sketch_pad import RedisFileSketchPadBackend from context.schemas import SketchPadItem from config.config import get_config def _create_sketch_pad(sketch_pad_id: str, file_path: str) -> RedisFileSketchPadBackend: config = get_config() return RedisFileSketchPadBackend( sketch_pad_id=sketch_pad_id, redis_host=config.REDIS_HOST, redis_port=int(config.REDIS_PORT), redis_db=int(config.REDIS_DB), file_path=file_path, ) async def test_sketch_pad_basic_operations(): """Test basic SketchPad operations.""" print("=== Test Basic SketchPad Operations ===") # Create a temporary file. with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: temp_file = f.name try: # Initialize SketchPad. sketch_pad = _create_sketch_pad("test_pad", temp_file) # Test setting items. print("1. Testing item creation...") await sketch_pad.set_item( key="user_preference", value={"theme": "dark", "language": "zh-CN"}, summary="User preference settings", tags={"preference", "settings"}, ) await sketch_pad.set_item( key="recent_files", value=["file1.txt", "file2.py", "file3.json"], summary="Recently accessed files", tags={"files", "recent"}, ) await sketch_pad.set_item( key="temp_data", value="This is temporary data", summary="Temporarily stored data", tags={"temp", "data"}, ttl=60, # Expires after 60 seconds. ) # Test retrieving items. print("2. Testing item retrieval...") item = sketch_pad.get_item("user_preference") print(f" user_preference: {item.value if item else 'Not found'}") value = sketch_pad.get_value("recent_files") print(f" recent_files: {value}") # Test existence checks. print("3. Testing existence checks...") print(f" user_preference exists: {sketch_pad.exists('user_preference')}") print(f" non_existent exists: {sketch_pad.exists('non_existent')}") # Test retrieving all keys. print("4. Testing key retrieval...") keys = sketch_pad.keys() print(f" All keys: {keys}") # Test tag search. print("5. Testing tag search...") preference_items = sketch_pad.search_by_tags({"preference"}) print(f" Items with the preference tag: {len(preference_items)}") recent_items = sketch_pad.search_by_tags({"recent", "files"}, match_all=True) print(f" Items with both recent and files tags: {len(recent_items)}") # Test content search. print("6. Testing content search...") search_results = sketch_pad.search_by_content("file", limit=3) print(f" Content containing 'file': {len(search_results)}") # Test statistics. print("7. Testing statistics...") stats = sketch_pad.get_statistics() print(f" Total item count: {stats.total_items}") print(f" Total accesses: {stats.total_accesses}") print(f" Popular tags: {stats.popular_tags}") # Test listing items. print("8. Testing item listing...") items = sketch_pad.list_items(include_value=False) print(f" Item list: {len(items)} items") for item in items: print(f" - {item.key}: {item.summary}") # Test persistence. print("9. Testing persistence...") sketch_pad.persist() print(" Persistence complete") # Test deletion. print("10. Testing deletion...") deleted = sketch_pad.delete("temp_data") print(f" Deleted temp_data: {deleted}") print(f" temp_data exists: {sketch_pad.exists('temp_data')}") print("=== Test Complete ===") finally: # Clean up. if os.path.exists(temp_file): os.unlink(temp_file) async def test_sketch_pad_advanced_features(): """Test advanced SketchPad features.""" print("\n=== Test Advanced SketchPad Features ===") with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: temp_file = f.name try: sketch_pad = _create_sketch_pad("advanced_test", temp_file) # Test complex data structures. print("1. Testing complex data structures...") complex_data = { "nested": {"list": [1, 2, 3], "dict": {"a": 1, "b": 2}, "string": "test"}, "array": [{"id": 1, "name": "item1"}, {"id": 2, "name": "item2"}], } await sketch_pad.set_item( key="complex_data", value=complex_data, summary="Complex nested data structure", tags={"complex", "nested", "data"}, ) # Test access counting. print("2. Testing access counting...") for i in range(5): item = sketch_pad.get_item("complex_data") print(f" Access {i + 1}, access count: {item.access_count if item else 0}") # Test expiration time. print("3. Testing expiration time...") await sketch_pad.set_item( key="expiring_item", value="This item will expire", summary="Test expiration behavior", tags={"expire", "test"}, ttl=2, # Expires after 2 seconds. ) print(" Waiting 3 seconds for the item to expire...") await asyncio.sleep(3) expired_item = sketch_pad.get_item("expiring_item") print(f" Expired item: {'expired' if expired_item is None else 'not expired'}") # Test serialization and deserialization. print("4. Testing serialization and deserialization...") serialized = sketch_pad.serialize() print(f" Serialized data size: {len(json.dumps(serialized, default=str))} characters") # Create a new sketch pad and deserialize the data. new_sketch_pad = _create_sketch_pad("restored_test", temp_file + ".restored") new_sketch_pad.deserialize(serialized) restored_item = new_sketch_pad.get_item("complex_data") print( f" Data after deserialization: {restored_item.value if restored_item else 'Not found'}" ) print("=== Advanced Feature Test Complete ===") finally: if os.path.exists(temp_file): os.unlink(temp_file) if os.path.exists(temp_file + ".restored"): os.unlink(temp_file + ".restored") async def main(): """Main test function.""" print("Starting SketchPad tests...") try: await test_sketch_pad_basic_operations() await test_sketch_pad_advanced_features() print("\nAll tests complete!") except Exception as e: print(f"Error during tests: {e}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(main())