tegwick
ef62cb3872
Meter-gated removal (7 consecutive zero-usage windows reached). Deletes:
- MCP tools: create_workstream, list_workstreams, update_workstream,
update_workstream_status
- MCP resource: state://workstreams/{topic_slug}
- Dead helpers: _LEGACY_MCP_* maps, _legacy_mcp_deprecation,
_meter_legacy_mcp, _attach_legacy_deprecation, _update_workplan_legacy_impl
TOOLS.md keeps a retired→preferred migration map (per backlog). Removed the
now-dead STATEHUB_MCP_LEGACY_METER guard in conftest. Retargeted the
create_workplan error-skip test to the preferred tool; dropped alias-only tests.
Staged on branch state-wp-0070-legacy-retirement — do not merge until the 7th
documented zero-usage window is captured.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
378 lines
14 KiB
Python
378 lines
14 KiB
Python
"""
|
|
Regression tests for State Hub MCP write tools.
|
|
|
|
These call the registered FastMCP tools in-process and monkeypatch only the
|
|
thin HTTP helpers. That exercises FastMCP argument validation and the MCP
|
|
wrapper logic without depending on a running State Hub API.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import mcp_server.server as server
|
|
|
|
|
|
async def _call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
|
|
result = await server.mcp.call_tool(name, arguments)
|
|
return json.loads(result.content[0].text)
|
|
|
|
|
|
class TestMCPWriteTools:
|
|
async def test_create_workplan_returns_rest_shape_and_emits_progress(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
if path == "/workplans":
|
|
return {
|
|
"id": "wp-1",
|
|
"repo_id": body["repo_id"],
|
|
"topic_id": body.get("topic_id"),
|
|
"title": body["title"],
|
|
"slug": body["slug"],
|
|
"status": body["status"],
|
|
}
|
|
if path == "/progress":
|
|
return {"id": "event-1", **body}
|
|
raise AssertionError(f"unexpected POST {path}")
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"create_workplan",
|
|
{"repo_id": "repo-1", "topic_id": "topic-1", "title": "MCP Reliable Write"},
|
|
)
|
|
|
|
assert body == {
|
|
"id": "wp-1",
|
|
"repo_id": "repo-1",
|
|
"topic_id": "topic-1",
|
|
"title": "MCP Reliable Write",
|
|
"slug": "mcp-reliable-write",
|
|
"status": "active",
|
|
}
|
|
assert [path for path, _ in calls] == ["/workplans", "/progress"]
|
|
assert calls[1][1]["workplan_id"] == "wp-1"
|
|
assert calls[1][1]["event_type"] == "workplan_created"
|
|
|
|
async def test_create_task_returns_rest_shape_and_emits_progress(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
if path == "/tasks":
|
|
return {
|
|
"id": "task-1",
|
|
"workplan_id": body.get("workplan_id") or body.get("workstream_id"),
|
|
"workstream_id": body.get("workplan_id") or body.get("workstream_id"),
|
|
"title": body["title"],
|
|
"priority": body["priority"],
|
|
"status": "todo",
|
|
}
|
|
if path == "/progress":
|
|
return {"id": "event-1", **body}
|
|
raise AssertionError(f"unexpected POST {path}")
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"create_task",
|
|
{"workstream_id": "ws-1", "title": "MCP task", "priority": "high"},
|
|
)
|
|
|
|
assert body == {
|
|
"id": "task-1",
|
|
"workplan_id": "ws-1",
|
|
"workstream_id": "ws-1",
|
|
"title": "MCP task",
|
|
"priority": "high",
|
|
"status": "todo",
|
|
}
|
|
assert [path for path, _ in calls] == ["/tasks", "/progress"]
|
|
assert calls[1][1]["task_id"] == "task-1"
|
|
assert calls[1][1]["workplan_id"] == "ws-1"
|
|
assert "workstream_id" not in calls[1][1]
|
|
assert calls[1][1]["event_type"] == "task_created"
|
|
|
|
async def test_update_task_status_api_error_is_clear_and_skips_progress(self, monkeypatch):
|
|
post_calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_patch(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
assert path == "/tasks/task-1"
|
|
assert body["status"] == "done"
|
|
return {"error": "API 404: task not found"}
|
|
|
|
monkeypatch.setattr(server, "_patch", fake_patch)
|
|
monkeypatch.setattr(server, "_post", lambda path, body: post_calls.append((path, body)))
|
|
|
|
body = await _call_tool(
|
|
"update_task_status",
|
|
{"task_id": "task-1", "status": "done"},
|
|
)
|
|
|
|
assert body["tool"] == "update_task_status"
|
|
assert body["error"] == "API 404: task not found"
|
|
assert post_calls == []
|
|
|
|
async def test_update_task_status_returns_rest_shape_and_emits_progress(self, monkeypatch):
|
|
post_calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_patch(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
assert path == "/tasks/task-1"
|
|
assert body["status"] == "done"
|
|
return {
|
|
"id": "task-1",
|
|
"workstream_id": "ws-1",
|
|
"title": "Finish MCP reliability",
|
|
"status": "done",
|
|
}
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
post_calls.append((path, body))
|
|
return {"id": "event-1", **body}
|
|
|
|
monkeypatch.setattr(server, "_patch", fake_patch)
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"update_task_status",
|
|
{"task_id": "task-1", "status": "done"},
|
|
)
|
|
|
|
assert body == {
|
|
"id": "task-1",
|
|
"workstream_id": "ws-1",
|
|
"title": "Finish MCP reliability",
|
|
"status": "done",
|
|
}
|
|
assert [path for path, _ in post_calls] == ["/progress"]
|
|
assert post_calls[0][1]["task_id"] == "task-1"
|
|
assert post_calls[0][1]["workplan_id"] == "ws-1"
|
|
assert "workstream_id" not in post_calls[0][1]
|
|
assert post_calls[0][1]["event_type"] == "task_status_changed"
|
|
|
|
async def test_add_progress_event_accepts_json_string_detail(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
return {"id": "event-1", **body}
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"add_progress_event",
|
|
{
|
|
"summary": "MCP progress",
|
|
"event_type": "note",
|
|
"workstream_id": "ws-1",
|
|
"detail": '{"files_changed": 3}',
|
|
},
|
|
)
|
|
|
|
assert body["id"] == "event-1"
|
|
assert body["detail"] == {"files_changed": 3}
|
|
assert calls == [
|
|
(
|
|
"/progress",
|
|
{
|
|
"topic_id": None,
|
|
"workplan_id": "ws-1",
|
|
"task_id": None,
|
|
"event_type": "note",
|
|
"summary": "MCP progress",
|
|
"author": "custodian",
|
|
"detail": {"files_changed": 3},
|
|
},
|
|
)
|
|
]
|
|
|
|
async def test_bulk_update_task_statuses_returns_rest_shape(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
assert path == "/tasks/bulk-status-sync"
|
|
return {
|
|
"updated": [
|
|
{"id": "task-1", "title": "First", "status": "done"},
|
|
{"id": "task-2", "title": "Second", "status": "wait"},
|
|
],
|
|
"progress_event_ids": ["event-1", "event-2"],
|
|
}
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"bulk_update_task_statuses",
|
|
{
|
|
"author": "codex",
|
|
"session_id": "session-1",
|
|
"updates": [
|
|
{"task_id": "task-1", "status": "done"},
|
|
{"task_id": "task-2", "status": "wait", "blocking_reason": "needs input"},
|
|
],
|
|
},
|
|
)
|
|
|
|
assert body["progress_event_ids"] == ["event-1", "event-2"]
|
|
assert [task["status"] for task in body["updated"]] == ["done", "wait"]
|
|
assert calls == [
|
|
(
|
|
"/tasks/bulk-status-sync",
|
|
{
|
|
"updates": [
|
|
{"task_id": "task-1", "status": "done"},
|
|
{"task_id": "task-2", "status": "wait", "blocking_reason": "needs input"},
|
|
],
|
|
"author": "codex",
|
|
"session_id": "session-1",
|
|
},
|
|
)
|
|
]
|
|
|
|
async def test_record_decision_returns_rest_shape_and_emits_progress(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
if path == "/decisions":
|
|
return {
|
|
"id": "decision-1",
|
|
"title": body["title"],
|
|
"decision_type": body["decision_type"],
|
|
"topic_id": body["topic_id"],
|
|
"workplan_id": body.get("workplan_id"),
|
|
"status": "open",
|
|
"escalation_note": None,
|
|
}
|
|
if path == "/progress":
|
|
return {"id": "event-1", **body}
|
|
raise AssertionError(f"unexpected POST {path}")
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"record_decision",
|
|
{
|
|
"title": "Keep MCP stateless",
|
|
"decision_type": "made",
|
|
"topic_id": "topic-1",
|
|
"workstream_id": "ws-1",
|
|
},
|
|
)
|
|
|
|
assert body["id"] == "decision-1"
|
|
assert body["title"] == "Keep MCP stateless"
|
|
assert [path for path, _ in calls] == ["/decisions", "/progress"]
|
|
assert calls[1][1]["decision_id"] == "decision-1"
|
|
assert calls[1][1]["event_type"] == "decision_recorded"
|
|
|
|
async def test_create_workplan_api_error_skips_progress(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
return {"error": "API 422: invalid repo"}
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"create_workplan",
|
|
{"repo_id": "bad-repo", "title": "No progress on failure"},
|
|
)
|
|
|
|
assert body["tool"] == "create_workplan"
|
|
assert body["error"] == "API 422: invalid repo"
|
|
assert [path for path, _ in calls] == ["/workplans"]
|
|
|
|
async def test_record_decision_missing_id_is_clear_and_skips_progress(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
return {"title": body["title"], "status": "open"}
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"record_decision",
|
|
{"title": "Malformed API response", "topic_id": "topic-1"},
|
|
)
|
|
|
|
assert body["tool"] == "record_decision"
|
|
assert body["error"] == "API response missing required field(s): id"
|
|
assert [path for path, _ in calls] == ["/decisions"]
|
|
|
|
async def test_record_decision_accepts_workplan_id_alias(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
if path == "/decisions":
|
|
return {
|
|
"id": "decision-1",
|
|
"title": body["title"],
|
|
"decision_type": body["decision_type"],
|
|
"topic_id": body["topic_id"],
|
|
"workplan_id": body["workplan_id"],
|
|
"status": "open",
|
|
"escalation_note": None,
|
|
}
|
|
if path == "/progress":
|
|
return {"id": "event-1", **body}
|
|
raise AssertionError(f"unexpected POST {path}")
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
|
|
body = await _call_tool(
|
|
"record_decision",
|
|
{
|
|
"title": "Alias check",
|
|
"decision_type": "made",
|
|
"topic_id": "topic-1",
|
|
"workplan_id": "wp-1",
|
|
},
|
|
)
|
|
|
|
assert body["id"] == "decision-1"
|
|
assert calls[0][1]["workplan_id"] == "wp-1"
|
|
assert calls[1][1]["workplan_id"] == "wp-1"
|
|
assert "workstream_id" not in calls[1][1]
|
|
|
|
async def test_list_blocked_tasks_accepts_workplan_id_alias(self, monkeypatch):
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_get(path: str, params: dict[str, Any] | None = None) -> Any:
|
|
captured["path"] = path
|
|
captured["params"] = params
|
|
return []
|
|
|
|
monkeypatch.setattr(server, "_get", fake_get)
|
|
|
|
await _call_tool("list_blocked_tasks", {"workplan_id": "wp-1"})
|
|
|
|
assert captured["params"]["workplan_id"] == "wp-1"
|
|
assert "workstream_id" not in captured["params"]
|
|
|
|
async def test_record_token_event_accepts_workplan_id_alias(self, monkeypatch):
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
|
|
calls.append((path, body))
|
|
return {"id": "token-1", "tokens_total": 1500}
|
|
|
|
def fake_get(path: str, params: dict[str, Any] | None = None) -> Any:
|
|
return {"tokens_total": 1500, "event_count": 1}
|
|
|
|
monkeypatch.setattr(server, "_post", fake_post)
|
|
monkeypatch.setattr(server, "_get", fake_get)
|
|
|
|
body = await _call_tool(
|
|
"record_token_event",
|
|
{"tokens_in": 1000, "tokens_out": 500, "workplan_id": "wp-1"},
|
|
)
|
|
|
|
assert body["event_id"] == "token-1"
|
|
assert calls[0][1]["workplan_id"] == "wp-1"
|