-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_debug.py
More file actions
114 lines (98 loc) · 3.7 KB
/
test_debug.py
File metadata and controls
114 lines (98 loc) · 3.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
#!/usr/bin/env python3
"""
Test the API after it's running to identify the 500 error.
"""
import requests
import json
import time
import logging
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def wait_for_server(max_attempts=30, delay=2):
"""Wait for the server to be ready."""
base_url = "http://localhost:8000"
for attempt in range(max_attempts):
try:
response = requests.get(f"{base_url}/health", timeout=5)
if response.status_code == 200:
logger.info("✅ Server is ready!")
return True
except requests.exceptions.ConnectionError:
logger.info(f"⏳ Waiting for server... (attempt {attempt + 1}/{max_attempts})")
time.sleep(delay)
except Exception as e:
logger.warning(f"Unexpected error checking server: {e}")
time.sleep(delay)
logger.error("❌ Server did not start within the expected time")
return False
def test_api():
"""Test the API with a simple request."""
base_url = "http://localhost:8000"
# Test data
test_request = {
"user_id": "test_user",
"comments": [
{
"comment_id": "1",
"comment": "This is a test comment to see if the API works.",
"created_at": "2024-01-01T00:00:00Z",
"parent_comment": None,
"child_comment": None
},
{
"comment_id": "2",
"comment": "Another test comment for debugging purposes.",
"created_at": "2024-01-01T00:01:00Z",
"parent_comment": None,
"child_comment": None
}
],
"options": {
"fast_only": True,
"include_breakdown": True,
"use_context": False,
"force_full_analysis": False
}
}
try:
# Test health endpoint first
logger.info("Testing health endpoint...")
health_response = requests.get(f"{base_url}/health", timeout=10)
logger.info(f"Health check status: {health_response.status_code}")
logger.info(f"Health check response: {health_response.json()}")
# Test the analyze endpoint
logger.info("Testing analyze endpoint...")
logger.info(f"Sending request: {json.dumps(test_request, indent=2)}")
response = requests.post(
f"{base_url}/api/v1/analyze/user",
json=test_request,
timeout=60 # Longer timeout for model loading
)
logger.info(f"Response status: {response.status_code}")
logger.info(f"Response headers: {dict(response.headers)}")
if response.status_code == 200:
logger.info("✅ API request successful!")
logger.info(f"Response: {json.dumps(response.json(), indent=2)}")
else:
logger.error(f"❌ API request failed with status {response.status_code}")
logger.error(f"Response text: {response.text}")
except requests.exceptions.ConnectionError:
logger.error("❌ Could not connect to API. Make sure the server is running on localhost:8000")
except requests.exceptions.Timeout:
logger.error("❌ Request timed out")
except Exception as e:
logger.error(f"❌ Unexpected error: {e}")
def main():
"""Main test function."""
logger.info("🧪 Starting API debug test...")
# Wait for server to be ready
if not wait_for_server():
return
# Test the API
test_api()
if __name__ == "__main__":
main()