261 lines
8.8 KiB
Python
261 lines
8.8 KiB
Python
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())
|