145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
"""End-to-end tests against the JSON API using FastAPI's TestClient.
|
|
|
|
These supersede the old direct-DB CLI smoke tests — they exercise the same
|
|
service.py paths the CLI now goes through over HTTP.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient:
|
|
monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "data"))
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
|
|
# Force re-import after env vars set so config picks up our tmp paths.
|
|
from claire.web.app import create_app
|
|
|
|
return TestClient(create_app())
|
|
|
|
|
|
def test_health(client: TestClient) -> None:
|
|
resp = client.get("/api/v1/health")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] == "ok"
|
|
assert len(body["machine_id"]) == 36
|
|
|
|
|
|
def test_project_lifecycle(client: TestClient) -> None:
|
|
# Empty list to start.
|
|
assert client.get("/api/v1/projects").json() == {"projects": []}
|
|
|
|
# Create.
|
|
r = client.post("/api/v1/projects", json={"name": "alpha", "goal": "ship it"})
|
|
assert r.status_code == 201, r.text
|
|
project = r.json()
|
|
assert project["name"] == "alpha"
|
|
assert project["goal"] == "ship it"
|
|
assert project["status"] == "active"
|
|
|
|
# Duplicate → 409.
|
|
r = client.post("/api/v1/projects", json={"name": "alpha"})
|
|
assert r.status_code == 409
|
|
|
|
# Show.
|
|
r = client.get("/api/v1/projects/alpha")
|
|
assert r.status_code == 200
|
|
assert r.json()["project"]["name"] == "alpha"
|
|
assert r.json()["tasks"] == []
|
|
|
|
# Missing → 404.
|
|
assert client.get("/api/v1/projects/zzz").status_code == 404
|
|
|
|
|
|
def test_task_create_and_done(client: TestClient) -> None:
|
|
client.post("/api/v1/projects", json={"name": "alpha"})
|
|
r = client.post(
|
|
"/api/v1/tasks",
|
|
json={"project": "alpha", "title": "do thing", "priority": 1},
|
|
)
|
|
assert r.status_code == 201
|
|
task = r.json()
|
|
assert task["priority"] == 1
|
|
assert task["status"] == "todo"
|
|
tid = task["id"]
|
|
|
|
# Mark done via task update.
|
|
r = client.post(f"/api/v1/tasks/{tid}", json={"status": "done"})
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "done"
|
|
|
|
# task list with status filter.
|
|
r = client.get("/api/v1/tasks", params={"status": "done"})
|
|
tasks = r.json()["tasks"]
|
|
assert len(tasks) == 1 and tasks[0]["id"] == tid
|
|
|
|
|
|
def test_assignment_creation(client: TestClient) -> None:
|
|
client.post("/api/v1/projects", json={"name": "alpha"})
|
|
tid = client.post(
|
|
"/api/v1/tasks", json={"project": "alpha", "title": "t"}
|
|
).json()["id"]
|
|
sid = "11111111-1111-4111-8111-111111111111"
|
|
r = client.post(
|
|
"/api/v1/assignments",
|
|
json={"task_id": tid, "session_uuid": sid},
|
|
)
|
|
assert r.status_code == 201
|
|
a = r.json()
|
|
assert a["task_id"] == tid
|
|
assert a["session_uuid"] == sid
|
|
|
|
# Surfaced in /assignments.
|
|
r = client.get("/api/v1/assignments")
|
|
assert len(r.json()["assignments"]) == 1
|
|
|
|
|
|
def test_status_rollup(client: TestClient) -> None:
|
|
client.post("/api/v1/projects", json={"name": "alpha"})
|
|
client.post("/api/v1/tasks", json={"project": "alpha", "title": "t1", "priority": 1})
|
|
client.post("/api/v1/tasks", json={"project": "alpha", "title": "t2"})
|
|
r = client.get("/api/v1/status")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
proj = next(p for p in body["projects"] if p["name"] == "alpha")
|
|
assert proj["counts"]["todo"] == 2
|
|
assert proj["counts"]["done"] == 0
|
|
|
|
|
|
def test_fleet_considered_shape(client: TestClient) -> None:
|
|
"""GET /fleet/considered returns the considered-work shape; an open
|
|
task with no free session lands in `remaining_tasks`, not `pairings`."""
|
|
client.post("/api/v1/projects", json={"name": "alpha"})
|
|
client.post(
|
|
"/api/v1/tasks", json={"project": "alpha", "title": "ship hud", "priority": 0}
|
|
)
|
|
r = client.get("/api/v1/fleet/considered")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert set(body) == {
|
|
"pairings",
|
|
"remaining_tasks",
|
|
"remaining_sessions",
|
|
"capped_out",
|
|
}
|
|
assert body["pairings"] == [] # no sessions → nothing pairs
|
|
titles = [t["title"] for t in body["remaining_tasks"]]
|
|
assert "ship hud" in titles
|
|
|
|
|
|
def test_broadcast_dry_run_warns_when_no_assignments(client: TestClient) -> None:
|
|
client.post("/api/v1/projects", json={"name": "alpha"})
|
|
r = client.post(
|
|
"/api/v1/broadcast", json={"target": "alpha", "text": "hello", "yes": False}
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["sent"] is False
|
|
assert body["delivered"] == 0
|
|
assert "no active assignments" in (body["error"] or "")
|