127 lines
4.8 KiB
Python
127 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
API integration test for data consistency endpoints.
|
|
|
|
This script tests the REST API endpoints for data consistency validation.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import requests
|
|
import json
|
|
from datetime import datetime
|
|
|
|
# Test configuration
|
|
BASE_URL = "http://localhost:8000"
|
|
TEST_EMAIL = "admin@example.com"
|
|
TEST_PASSWORD = "admin123"
|
|
|
|
|
|
def get_auth_token():
|
|
"""Get authentication token for API requests."""
|
|
login_data = {
|
|
"username": TEST_EMAIL,
|
|
"password": TEST_PASSWORD
|
|
}
|
|
|
|
response = requests.post(f"{BASE_URL}/auth/login", data=login_data)
|
|
if response.status_code == 200:
|
|
return response.json()["access_token"]
|
|
else:
|
|
print(f"Failed to authenticate: {response.status_code} - {response.text}")
|
|
return None
|
|
|
|
|
|
def test_consistency_endpoints():
|
|
"""Test data consistency API endpoints."""
|
|
print("=== Testing Data Consistency API Endpoints ===")
|
|
|
|
# Get authentication token
|
|
token = get_auth_token()
|
|
if not token:
|
|
print("❌ Failed to get authentication token")
|
|
return False
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
try:
|
|
# Test health check endpoint
|
|
print("\n1. Testing consistency health check...")
|
|
response = requests.get(f"{BASE_URL}/data-consistency/health", headers=headers)
|
|
if response.status_code == 200:
|
|
health_data = response.json()
|
|
print(f"✓ Health check successful: {health_data['status']}")
|
|
print(f" Consistency: {health_data.get('consistency_percentage', 0):.1f}%")
|
|
else:
|
|
print(f"❌ Health check failed: {response.status_code}")
|
|
return False
|
|
|
|
# Test consistency report endpoint
|
|
print("\n2. Testing consistency report...")
|
|
response = requests.get(f"{BASE_URL}/data-consistency/report", headers=headers)
|
|
if response.status_code == 200:
|
|
report_data = response.json()
|
|
print(f"✓ Report generated successfully")
|
|
print(f" Total entities: {report_data['summary']['total_entities']}")
|
|
print(f" Valid entities: {report_data['summary']['valid_entities']}")
|
|
print(f" Consistency: {report_data['summary']['consistency_percentage']:.1f}%")
|
|
else:
|
|
print(f"❌ Report generation failed: {response.status_code}")
|
|
return False
|
|
|
|
# Test individual entity validation (if we have entities)
|
|
if report_data['summary']['total_entities'] > 0:
|
|
print("\n3. Testing individual entity validation...")
|
|
|
|
# Try to validate a shot if available
|
|
if report_data['shots']['total_entities'] > 0:
|
|
# Get first shot ID from the system (this is a simplified approach)
|
|
# In a real test, you'd have specific test data
|
|
shot_id = 1 # Assuming there's at least one shot with ID 1
|
|
response = requests.get(f"{BASE_URL}/data-consistency/validate/shot/{shot_id}", headers=headers)
|
|
if response.status_code == 200:
|
|
validation_data = response.json()
|
|
print(f"✓ Shot validation: {'VALID' if validation_data['valid'] else 'INVALID'}")
|
|
else:
|
|
print(f"⚠ Shot validation failed (might not exist): {response.status_code}")
|
|
|
|
# Try to validate an asset if available
|
|
if report_data['assets']['total_entities'] > 0:
|
|
asset_id = 1 # Assuming there's at least one asset with ID 1
|
|
response = requests.get(f"{BASE_URL}/data-consistency/validate/asset/{asset_id}", headers=headers)
|
|
if response.status_code == 200:
|
|
validation_data = response.json()
|
|
print(f"✓ Asset validation: {'VALID' if validation_data['valid'] else 'INVALID'}")
|
|
else:
|
|
print(f"⚠ Asset validation failed (might not exist): {response.status_code}")
|
|
|
|
print(f"\n✓ All API endpoint tests completed successfully")
|
|
return True
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
print("❌ Failed to connect to the API server. Make sure the server is running on http://localhost:8000")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ API test failed with error: {str(e)}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""Main test function."""
|
|
print("=== Data Consistency API Integration Test ===")
|
|
print(f"Test started at: {datetime.now()}")
|
|
print(f"Testing API at: {BASE_URL}")
|
|
|
|
success = test_consistency_endpoints()
|
|
|
|
if success:
|
|
print(f"\n=== All API Tests Completed Successfully ===")
|
|
else:
|
|
print(f"\n❌ API Tests Failed")
|
|
|
|
return success
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
sys.exit(0 if success else 1) |