File size: 2,638 Bytes
6a0ff33 | 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 | """
Tests for workflow creation and retrieval.
"""
class TestCreateWorkflow:
def test_create_workflow(self, client, auth_headers):
resp = client.post("/api/workflows", json={
"prompt": "Create a Python calculator",
}, headers=auth_headers)
assert resp.status_code == 201
data = resp.json()
assert data["prompt"] == "Create a Python calculator"
assert data["status"] == "pending"
assert "id" in data
def test_create_workflow_requires_auth(self, client):
resp = client.post("/api/workflows", json={
"prompt": "No auth test",
})
assert resp.status_code in (401, 403)
def test_create_workflow_requires_prompt(self, client, auth_headers):
resp = client.post("/api/workflows", json={}, headers=auth_headers)
assert resp.status_code == 422
class TestListWorkflows:
def test_list_empty(self, client, auth_headers):
resp = client.get("/api/workflows", headers=auth_headers)
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 0
assert data["workflows"] == []
def test_list_after_create(self, client, auth_headers):
client.post("/api/workflows", json={
"prompt": "Workflow 1",
}, headers=auth_headers)
client.post("/api/workflows", json={
"prompt": "Workflow 2",
}, headers=auth_headers)
resp = client.get("/api/workflows", headers=auth_headers)
data = resp.json()
assert data["total"] == 2
assert len(data["workflows"]) == 2
def test_list_requires_auth(self, client):
resp = client.get("/api/workflows")
assert resp.status_code in (401, 403)
class TestGetWorkflow:
def test_get_workflow_by_id(self, client, auth_headers):
create_resp = client.post("/api/workflows", json={
"prompt": "Get me by ID",
}, headers=auth_headers)
workflow_id = create_resp.json()["id"]
resp = client.get(f"/api/workflows/{workflow_id}", headers=auth_headers)
assert resp.status_code == 200
assert resp.json()["id"] == workflow_id
assert resp.json()["prompt"] == "Get me by ID"
def test_get_nonexistent_workflow(self, client, auth_headers):
resp = client.get(
"/api/workflows/00000000-0000-0000-0000-000000000000",
headers=auth_headers,
)
assert resp.status_code == 404
def test_get_workflow_requires_auth(self, client):
resp = client.get("/api/workflows/some-id")
assert resp.status_code in (401, 403)
|