ab-ms-core / tests /unit /test_query_timing.py
PupaClic
Add comprehensive unit tests for various endpoints and features
7ff20d3
#!/usr/bin/env python3
"""
Test script to verify query timing is working in the running application.
This script makes API requests and checks for timing information in:
1. Response headers (X-Process-Time)
2. Server logs (query and request timing)
Prerequisites:
- Application must be running: uvicorn app.app:app --reload --port 8124
- Database connection must be configured
Usage:
python test_query_timing.py
"""
import requests
import time
import sys
BASE_URL = "http://localhost:8124"
def test_endpoint(endpoint: str, description: str):
"""Test an endpoint and display timing information."""
print(f"\n{'='*60}")
print(f"Testing: {description}")
print(f"Endpoint: {endpoint}")
print(f"{'='*60}")
try:
start = time.time()
response = requests.get(f"{BASE_URL}{endpoint}", timeout=10)
client_time = time.time() - start
print(f"Status Code: {response.status_code}")
print(f"Client-side Time: {client_time:.3f}s")
# Check for X-Process-Time header
if 'X-Process-Time' in response.headers:
server_time = response.headers['X-Process-Time']
print(f"Server-side Time (X-Process-Time): {server_time}s")
print("βœ“ Timing header present")
else:
print("βœ— X-Process-Time header not found")
# Display content size
print(f"Response Size: {len(response.content)} bytes")
return True
except requests.exceptions.ConnectionError:
print("βœ— ERROR: Could not connect to server")
print(" Make sure the application is running:")
print(" uvicorn app.app:app --reload --port 8124")
return False
except Exception as e:
print(f"βœ— ERROR: {e}")
return False
def main():
"""Run timing tests on various endpoints."""
print("="*60)
print("Query Timing Verification Test")
print("="*60)
print("\nThis test will make requests to the running application")
print("and verify that timing information is being captured.")
print("\nCheck the application logs for query timing details.")
# Test various endpoints
endpoints = [
("/api/v1/dashboard/widgets/info", "Dashboard Widgets (multiple queries)"),
("/api/v1/reference/states", "Reference Data - States"),
("/api/v1/reference/project-statuses", "Reference Data - Project Statuses"),
]
success_count = 0
for endpoint, description in endpoints:
if test_endpoint(endpoint, description):
success_count += 1
time.sleep(0.5) # Brief pause between requests
# Summary
print(f"\n{'='*60}")
print("Test Summary")
print(f"{'='*60}")
print(f"Tests Passed: {success_count}/{len(endpoints)}")
if success_count == len(endpoints):
print("\nβœ“ All tests passed!")
print("\nQuery timing features are working correctly.")
print("\nCheck your server logs for detailed timing information:")
print(" - Query execution times from SQLAlchemy")
print(" - Request processing times from middleware")
print(" - Warnings for slow operations")
else:
print("\nβœ— Some tests failed")
print("\nTroubleshooting:")
print(" 1. Ensure the application is running on port 8124")
print(" 2. Check database connection is configured")
print(" 3. Verify no firewall/network issues")
print(f"\n{'='*60}")
print("Timing Information Details:")
print(f"{'='*60}")
print("In the server logs, look for:")
print(" - INFO sqlalchemy.query.timing - Medium queries (500ms-1s)")
print(" - WARNING sqlalchemy.query.timing - Slow queries (>1s)")
print(" - INFO app.request.timing - Medium requests (1s-2s)")
print(" - WARNING app.request.timing - Slow requests (>2s)")
print("\nResponse headers include:")
print(" - X-Process-Time: Server-side processing time in seconds")
return 0 if success_count == len(endpoints) else 1
if __name__ == "__main__":
sys.exit(main())