63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
import httpx
|
||
import asyncio
|
||
import time
|
||
|
||
# Testing against the running service on localhost
|
||
BASE_URL = "http://localhost:8000"
|
||
|
||
async def test_ai_proxy_integration():
|
||
async with httpx.AsyncClient(base_url=BASE_URL, timeout=30.0) as client:
|
||
print("--- Starting AI Proxy Integration Test ---")
|
||
|
||
# 1. Register a new user
|
||
username = f"user_{int(time.time())}"
|
||
print(f"1. Registering user: {username}")
|
||
reg_res = await client.post("/register", json={
|
||
"username": username,
|
||
"password": "testpassword",
|
||
"email": f"user_{int(time.time())}@example.com"
|
||
})
|
||
if reg_res.status_code != 200:
|
||
print(f"Registration failed: {reg_res.text}")
|
||
return
|
||
|
||
# 2. Login to get token
|
||
print("2. Logging in...")
|
||
login_res = await client.post("/login", json={
|
||
"username": username,
|
||
"password": "testpassword"
|
||
})
|
||
if login_res.status_code != 200:
|
||
print(f"Login failed: {login_res.text}")
|
||
return
|
||
token = login_res.json()["access_token"]
|
||
|
||
# 3. Request AI Proxy
|
||
print("3. Sending AI Proxy request...")
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
ai_request = {
|
||
"messages": [
|
||
{"role": "user", "content": "Tell me a joke."}
|
||
],
|
||
"model": "some-model"
|
||
}
|
||
|
||
try:
|
||
response = await client.post("/ai/proxy", json=ai_request, headers=headers)
|
||
print(f"Response Status Code: {response.status_code}")
|
||
print(f"Response Content: {response.text[:200]}...") # Print first 200 chars
|
||
|
||
if response.status_code == 200:
|
||
print("✅ Success: AI Proxy returned 200 OK")
|
||
elif response.status_code in [400, 401]:
|
||
print("ℹ️ Proxy worked, but AI provider returned error (likely invalid/missing API key)")
|
||
else:
|
||
print(f"❌ Unexpected status code: {response.status_code}")
|
||
except Exception as e:
|
||
print(f"❌ Request failed: {str(e)}")
|
||
|
||
print("--- Test Completed ---")
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(test_ai_proxy_integration())
|