Adriano Dal Pastro
e54f828722
session_id "new" now returns null at creation (set from SDK after completion). Session tests use custom session_ids directly. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
115 lines
3.0 KiB
Python
115 lines
3.0 KiB
Python
import pytest
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
from src.backend.main import app
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
async def client(app):
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(
|
|
transport=transport,
|
|
base_url="http://test",
|
|
headers={"X-Api-Key": "test-key-12345"},
|
|
) as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest.fixture
|
|
async def topic_id(client):
|
|
resp = await client.post(
|
|
"/api/topics",
|
|
json={"name": "test-topic", "system_prompt": "You are helpful."},
|
|
)
|
|
return resp.json()["data"]["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_request(client, topic_id):
|
|
resp = await client.post(
|
|
"/api/requests",
|
|
json={"prompt": "Hello", "topic_id": topic_id},
|
|
)
|
|
assert resp.status_code == 202
|
|
data = resp.json()
|
|
assert data["success"] is True
|
|
assert data["data"]["status"] == "pending"
|
|
assert "id" in data["data"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_request_with_new_session(client, topic_id):
|
|
resp = await client.post(
|
|
"/api/requests",
|
|
json={"prompt": "Hello", "topic_id": topic_id, "session_id": "new"},
|
|
)
|
|
assert resp.status_code == 202
|
|
data = resp.json()
|
|
# session_id is None at creation; gets set from SDK after completion
|
|
assert data["data"]["session_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_request_invalid_topic_fails(client):
|
|
resp = await client.post(
|
|
"/api/requests",
|
|
json={"prompt": "Hello", "topic_id": "nonexistent"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_request(client, topic_id):
|
|
create_resp = await client.post(
|
|
"/api/requests",
|
|
json={"prompt": "Hello", "topic_id": topic_id},
|
|
)
|
|
req_id = create_resp.json()["data"]["id"]
|
|
|
|
resp = await client.get(f"/api/requests/{req_id}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data"]["prompt"] == "Hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_requests_with_filters(client, topic_id):
|
|
await client.post(
|
|
"/api/requests",
|
|
json={"prompt": "First", "topic_id": topic_id},
|
|
)
|
|
await client.post(
|
|
"/api/requests",
|
|
json={"prompt": "Second", "topic_id": topic_id},
|
|
)
|
|
|
|
resp = await client.get("/api/requests")
|
|
assert len(resp.json()["data"]) == 2
|
|
|
|
resp = await client.get(f"/api/requests?topic_id={topic_id}")
|
|
assert len(resp.json()["data"]) == 2
|
|
|
|
resp = await client.get("/api/requests?status=completed")
|
|
assert len(resp.json()["data"]) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_pending_request(client, topic_id):
|
|
create_resp = await client.post(
|
|
"/api/requests",
|
|
json={"prompt": "Hello", "topic_id": topic_id},
|
|
)
|
|
req_id = create_resp.json()["data"]["id"]
|
|
|
|
resp = await client.delete(f"/api/requests/{req_id}")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_nonexistent_request_fails(client):
|
|
resp = await client.delete("/api/requests/nonexistent")
|
|
assert resp.status_code == 404
|