e75c4d6f07
- Remove session-ses_2f27.md (161KB raw session log) - Remove 49 ROOT_* duplicate files across REFERENCE/ - Remove 14 duplicate files between REFERENCE/ root and history/ - Remove asr_legacy.rs (dead code, replaced by asr.rs) - Remove src/core/worker/ (duplicate JobWorker) - Remove src/core/layers/ (empty directory) - Remove 4 .bak files in src/ - Remove 7 dead private methods in worker/processor.rs - Remove backup directory from git tracking
130 lines
3.5 KiB
Python
130 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple API test to check basic functionality
|
|
"""
|
|
|
|
import requests
|
|
import time
|
|
|
|
BASE_URL = "http://localhost:3002"
|
|
API_KEY = "muser_243c6725b09f43e29f319a648645b992_1774874668_f224a6d2"
|
|
|
|
|
|
def test_endpoint(endpoint, method="GET", data=None):
|
|
"""Test a single endpoint"""
|
|
print(f"\n🔍 Testing {method} {endpoint}...")
|
|
|
|
headers = {"X-API-Key": API_KEY}
|
|
|
|
try:
|
|
if method == "GET":
|
|
response = requests.get(
|
|
f"{BASE_URL}{endpoint}", headers=headers, timeout=10
|
|
)
|
|
elif method == "POST":
|
|
headers["Content-Type"] = "application/json"
|
|
response = requests.post(
|
|
f"{BASE_URL}{endpoint}", headers=headers, json=data, timeout=10
|
|
)
|
|
else:
|
|
print(f"❌ Unsupported method: {method}")
|
|
return False
|
|
|
|
print(f"Status: {response.status_code}")
|
|
print(f"Headers: {dict(response.headers)}")
|
|
|
|
if response.status_code == 200:
|
|
print("✅ Success!")
|
|
if response.text:
|
|
print(f"Response (first 500 chars): {response.text[:500]}")
|
|
return True
|
|
elif response.status_code == 404:
|
|
print(f"⚠️ Endpoint not found: {endpoint}")
|
|
return False
|
|
else:
|
|
print(
|
|
f"❌ Failed: {response.text[:200] if response.text else 'No response body'}"
|
|
)
|
|
return False
|
|
|
|
except requests.exceptions.Timeout:
|
|
print("❌ Request timeout")
|
|
return False
|
|
except requests.exceptions.ConnectionError:
|
|
print("❌ Connection error")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print("🧪 Simple API Functionality Test")
|
|
print("=" * 60)
|
|
|
|
# Wait for server to be ready
|
|
print("⏳ Waiting for server to be ready...")
|
|
time.sleep(3)
|
|
|
|
# Test endpoints in order
|
|
endpoints = [
|
|
("/api/v1/face/list", "GET"),
|
|
("/api/v1/face/results/384b0ff44aaaa1f1", "GET"),
|
|
("/api/v1/health", "GET"),
|
|
("/", "GET"),
|
|
]
|
|
|
|
success_count = 0
|
|
total_count = len(endpoints)
|
|
|
|
for endpoint, method in endpoints:
|
|
if test_endpoint(endpoint, method):
|
|
success_count += 1
|
|
|
|
print("\n" + "=" * 60)
|
|
print(f"📊 Results: {success_count}/{total_count} endpoints working")
|
|
|
|
if success_count == total_count:
|
|
print("✅ All tests passed!")
|
|
else:
|
|
print("⚠️ Some tests failed. Check server logs for details.")
|
|
|
|
print("=" * 60)
|
|
|
|
# Check database connection
|
|
print("\n🗄️ Checking database connection...")
|
|
try:
|
|
import psycopg2
|
|
|
|
conn = psycopg2.connect(
|
|
host="localhost",
|
|
port=5432,
|
|
database="momentry",
|
|
user="accusys",
|
|
password="accusys",
|
|
)
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT version();")
|
|
version = cursor.fetchone()
|
|
print(f"✅ PostgreSQL connected: {version[0]}")
|
|
|
|
# Check face tables
|
|
cursor.execute("SELECT COUNT(*) FROM face_identities;")
|
|
face_count = cursor.fetchone()[0]
|
|
print(f" face_identities: {face_count} rows")
|
|
|
|
cursor.execute("SELECT COUNT(*) FROM face_detections;")
|
|
detections_count = cursor.fetchone()[0]
|
|
print(f" face_detections: {detections_count} rows")
|
|
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
except Exception as e:
|
|
print(f"❌ Database connection failed: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|