tegwick
242dd25253
_emit_progress_event strips legacy workstream_id from POST /progress/ payloads. Task and decision side-effect events now set workplan_id explicitly so automatic progress logging stays workplan-first.
94 lines
No EOL
3.2 KiB
Python
94 lines
No EOL
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import mcp_server.server as server
|
|
|
|
|
|
def test_normalize_progress_body_prefers_workplan_id_only() -> None:
|
|
body = server._normalize_progress_body({
|
|
"workplan_id": "wp-1",
|
|
"workstream_id": "wp-1",
|
|
"event_type": "note",
|
|
"summary": "test",
|
|
})
|
|
assert body == {
|
|
"workplan_id": "wp-1",
|
|
"event_type": "note",
|
|
"summary": "test",
|
|
}
|
|
|
|
legacy_only = server._normalize_progress_body({
|
|
"workstream_id": "wp-legacy",
|
|
"event_type": "note",
|
|
"summary": "legacy",
|
|
})
|
|
assert legacy_only["workplan_id"] == "wp-legacy"
|
|
assert "workstream_id" not in legacy_only
|
|
|
|
|
|
def test_legacy_mcp_deprecation_payload_names_replacement() -> None:
|
|
payload = server._legacy_mcp_deprecation(tool="create_workstream")
|
|
assert payload["replacement"] == "create_workplan"
|
|
assert payload["deprecated"] == "true"
|
|
|
|
|
|
def test_attach_legacy_deprecation_preserves_success_payload() -> None:
|
|
raw = server._attach_legacy_deprecation(
|
|
"update_workstream_status",
|
|
json.dumps({"id": "wp-1", "status": "active"}),
|
|
)
|
|
body = json.loads(raw)
|
|
assert body["id"] == "wp-1"
|
|
assert body["_deprecation"]["tool"] == "update_workstream_status"
|
|
|
|
|
|
def _fake_get(path: str, params: dict | None = None):
|
|
if path == "/topics":
|
|
return [{"id": "topic-1", "domain_slug": "infotech", "title": "Infotech"}]
|
|
if path == "/state/summary":
|
|
return {
|
|
"open_workplans": [
|
|
{
|
|
"id": "wp-1",
|
|
"topic_id": "topic-1",
|
|
"title": "Goalless",
|
|
"repo_id": "repo-1",
|
|
}
|
|
]
|
|
}
|
|
if path == "/decisions":
|
|
return []
|
|
if path == "/progress":
|
|
return []
|
|
if path == "/repos":
|
|
return [{"id": "repo-1", "slug": "demo-repo", "domain_slug": "infotech"}]
|
|
if path == "/repo-goals":
|
|
return [{"id": "goal-1", "title": "Ship it", "description": "d", "priority": "high"}]
|
|
if path == "/capability-catalog/":
|
|
return []
|
|
return []
|
|
|
|
|
|
def test_legacy_workstreams_resource_includes_deprecation(monkeypatch) -> None:
|
|
def fake_get(path: str, params: dict | None = None):
|
|
if path == "/topics":
|
|
return [{"id": "topic-1", "slug": "infotech", "domain_slug": "infotech"}]
|
|
if path == "/workstreams":
|
|
return [{"id": "wp-1", "title": "Demo", "slug": "demo-wp"}]
|
|
return []
|
|
|
|
monkeypatch.setattr(server, "_get", fake_get)
|
|
payload = json.loads(server.resource_workstreams("infotech"))
|
|
assert payload["_deprecation"]["replacement"] == "state://workplans/{topic_slug}"
|
|
assert payload["workplans"] == payload["workstreams"]
|
|
|
|
|
|
def test_get_domain_summary_goal_guidance_is_workplan_first(monkeypatch) -> None:
|
|
monkeypatch.setattr(server, "_get", lambda path, params=None: _fake_get(path, params))
|
|
payload = json.loads(server.get_domain_summary("infotech"))
|
|
assert "workplans" in payload
|
|
assert payload["workstreams"] == payload["workplans"]
|
|
action = payload["goal_guidance"]["needs_workplan"][0]["action"]
|
|
assert "workplan" in action.lower()
|
|
assert "workstream is linked" not in action.lower() |