import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import asyncio import tempfile import time from context.sketch_pad import RedisFileSketchPadBackend 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_comprehensive_operations(): """Comprehensive SketchPad feature test.""" print("=== Comprehensive SketchPad Feature Test ===") # 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("comprehensive_test", temp_file) print("1. Testing basic CRUD operations...") # Create items. await sketch_pad.set_item( key="user_profile", value={"name": "Zhang San", "age": 30, "city": "Beijing"}, summary="User profile information", tags={"profile", "user", "personal"}, ) await sketch_pad.set_item( key="project_config", value={"debug": True, "log_level": "INFO", "max_workers": 4}, summary="Project configuration information", tags={"config", "project", "settings"}, ) await sketch_pad.set_item( key="temp_note", value="This is a temporary note for testing", summary="Temporary note", tags={"note", "temp"}, ttl=10, # Expires after 10 seconds. ) # Read items. profile_item = sketch_pad.get_item("user_profile") print(f" User information: {profile_item.value if profile_item else 'Not found'}") config_value = sketch_pad.get_value("project_config") print(f" Project configuration: {config_value}") # Check existence. print(f" user_profile exists: {sketch_pad.exists('user_profile')}") print(f" non_existent exists: {sketch_pad.exists('non_existent')}") print("2. Testing tag search...") # Single-tag search. profile_items = sketch_pad.search_by_tags({"profile"}) print(f" Items with the profile tag: {len(profile_items)}") # Multi-tag search (match any). config_items = sketch_pad.search_by_tags({"config", "settings"}) print(f" Items with the config or settings tag: {len(config_items)}") # Multi-tag search (match all). all_match_items = sketch_pad.search_by_tags( {"config", "project"}, match_all=True ) print(f" Items with both config and project tags: {len(all_match_items)}") print("3. Testing content search...") # Search for items containing specific content. search_results = sketch_pad.search_by_content("configuration", limit=5) print(f" Items containing 'configuration': {len(search_results)}") search_results = sketch_pad.search_by_content("Beijing", limit=5) print(f" Items containing 'Beijing': {len(search_results)}") print("4. Testing access statistics...") # Access the same item multiple times. for i in range(3): item = sketch_pad.get_item("user_profile") print( f" Access {i + 1} to user_profile, access count: {item.access_count if item else 0}" ) print("5. Testing expiration behavior...") # Check whether the temporary item has expired. print(" Waiting 5 seconds to check expiration behavior...") await asyncio.sleep(5) temp_item = sketch_pad.get_item("temp_note") print(f" Temporary item status: {'expired' if temp_item is None else 'not expired'}") print("6. Testing statistics...") stats = sketch_pad.get_statistics() print(f" Total item count: {stats.total_items}") print(f" Total accesses: {stats.total_accesses}") print(f" Items with summaries: {stats.items_with_summary}") print(f" Popular tags: {stats.popular_tags}") print(f" Content type statistics: {stats.content_types}") print(f" Average access count: {stats.avg_access_per_item:.2f}") print("7. Testing listing functionality...") items = sketch_pad.list_items(include_value=False) print(f" Item list (without values): {len(items)} items") for item in items: print(f" - {item.key}: {item.summary} (access count: {item.access_count})") items_with_values = sketch_pad.list_items(include_value=True) print(f" Item list (with values): {len(items_with_values)} items") print("8. Testing persistence and restoration...") # Persist data. sketch_pad.persist() print(" Data persisted to file") # Create a new sketch pad and restore data. new_sketch_pad = _create_sketch_pad("restored_test", temp_file + ".restored") # Copy data. data = sketch_pad.serialize() new_sketch_pad.deserialize(data) # Verify restored data. restored_profile = new_sketch_pad.get_item("user_profile") print( f" Restored user information: {restored_profile.value if restored_profile else 'Not found'}" ) print("9. Testing deletion functionality...") # Delete one item. deleted = sketch_pad.delete("project_config") print(f" Deleted project_config: {deleted}") print(f" project_config exists: {sketch_pad.exists('project_config')}") # Verify that the tag index was also deleted. config_search = sketch_pad.search_by_tags({"config"}) print(f" Items with the config tag after deletion: {len(config_search)}") print("10. Testing clear functionality...") # Clear all data. sketch_pad.clear() print(" Data cleared") # Verify the clear result. remaining_keys = sketch_pad.keys() print(f" Remaining key count: {len(remaining_keys)}") print("=== Comprehensive 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 test_performance(): """Test performance.""" print("\n=== SketchPad Performance Test ===") with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: temp_file = f.name try: sketch_pad = _create_sketch_pad("performance_test", temp_file) print("1. Testing batch write performance...") start_time = time.time() for i in range(100): await sketch_pad.set_item( key=f"item_{i}", value=f"This is the data for item {i}", summary=f"Summary for item {i}", tags={f"tag_{i % 10}", f"category_{i % 5}"}, ) write_time = time.time() - start_time print(f" Time to write 100 items: {write_time:.3f} seconds") print(f" Average write speed: {100 / write_time:.1f} items/second") print("2. Testing batch read performance...") start_time = time.time() for i in range(100): item = sketch_pad.get_item(f"item_{i}") read_time = time.time() - start_time print(f" Time to read 100 items: {read_time:.3f} seconds") print(f" Average read speed: {100 / read_time:.1f} items/second") print("3. Testing search performance...") start_time = time.time() search_results = sketch_pad.search_by_tags({"tag_1"}) search_time = time.time() - start_time print(f" Tag search time: {search_time:.3f} seconds") print(f" Search result count: {len(search_results)}") print("4. Testing statistics performance...") start_time = time.time() stats = sketch_pad.get_statistics() stats_time = time.time() - start_time print(f" Statistics calculation time: {stats_time:.3f} seconds") print(f" Statistics result: {stats.total_items} items") print("=== Performance Test Complete ===") finally: if os.path.exists(temp_file): os.unlink(temp_file) async def main(): """Main test function.""" print("Starting comprehensive SketchPad tests...") try: await test_comprehensive_operations() await test_performance() print("\nAll tests complete! SketchPad is functioning normally.") except Exception as e: print(f"Error during tests: {e}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(main())