Files
momentry_core/scripts/simple_api_test.py
Warren 8f05a7c188 feat: update Python processors and add utility scripts
- Update ASR, face, OCR, pose processors
- Add release pre-flight check script
- Add synonym generation, chunk processing scripts
- Add face recognition, stamp search utilities
2026-04-30 15:07:49 +08:00

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(f"✅ 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()