generated from coulomb/repo-seed
230 lines
8.3 KiB
Python
230 lines
8.3 KiB
Python
"""
|
|
MCP server smoke tests — exercise the HTTP-level correctness of the core tools.
|
|
|
|
Rather than testing the MCP stdio protocol (which requires a running subprocess),
|
|
these tests call the underlying FastAPI endpoints that the MCP tools wrap.
|
|
This verifies that the request/response shapes the MCP tools depend on are correct.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _create_domain(client, slug="mcp-domain"):
|
|
r = await client.post("/domains/", json={"slug": slug, "name": "MCP Domain"})
|
|
assert r.status_code == 201
|
|
return r.json()
|
|
|
|
|
|
async def _create_topic(client, domain_slug="mcp-domain"):
|
|
r = await client.post("/topics/", json={
|
|
"slug": "mcp-topic", "title": "MCP Topic", "domain": domain_slug,
|
|
})
|
|
assert r.status_code == 201
|
|
return r.json()
|
|
|
|
|
|
async def _create_workstream(client, topic_id):
|
|
r = await client.post("/workstreams/", json={
|
|
"topic_id": topic_id, "slug": "mcp-ws", "title": "MCP Workstream",
|
|
})
|
|
assert r.status_code == 201
|
|
return r.json()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_state_summary — GET /state/summary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetStateSummary:
|
|
async def test_returns_valid_shape(self, client):
|
|
r = await client.get("/state/summary")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
# Required top-level fields
|
|
for key in ("open_workstreams", "blocking_decisions", "blocked_tasks",
|
|
"domains", "contribution_counts", "licence_risk_count"):
|
|
assert key in body, f"missing key: {key}"
|
|
|
|
async def test_empty_db_returns_zero_counts(self, client):
|
|
r = await client.get("/state/summary")
|
|
body = r.json()
|
|
assert body["open_workstreams"] == []
|
|
assert body["blocking_decisions"] == []
|
|
assert body["blocked_tasks"] == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_task — POST /tasks/
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCreateTask:
|
|
async def test_creates_task_with_defaults(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
ws = await _create_workstream(client, topic["id"])
|
|
|
|
r = await client.post("/tasks/", json={
|
|
"workstream_id": ws["id"],
|
|
"title": "MCP-created task",
|
|
})
|
|
assert r.status_code == 201
|
|
body = r.json()
|
|
assert body["title"] == "MCP-created task"
|
|
assert body["status"] == "todo"
|
|
assert body["priority"] == "medium"
|
|
|
|
async def test_create_task_with_priority(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
ws = await _create_workstream(client, topic["id"])
|
|
|
|
r = await client.post("/tasks/", json={
|
|
"workstream_id": ws["id"],
|
|
"title": "Urgent task",
|
|
"priority": "high",
|
|
})
|
|
assert r.status_code == 201
|
|
assert r.json()["priority"] == "high"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# update_task_status — PATCH /tasks/{id}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestUpdateTaskStatus:
|
|
async def test_transition_todo_to_done(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
ws = await _create_workstream(client, topic["id"])
|
|
r = await client.post("/tasks/", json={
|
|
"workstream_id": ws["id"], "title": "Task to complete",
|
|
})
|
|
t_id = r.json()["id"]
|
|
|
|
r2 = await client.patch(f"/tasks/{t_id}", json={"status": "done"})
|
|
assert r2.status_code == 200
|
|
assert r2.json()["status"] == "done"
|
|
|
|
async def test_get_task_not_found_returns_404(self, client):
|
|
import uuid
|
|
r = await client.get(f"/tasks/{uuid.uuid4()}")
|
|
assert r.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# add_progress_event — POST /progress/
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAddProgressEvent:
|
|
async def test_creates_progress_event(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
|
|
r = await client.post("/progress/", json={
|
|
"topic_id": topic["id"],
|
|
"summary": "Completed initial setup",
|
|
"event_type": "milestone",
|
|
})
|
|
assert r.status_code == 201
|
|
body = r.json()
|
|
assert body["summary"] == "Completed initial setup"
|
|
assert body["event_type"] == "milestone"
|
|
|
|
async def test_progress_with_detail_dict(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
|
|
r = await client.post("/progress/", json={
|
|
"topic_id": topic["id"],
|
|
"summary": "Event with data",
|
|
"event_type": "note",
|
|
"detail": {"files_changed": 3, "lines": 42},
|
|
})
|
|
assert r.status_code == 201
|
|
assert r.json()["detail"]["files_changed"] == 3
|
|
|
|
async def test_progress_without_topic_is_allowed(self, client):
|
|
r = await client.post("/progress/", json={
|
|
"summary": "Freeform progress note",
|
|
"event_type": "note",
|
|
})
|
|
assert r.status_code == 201
|
|
|
|
async def test_progress_list_filters_by_topic(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
|
|
scoped = await client.post("/progress/", json={
|
|
"topic_id": topic["id"],
|
|
"summary": "Topic scoped note",
|
|
"event_type": "note",
|
|
})
|
|
assert scoped.status_code == 201
|
|
freeform = await client.post("/progress/", json={
|
|
"summary": "Freeform note",
|
|
"event_type": "note",
|
|
})
|
|
assert freeform.status_code == 201
|
|
|
|
r = await client.get("/progress/", params={"topic_id": topic["id"]})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert [event["id"] for event in body] == [scoped.json()["id"]]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# flag_for_human — PATCH /tasks/{id} with needs_human=True
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFlagForHuman:
|
|
async def test_flag_adds_intervention_note(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
ws = await _create_workstream(client, topic["id"])
|
|
r = await client.post("/tasks/", json={
|
|
"workstream_id": ws["id"], "title": "Task needing help",
|
|
})
|
|
t_id = r.json()["id"]
|
|
|
|
r2 = await client.patch(f"/tasks/{t_id}", json={
|
|
"needs_human": True,
|
|
"intervention_note": "Needs security review before deploying",
|
|
})
|
|
assert r2.status_code == 200
|
|
body = r2.json()
|
|
assert body["needs_human"] is True
|
|
assert "security review" in body["intervention_note"]
|
|
|
|
async def test_flag_without_note_returns_422(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
ws = await _create_workstream(client, topic["id"])
|
|
r = await client.post("/tasks/", json={
|
|
"workstream_id": ws["id"], "title": "Task without note",
|
|
})
|
|
t_id = r.json()["id"]
|
|
|
|
r2 = await client.patch(f"/tasks/{t_id}", json={"needs_human": True})
|
|
assert r2.status_code == 422
|
|
|
|
async def test_clear_flag(self, client):
|
|
await _create_domain(client)
|
|
topic = await _create_topic(client)
|
|
ws = await _create_workstream(client, topic["id"])
|
|
r = await client.post("/tasks/", json={
|
|
"workstream_id": ws["id"], "title": "Task to clear",
|
|
})
|
|
t_id = r.json()["id"]
|
|
|
|
await client.patch(f"/tasks/{t_id}", json={
|
|
"needs_human": True, "intervention_note": "needs review",
|
|
})
|
|
r2 = await client.patch(f"/tasks/{t_id}", json={"needs_human": False})
|
|
assert r2.status_code == 200
|
|
assert r2.json()["needs_human"] is False
|