64 lines
2.4 KiB
Python
64 lines
2.4 KiB
Python
import asyncio
|
|
import httpx
|
|
|
|
BASE_URL = "http://localhost:8002"
|
|
|
|
async def test_text_query(session_id: int):
|
|
"""Send a POST request to the /api/text endpoint."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
# Correctly send form data
|
|
response = await client.post(
|
|
"http://localhost:8002/api/text",
|
|
data={"question": f"Test question: what is AI from session {session_id}"}
|
|
)
|
|
print(f"Session {session_id} Status Code: {response.status_code}")
|
|
print(f"Session {session_id} Headers: {response.headers}")
|
|
if response.content:
|
|
print(f"Session {session_id} Response: {response.json()}")
|
|
else:
|
|
print(f"Session {session_id} Response is empty.")
|
|
except Exception as e:
|
|
print(f"Session {session_id} encountered an error: {e}")
|
|
|
|
|
|
async def test_image_query(session_id: int):
|
|
"""Send a POST request to the /api/image endpoint with a dummy image."""
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
files = {'file': ('../profile/1.jpg', b'binary data of the file', 'image/jpeg')}
|
|
response = await client.post(
|
|
f"{BASE_URL}/api/image",
|
|
data={"question": f"Image query: what is in the picture from session {session_id}"},
|
|
files=files
|
|
)
|
|
print(f"Session {session_id} Raw Response: {response.text}")
|
|
print(f"Session {session_id} Response: {response.json()}")
|
|
|
|
|
|
async def test_video_query(session_id: int):
|
|
"""Send a POST request to the /api/video endpoint with a dummy video."""
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
files = {'file': ('../video/1.mp4', b'binary data of the file', 'video/mp4')}
|
|
response = await client.post(
|
|
f"{BASE_URL}/api/video",
|
|
data={"question": f"Video query: what is in the video from session {session_id}"},
|
|
files=files
|
|
)
|
|
print(f"Session {session_id} Raw Response: {response.text}")
|
|
print(f"Session {session_id} Response: {response.json()}")
|
|
|
|
|
|
async def main():
|
|
tasks = []
|
|
for i in range(1, 3):
|
|
tasks.append(test_text_query(i))
|
|
# tasks.append(test_image_query(i))
|
|
# tasks.append(test_video_query(i))
|
|
|
|
# Run all tasks concurrently
|
|
await asyncio.gather(*tasks)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|