208 lines
6.9 KiB
Python
208 lines
6.9 KiB
Python
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())
|