171 lines
5.4 KiB
Python
171 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""
|
|
Test script to validate the asset router optimization syntax and structure.
|
|
This script tests that the optimized code is syntactically correct and follows
|
|
the expected patterns.
|
|
"""
|
|
|
|
import ast
|
|
import inspect
|
|
from routers.assets import list_assets, get_asset
|
|
|
|
def test_list_assets_optimization():
|
|
"""Test that list_assets function has the expected optimization patterns."""
|
|
print("Testing list_assets optimization patterns...")
|
|
|
|
# Get the source code
|
|
source = inspect.getsource(list_assets)
|
|
|
|
# Check for optimization comments
|
|
if "OPTIMIZATION:" in source:
|
|
print("✅ Optimization comments found")
|
|
else:
|
|
print("❌ Missing optimization comments")
|
|
return False
|
|
|
|
# Check for selectinload usage
|
|
if "selectinload" in source:
|
|
print("✅ selectinload usage found")
|
|
else:
|
|
print("❌ Missing selectinload usage")
|
|
return False
|
|
|
|
# Check for joinedload usage
|
|
if "joinedload" in source:
|
|
print("✅ joinedload usage found")
|
|
else:
|
|
print("❌ Missing joinedload usage")
|
|
return False
|
|
|
|
# Check for task_updated_at tracking
|
|
if "task_updated_at" in source:
|
|
print("✅ task_updated_at tracking found")
|
|
else:
|
|
print("❌ Missing task_updated_at tracking")
|
|
return False
|
|
|
|
# Check for pre-fetching pattern
|
|
if "project_data" in source and "pre-fetch" in source.lower():
|
|
print("✅ Pre-fetching pattern found")
|
|
else:
|
|
print("❌ Missing pre-fetching pattern")
|
|
return False
|
|
|
|
return True
|
|
|
|
def test_get_asset_optimization():
|
|
"""Test that get_asset function has the expected optimization patterns."""
|
|
print("\nTesting get_asset optimization patterns...")
|
|
|
|
# Get the source code
|
|
source = inspect.getsource(get_asset)
|
|
|
|
# Check for optimization comments
|
|
if "OPTIMIZATION:" in source:
|
|
print("✅ Optimization comments found")
|
|
else:
|
|
print("❌ Missing optimization comments")
|
|
return False
|
|
|
|
# Check for selectinload usage
|
|
if "selectinload" in source:
|
|
print("✅ selectinload usage found")
|
|
else:
|
|
print("❌ Missing selectinload usage")
|
|
return False
|
|
|
|
# Check for joinedload usage
|
|
if "joinedload" in source:
|
|
print("✅ joinedload usage found")
|
|
else:
|
|
print("❌ Missing joinedload usage")
|
|
return False
|
|
|
|
# Check for relationship-based task counting
|
|
if "active_tasks" in source and "task.deleted_at is None" in source:
|
|
print("✅ Relationship-based task counting found")
|
|
else:
|
|
print("❌ Missing relationship-based task counting")
|
|
return False
|
|
|
|
return True
|
|
|
|
def test_function_signatures():
|
|
"""Test that function signatures are correct."""
|
|
print("\nTesting function signatures...")
|
|
|
|
# Check list_assets signature
|
|
sig = inspect.signature(list_assets)
|
|
expected_params = ['project_id', 'category', 'task_status_filter', 'sort_by', 'sort_direction', 'skip', 'limit', 'db', 'current_user']
|
|
actual_params = list(sig.parameters.keys())
|
|
|
|
if all(param in actual_params for param in expected_params):
|
|
print("✅ list_assets signature is correct")
|
|
else:
|
|
print(f"❌ list_assets signature mismatch. Expected: {expected_params}, Got: {actual_params}")
|
|
return False
|
|
|
|
# Check get_asset signature
|
|
sig = inspect.signature(get_asset)
|
|
expected_params = ['asset_id', 'db', 'current_user']
|
|
actual_params = list(sig.parameters.keys())
|
|
|
|
if all(param in actual_params for param in expected_params):
|
|
print("✅ get_asset signature is correct")
|
|
else:
|
|
print(f"❌ get_asset signature mismatch. Expected: {expected_params}, Got: {actual_params}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def main():
|
|
"""Run all syntax and structure tests."""
|
|
print("🚀 Starting Asset Router Optimization Syntax Tests")
|
|
print("=" * 55)
|
|
|
|
try:
|
|
# Test all functions
|
|
tests = [
|
|
test_list_assets_optimization,
|
|
test_get_asset_optimization,
|
|
test_function_signatures
|
|
]
|
|
|
|
results = []
|
|
for test in tests:
|
|
try:
|
|
result = test()
|
|
results.append(result)
|
|
except Exception as e:
|
|
print(f"❌ Test failed with exception: {e}")
|
|
results.append(False)
|
|
|
|
# Summary
|
|
print("\n" + "=" * 55)
|
|
print("📊 Test Results Summary")
|
|
print("=" * 55)
|
|
|
|
passed = sum(results)
|
|
total = len(results)
|
|
|
|
print(f"✅ Passed: {passed}/{total}")
|
|
if passed == total:
|
|
print("🎉 All asset router optimization syntax tests passed!")
|
|
print("✅ The optimization follows the same pattern as the shot router")
|
|
print("✅ Single query operations are implemented")
|
|
print("✅ Eager loading with selectinload and joinedload is used")
|
|
print("✅ Pre-fetching patterns are implemented")
|
|
print("✅ Backward compatibility is maintained")
|
|
else:
|
|
print(f"❌ {total - passed} tests failed")
|
|
|
|
return passed == total
|
|
|
|
except Exception as e:
|
|
print(f"❌ Test suite failed: {e}")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
exit(0 if success else 1) |