930 lines
36 KiB
Python
930 lines
36 KiB
Python
#!/usr/bin/env python3
|
|
"""Smoke-test active plan, kanban, and plan lifecycle MCP helpers."""
|
|
|
|
import importlib.util
|
|
import pathlib
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
|
|
def load_module():
|
|
script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py"
|
|
sys.path.insert(0, str(script_path.parent))
|
|
spec = importlib.util.spec_from_file_location("napseer_mcp_server_ergonomics_test", script_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(module)
|
|
module.AUTH = {"account_id": "acct-1", "token": "token", "token_expires_at": "later"}
|
|
module.DEFAULT_PROJECT_ID = "project-1"
|
|
module.project_memory_encryption_active = lambda project_id: False
|
|
return module
|
|
|
|
|
|
@pytest.fixture
|
|
def mod():
|
|
return load_module()
|
|
|
|
|
|
def node(path, status="planned", metadata=None, tags=None, content="body text", archived_at=None, node_type="plan", links=None):
|
|
folder, name = path.rsplit("/", 1)
|
|
merged_metadata = {"status": status}
|
|
if metadata:
|
|
merged_metadata.update(metadata)
|
|
return {
|
|
"id": f"id-{name}",
|
|
"project_id": "project-1",
|
|
"full_path": path,
|
|
"folder_path": folder,
|
|
"name": name,
|
|
"type": node_type,
|
|
"tags": tags or [],
|
|
"links": links or [],
|
|
"metadata": merged_metadata,
|
|
"content_text": content,
|
|
"updated_at": "2026-05-06T12:00:00Z",
|
|
"archived_at": archived_at,
|
|
"encryption_state": "plaintext",
|
|
}
|
|
|
|
|
|
def test_active_plan_listing(mod):
|
|
active = node("/plans/active", "in_progress", tags=["plan"])
|
|
completed = node("/plans/completed", "completed", tags=["plan", "completed"])
|
|
archived = node("/plans/archived", "planned", tags=["plan"], archived_at="2026-05-06T12:01:00Z")
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert "/nodes?" in path
|
|
assert "folder_path=%2Fplans" in path
|
|
return {"items": [active, completed, archived], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.list_active_plans({})
|
|
assert result["ok"] is True
|
|
assert result["view"] == "summary"
|
|
assert [item["full_path"] for item in result["items"]] == ["/plans/active"]
|
|
assert "content_text" not in result["items"][0]
|
|
assert result["items"][0]["status"] == "in_progress"
|
|
|
|
|
|
def test_node_discovery_allows_large_requested_limit(mod):
|
|
calls = []
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
calls.append(path)
|
|
assert method == "GET"
|
|
assert "limit=250" in path
|
|
return {"items": [node(f"/plans/item-{index}") for index in range(250)], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.list_project_nodes({"folder_path": "/plans", "limit": 250})
|
|
assert result["ok"] is True
|
|
assert len(result["items"]) == 250
|
|
assert result["budget"]["limit"] == 250
|
|
assert result["budget"]["max_limit"] > 200
|
|
assert "content_text" not in result["items"][0]
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_node_by_path_returns_canonical_identity_uri(mod):
|
|
mod.AUTH["project_slug"] = "napseer"
|
|
found = node("/documentation/product/prd", "active", node_type="product_spec")
|
|
found["id"] = "node-123"
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert "/nodes/by-path?" in path
|
|
assert "path=%2Fdocumentation%2Fproduct%2Fprd" in path
|
|
assert "view=render" not in path
|
|
return {
|
|
"identity": {
|
|
"node_id": "node-123",
|
|
"canonical_uri": "nap://napseer/project-1/node-123",
|
|
"project_id": "project-1",
|
|
"project_name": "napseer",
|
|
"lookup_kind": "path",
|
|
"lookup_value": "/documentation/product/prd",
|
|
"legacy_full_path": "/documentation/product/prd",
|
|
},
|
|
"route": {
|
|
"id": "node-123",
|
|
"project_id": "project-1",
|
|
"full_path": "/documentation/product/prd",
|
|
"type": "product_spec",
|
|
"name": "prd",
|
|
"folder_path": "/documentation/product",
|
|
},
|
|
"warnings": ["Path lookup is index/route resolution only."],
|
|
}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.node_by_path({"path": "documentation/product/prd"})
|
|
assert result["ok"] is True
|
|
assert result["status"] == "resolved"
|
|
assert result["identity"] == {
|
|
"node_id": "node-123",
|
|
"canonical_uri": "nap://napseer/project-1/node-123",
|
|
"project_id": "project-1",
|
|
"project_name": "napseer",
|
|
"lookup_kind": "path",
|
|
"lookup_value": "/documentation/product/prd",
|
|
"legacy_full_path": "/documentation/product/prd",
|
|
}
|
|
assert result["route"]["full_path"] == "/documentation/product/prd"
|
|
assert "content_text" not in result["route"]
|
|
assert "preview" not in result["route"]
|
|
assert "read_fingerprint" not in result["route"]
|
|
assert "Path lookup is index/route resolution only" in result["warnings"][0]
|
|
|
|
|
|
def test_path_resolver_then_read_uses_node_id_route(mod):
|
|
calls = []
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
calls.append(path)
|
|
assert method == "GET"
|
|
if "/nodes/by-path?" in path:
|
|
assert "view=" not in path
|
|
return {
|
|
"identity": {"node_id": "node-123", "project_id": "project-1"},
|
|
"route": {"id": "node-123", "project_id": "project-1", "full_path": "/documentation/product/prd"},
|
|
}
|
|
assert path == "/v1/projects/project-1/nodes/node-123"
|
|
return {**node("/documentation/product/prd", "active", node_type="product_spec"), "id": "node-123"}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.read_node_by_path({"path": "/documentation/product/prd"})
|
|
assert result["id"] == "node-123"
|
|
assert calls == [
|
|
"/v1/projects/project-1/nodes/by-path?path=%2Fdocumentation%2Fproduct%2Fprd",
|
|
"/v1/projects/project-1/nodes/node-123",
|
|
]
|
|
|
|
|
|
def test_node_by_id_returns_canonical_identity_uri(mod):
|
|
mod.AUTH["project_slug"] = "napseer"
|
|
found = node("/documentation/product/prd", "active", node_type="product_spec")
|
|
found["id"] = "node-123"
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert path == "/v1/projects/project-1/nodes/node-123"
|
|
return found
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.node_by_id({"node_id": "node-123"})
|
|
assert result["ok"] is True
|
|
assert result["identity"]["lookup_kind"] == "node_id"
|
|
assert result["identity"]["canonical_uri"] == "nap://napseer/project-1/node-123"
|
|
assert result["node"]["id"] == "node-123"
|
|
|
|
|
|
def test_node_by_uri_resolves_canonical_uri(mod):
|
|
mod.AUTH["project_slug"] = "napseer"
|
|
found = node("/documentation/product/prd", "active", node_type="product_spec")
|
|
found["id"] = "node-123"
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert path == "/v1/projects/project-1/nodes/node-123"
|
|
return found
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.node_by_uri({"uri": "nap://napseer/project-1/node-123"})
|
|
assert result["ok"] is True
|
|
assert result["identity"]["lookup_kind"] == "uri"
|
|
assert result["identity"]["lookup_value"] == "nap://napseer/project-1/node-123"
|
|
assert result["identity"]["canonical_uri"] == "nap://napseer/project-1/node-123"
|
|
|
|
|
|
def test_node_get_accepts_node_id_without_path(mod):
|
|
found = node("/documentation/product/prd", "active", node_type="product_spec")
|
|
found["id"] = "node-123"
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert path == "/v1/projects/project-1/nodes/node-123"
|
|
return found
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.node_get({"node_id": "node-123", "view": "render"})
|
|
assert result["id"] == "node-123"
|
|
assert result["full_path"] == "/documentation/product/prd"
|
|
|
|
|
|
def test_node_get_edit_view_uses_id_route_not_by_path(mod):
|
|
edit_payload = {
|
|
"node": node("/documentation/product/prd", "active", node_type="product_spec"),
|
|
"revision": "2026-05-06T12:00:00Z",
|
|
"read_fingerprint": "fingerprint-1",
|
|
"content_lines": ["body text"],
|
|
"line_count": 1,
|
|
"ends_with_newline": False,
|
|
}
|
|
edit_payload["node"]["id"] = "node-123"
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert path == "/v1/projects/project-1/nodes/node-123?view=edit"
|
|
return edit_payload
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.node_get({"node_id": "node-123", "view": "edit"})
|
|
assert result["revision"] == "2026-05-06T12:00:00Z"
|
|
assert result["node"]["id"] == "node-123"
|
|
|
|
|
|
def test_node_get_rejects_path_reads(mod):
|
|
try:
|
|
mod.node_get({"path": "/documentation/product/prd", "view": "render"})
|
|
except RuntimeError as exc:
|
|
assert "path is an index/route only" in str(exc)
|
|
else:
|
|
raise AssertionError("path reads should be rejected")
|
|
|
|
|
|
def test_tools_include_node_by_path_descriptor(mod):
|
|
get_tool = next(item for item in mod.tools() if item["name"] == "nap_node_get")
|
|
assert get_tool["inputSchema"].get("required") is None
|
|
assert "node_id" in get_tool["inputSchema"]["properties"]
|
|
assert "uri" in get_tool["inputSchema"]["properties"]
|
|
assert "path" not in get_tool["inputSchema"]["properties"]
|
|
|
|
by_id = next(item for item in mod.tools() if item["name"] == "nap_node_by_id")
|
|
assert by_id["inputSchema"]["required"] == ["node_id"]
|
|
assert "nap:// URI" in by_id["description"]
|
|
|
|
by_uri = next(item for item in mod.tools() if item["name"] == "nap_node_by_uri")
|
|
assert by_uri["inputSchema"]["required"] == ["uri"]
|
|
assert "nap://" in by_uri["description"]
|
|
|
|
tool = next(item for item in mod.tools() if item["name"] == "nap_node_by_path")
|
|
assert tool["inputSchema"]["required"] == ["path"]
|
|
assert "preview_chars" not in tool["inputSchema"]["properties"]
|
|
assert "does not read full node content" in tool["description"]
|
|
|
|
assert all(item["name"] != "nap_cat" for item in mod.tools())
|
|
|
|
for name in ["nap_discover", "nap_find", "nap_context", "nap_find_related", "nap_recent"]:
|
|
schema = next(item for item in mod.tools() if item["name"] == name)["inputSchema"]
|
|
view = schema["properties"].get("view")
|
|
if view:
|
|
assert "full" not in view["enum"]
|
|
|
|
for name in ["nap_patch", "nap_node_patch", "nap_rm"]:
|
|
schema = next(item for item in mod.tools() if item["name"] == name)["inputSchema"]
|
|
assert "node_id" in schema["properties"]
|
|
assert "uri" in schema["properties"]
|
|
assert "path" not in schema["properties"]
|
|
|
|
|
|
def test_mutation_tools_reject_path_and_patch_by_node_id(mod):
|
|
found = node("/notes/current", "active", node_type="note")
|
|
found["id"] = "node-123"
|
|
writes = []
|
|
|
|
mod.get_node_by_id = lambda args: found
|
|
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload, args)) or {
|
|
**found,
|
|
"content_text": payload.get("content_text", found["content_text"]) if payload else found["content_text"],
|
|
}
|
|
mod.index_node = lambda node: None
|
|
mod.remove_indexed_node = lambda node_id: None
|
|
|
|
updated = mod.update_node_by_path({"node_id": "node-123", "content_text": "updated"})
|
|
assert updated["updated"] is True
|
|
assert writes[-1][0] == "PATCH"
|
|
assert writes[-1][1] == "/v1/projects/project-1/nodes/node-123"
|
|
assert writes[-1][3][2:] == ("node", "node-123")
|
|
|
|
patched = mod.guarded_node_patch({
|
|
"node_id": "node-123",
|
|
"precondition": {"revision": "r1", "read_fingerprint": "fp1"},
|
|
"set": {"metadata": {"status": "active"}},
|
|
})
|
|
assert patched["id"] == "node-123"
|
|
assert writes[-1][1] == "/v1/projects/project-1/nodes/node-123"
|
|
|
|
archived = mod.archive_node_by_path({"node_id": "node-123"})
|
|
assert archived["archived"] is True
|
|
assert writes[-1][0] == "DELETE"
|
|
assert writes[-1][1] == "/v1/projects/project-1/nodes/node-123"
|
|
|
|
with pytest.raises(RuntimeError, match="path is an index/route only"):
|
|
mod.update_node_by_path({"path": "/notes/current", "content_text": "bad"})
|
|
|
|
|
|
def test_discovery_rejects_full_content_view(mod):
|
|
with pytest.raises(RuntimeError, match="full node bodies require nap_node_get"):
|
|
mod.list_project_nodes({"view": "full"})
|
|
|
|
|
|
def test_include_content_is_legacy_noop(mod):
|
|
found = node("/plans/found", "planned", content="body must not be returned")
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert "view=paths" in path
|
|
return {"items": [found], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.list_project_nodes({"include_content": True, "view": "paths"})
|
|
assert result["view"] == "paths"
|
|
assert "content_text" not in result["items"][0]
|
|
|
|
|
|
def test_recent_memory_defaults_to_paths(mod):
|
|
recent = node(
|
|
"/notes/recent-title",
|
|
"active",
|
|
metadata={"title": "Readable Recent Title", "priority": "high"},
|
|
tags=["recent"],
|
|
content="large body that should not appear in the default recent response",
|
|
node_type="note",
|
|
)
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert "/nodes?" in path
|
|
assert "view=paths" in path
|
|
return {"items": [recent], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.recent_memory({})
|
|
assert result["ok"] is True
|
|
assert result["view"] == "paths"
|
|
assert result["items"] == [{
|
|
"full_path": "/notes/recent-title",
|
|
"type": "note",
|
|
"status": "active",
|
|
"updated_at": "2026-05-06T12:00:00Z",
|
|
}]
|
|
assert "content_text" not in result["items"][0]
|
|
assert "metadata_summary" not in result["items"][0]
|
|
|
|
|
|
def test_recent_memory_accepts_date_aliases_and_returns_dates(mod):
|
|
recent = node(
|
|
"/notes/recent-title",
|
|
"active",
|
|
metadata={"date": "2026-05-06"},
|
|
tags=["recent"],
|
|
node_type="note",
|
|
)
|
|
recent["created_at"] = "2026-05-05T10:00:00Z"
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert "updated_after=2026-05-05T00%3A00%3A00Z" in path
|
|
assert "updated_before=2026-05-06T23%3A59%3A59Z" in path
|
|
return {"items": [recent], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.recent_memory({"since": "2026-05-05", "until": "2026-05-06"})
|
|
assert result["items"] == [{
|
|
"full_path": "/notes/recent-title",
|
|
"type": "note",
|
|
"status": "active",
|
|
"date": "2026-05-06",
|
|
"created_at": "2026-05-05T10:00:00Z",
|
|
"updated_at": "2026-05-06T12:00:00Z",
|
|
}]
|
|
|
|
|
|
def test_find_defaults_to_paths_and_requests_compact_rest_view(mod):
|
|
found = node(
|
|
"/plans/found",
|
|
"planned",
|
|
content="large body that should not be fetched by default",
|
|
)
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert "/nodes?" in path
|
|
assert "folder_path=%2Fplans" in path
|
|
assert "view=paths" in path
|
|
return {"items": [found], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.list_project_nodes({"folder_path": "/plans"})
|
|
assert result["ok"] is True
|
|
assert result["view"] == "paths"
|
|
assert result["items"] == [{
|
|
"full_path": "/plans/found",
|
|
"type": "plan",
|
|
"status": "planned",
|
|
"updated_at": "2026-05-06T12:00:00Z",
|
|
}]
|
|
assert "content_text" not in result["items"][0]
|
|
|
|
|
|
def test_discover_uses_search_when_keywords_present(mod):
|
|
calls = []
|
|
|
|
def fake_search(args):
|
|
calls.append(args)
|
|
return {"ok": True, "items": [], "view": args.get("view")}
|
|
|
|
mod.search_memory = fake_search
|
|
result = mod.discover_memory({"keywords": "gateway relay", "since": "2026-05-01"})
|
|
assert result["tool_intent"] == "search"
|
|
assert result["range"]["updated_after"] == "2026-05-01T00:00:00Z"
|
|
assert calls and calls[-1]["q"] == "gateway relay"
|
|
assert calls[-1]["view"] == "summary"
|
|
assert "date" in result["awareness"]["date_fields_in_rows"]
|
|
|
|
|
|
def test_classify_query_points_exact_paths_to_context(mod):
|
|
result = mod.classify_memory_query({"q": "explain /documentation/security/overview since May", "since": "2026-05-01"})
|
|
assert result["classification"]["primary_tool"] == "nap_context"
|
|
assert result["classification"]["exact_paths"] == ["/documentation/security/overview"]
|
|
assert result["classification"]["range"]["updated_after"] == "2026-05-01T00:00:00Z"
|
|
|
|
|
|
def test_title_view_requests_summary_rest_view_without_content(mod):
|
|
found = node(
|
|
"/notes/titled",
|
|
"active",
|
|
metadata={"title": "Readable Title"},
|
|
content="large body that should not be fetched by title view",
|
|
)
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert "/nodes?" in path
|
|
assert "view=summary" in path
|
|
return {"items": [found], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.list_project_nodes({"view": "titles"})
|
|
assert result["ok"] is True
|
|
assert result["view"] == "titles"
|
|
assert result["items"] == [{"title": "Readable Title", "full_path": "/notes/titled"}]
|
|
assert "content_text" not in result["items"][0]
|
|
|
|
|
|
def test_rest_query_string_serializes_booleans_lowercase(mod):
|
|
query = mod.rest_query_string({
|
|
"folder_path": "/kanban",
|
|
"include_archived": False,
|
|
"archived_only": False,
|
|
"active_only": True,
|
|
"status": ["todo", "doing"],
|
|
})
|
|
|
|
assert "include_archived=false" in query
|
|
assert "archived_only=false" in query
|
|
assert "active_only=true" in query
|
|
assert "include_archived=False" not in query
|
|
assert "active_only=True" not in query
|
|
assert "status=todo" in query
|
|
assert "status=doing" in query
|
|
|
|
|
|
def test_kanban_list_grouping_and_filters(mod):
|
|
doing = node(
|
|
"/kanban/doing/fix-wrapper",
|
|
"doing",
|
|
metadata={"column": "doing", "priority": "high", "owner": "codex", "blocked": True},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
todo = node(
|
|
"/kanban/todo/add-tests",
|
|
"todo",
|
|
metadata={"column": "todo", "priority": "normal", "owner": "codex", "blocked": False},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
done = node(
|
|
"/kanban/done/old-card",
|
|
"done",
|
|
metadata={"column": "done", "owner": "codex"},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
outside = node("/plans/not-kanban", "planned")
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
assert "q=" in path
|
|
assert "include_archived=false" in path
|
|
assert "archived_only=false" in path
|
|
return {"items": [doing, todo, done, outside], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.list_kanban_cards({"owner": "codex", "blocked": True, "group_by": "column"})
|
|
assert result["ok"] is True
|
|
assert result["kanban_root"] == "/kanban"
|
|
assert [item["full_path"] for item in result["items"]] == ["/kanban/doing/fix-wrapper"]
|
|
assert "content_text" not in result["items"][0]
|
|
assert result["items"][0]["column"] == "doing"
|
|
assert result["items"][0]["blocked"] is True
|
|
assert list(result["groups"]) == ["doing"]
|
|
|
|
|
|
def test_kanban_dependency_summary_filters_ready_and_blocking(mod):
|
|
plan = node(
|
|
"/kanban/backlog/plan",
|
|
"backlog",
|
|
metadata={"column": "backlog", "priority": "urgent", "blocked": False},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
links=[{"path": "/kanban/backlog/implementation", "relation": "blocks"}],
|
|
)
|
|
implementation = node(
|
|
"/kanban/todo/implementation",
|
|
"todo",
|
|
metadata={
|
|
"column": "todo",
|
|
"priority": "high",
|
|
"blocked_by": "/kanban/backlog/plan",
|
|
"external_blockers": [{"id": "deploy", "state": "open", "strength": "hard"}],
|
|
},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
links=[
|
|
{"path": "/kanban/backlog/plan", "relation": "blocked-by"},
|
|
{"path": "/kanban/backlog/context", "relation": "relates-to"},
|
|
{"path": "/kanban/backlog/child", "relation": "parent_of"},
|
|
{"path": "/kanban/backlog/canonical", "relation": "duplicate-of"},
|
|
],
|
|
)
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
return {"items": [plan, implementation], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
|
|
blocked = mod.list_kanban_cards({"dependency_view": "blocked", "limit": 10})
|
|
assert [item["full_path"] for item in blocked["items"]] == ["/kanban/todo/implementation"]
|
|
assert blocked["items"][0]["dependency_summary"]["hard_blockers"] == 1
|
|
assert blocked["items"][0]["dependency_summary"]["external_blockers"] == 1
|
|
assert blocked["items"][0]["hard_blockers"] == ["/kanban/backlog/plan"]
|
|
assert blocked["items"][0]["related"] == ["/kanban/backlog/context"]
|
|
assert blocked["items"][0]["children"] == ["/kanban/backlog/child"]
|
|
assert blocked["items"][0]["duplicates"] == ["/kanban/backlog/canonical"]
|
|
|
|
ready = mod.list_kanban_cards({"dependency_view": "ready", "limit": 10})
|
|
assert [item["full_path"] for item in ready["items"]] == ["/kanban/backlog/plan"]
|
|
|
|
blocking = mod.list_kanban_cards({"dependency_view": "blocking", "limit": 10})
|
|
assert blocking["items"][0]["blocking"] == ["/kanban/backlog/implementation"]
|
|
|
|
|
|
def test_kanban_list_reads_beyond_first_source_page(mod):
|
|
first_page = [
|
|
node(
|
|
"/kanban/todo/other-card",
|
|
"todo",
|
|
metadata={"column": "todo", "owner": "codex", "blocked": False},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
]
|
|
second_page = [
|
|
node(
|
|
"/kanban/doing/blocked-card",
|
|
"doing",
|
|
metadata={"column": "doing", "owner": "codex", "blocked": True},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
]
|
|
calls = []
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
calls.append(path)
|
|
assert method == "GET"
|
|
assert "q=%2Fkanban" in path
|
|
assert "limit=10000" in path
|
|
assert "include_archived=false" in path
|
|
assert "archived_only=false" in path
|
|
if "cursor=page-2" in path:
|
|
return {"items": second_page, "next_cursor": None}
|
|
return {"items": first_page, "next_cursor": "page-2"}
|
|
|
|
mod.request_json = fake_request
|
|
result = mod.list_kanban_cards({"blocked": True, "limit": 10})
|
|
assert [item["full_path"] for item in result["items"]] == ["/kanban/doing/blocked-card"]
|
|
assert result["truncated"] is False
|
|
assert len(calls) == 2
|
|
|
|
|
|
def test_kanban_workflow_reads_pending_and_pick_next(mod):
|
|
backlog_urgent = node(
|
|
"/kanban/backlog/later-urgent",
|
|
"backlog",
|
|
metadata={"column": "backlog", "priority": "urgent", "rank": "00001024", "blocked": False},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
todo_high = node(
|
|
"/kanban/todo/next-high",
|
|
"todo",
|
|
metadata={"column": "todo", "priority": "high", "rank": "00002048", "blocked": False},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
todo_blocked = node(
|
|
"/kanban/todo/blocked",
|
|
"todo",
|
|
metadata={"column": "todo", "priority": "urgent", "rank": "00001024", "blocked": True},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
|
|
def fake_request(method, path, payload=None, **kwargs):
|
|
assert method == "GET"
|
|
return {"items": [backlog_urgent, todo_high, todo_blocked], "next_cursor": None}
|
|
|
|
mod.request_json = fake_request
|
|
pending = mod.get_pending_kanban_cards({})
|
|
assert [item["full_path"] for item in pending["items"]] == ["/kanban/todo/next-high", "/kanban/backlog/later-urgent"]
|
|
|
|
picked = mod.pick_next_kanban_card({})
|
|
assert picked["selected"]["full_path"] == "/kanban/todo/next-high"
|
|
assert picked["selection"]["blocked_excluded"] is True
|
|
|
|
|
|
def test_kanban_workflow_create_move_and_block(mod):
|
|
writes = []
|
|
existing = node(
|
|
"/kanban/todo/add-mcp-commands",
|
|
"todo",
|
|
metadata={"column": "todo", "priority": "normal", "rank": "00001024", "blocked": False, "title": "Add MCP commands"},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
|
|
mod.list_kanban_cards = lambda args: {"items": []}
|
|
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload)) or {
|
|
**existing,
|
|
"id": "id-add-mcp-commands",
|
|
"full_path": f"{payload.get('folder_path', '/kanban/todo')}/{payload.get('name', 'add-mcp-commands')}".replace("//", "/"),
|
|
"folder_path": payload.get("folder_path", "/kanban/todo"),
|
|
"name": payload.get("name", "add-mcp-commands"),
|
|
"metadata": payload.get("metadata", existing["metadata"]),
|
|
"content_text": payload.get("content_text", existing["content_text"]),
|
|
"tags": payload.get("tags", existing["tags"]),
|
|
}
|
|
mod.get_node_by_path = lambda args, allow_agent=False: existing
|
|
mod.index_node = lambda node: None
|
|
mod.remove_indexed_node = lambda node_id: None
|
|
|
|
created = mod.create_kanban_card({"title": "Add MCP commands", "priority": "HIGH"})
|
|
assert created["created"] is True
|
|
assert "node" not in created
|
|
assert writes[0][0] == "POST"
|
|
assert writes[0][2]["folder_path"] == "/kanban/todo"
|
|
assert writes[0][2]["metadata"]["priority"] == "high"
|
|
assert writes[0][2]["metadata"]["rank"] == "00001024"
|
|
|
|
moved = mod.move_kanban_card({"path": "/kanban/todo/add-mcp-commands"}, "doing")
|
|
assert moved["to"] == "/kanban/doing/add-mcp-commands"
|
|
assert "node" not in moved
|
|
assert writes[1][0] == "PATCH"
|
|
assert writes[1][2]["folder_path"] == "/kanban/doing"
|
|
assert writes[1][2]["metadata"]["column"] == "doing"
|
|
assert writes[1][2]["metadata"]["lifecycle_state"] == "active"
|
|
|
|
blocked = mod.block_kanban_card({"path": "/kanban/todo/add-mcp-commands", "blocked_reason": "waiting"}, True)
|
|
assert blocked["updated"] is True
|
|
assert "node" not in blocked
|
|
assert writes[2][2]["metadata"]["blocked"] is True
|
|
assert writes[2][2]["metadata"]["blocked_by"] == "waiting"
|
|
|
|
|
|
def test_kanban_create_deduplicates_overlapping_titles(mod):
|
|
writes = []
|
|
existing_cards = [
|
|
node("/kanban/todo/add-mcp-commands", "todo", metadata={"column": "todo"}, tags=["kanban"], node_type="kanban_card"),
|
|
node("/kanban/todo/add-mcp-commands-20260510123456", "todo", metadata={"column": "todo"}, tags=["kanban"], node_type="kanban_card"),
|
|
node("/kanban/doing/add-mcp-commands", "doing", metadata={"column": "doing"}, tags=["kanban"], node_type="kanban_card"),
|
|
]
|
|
|
|
mod.list_kanban_cards = lambda args: {"items": existing_cards if args.get("column") == "todo" else []}
|
|
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload)) or {
|
|
**existing_cards[0],
|
|
"id": "id-deduped",
|
|
"full_path": f"{payload.get('folder_path')}/{payload.get('name')}",
|
|
"folder_path": payload.get("folder_path"),
|
|
"name": payload.get("name"),
|
|
"metadata": payload.get("metadata", {}),
|
|
"tags": payload.get("tags", []),
|
|
}
|
|
mod.index_node = lambda node: None
|
|
|
|
created = mod.create_kanban_card({"title": "Add MCP commands", "column": "todo", "now": "2026-05-10T12:34:56Z"})
|
|
assert created["path"] == "/kanban/todo/add-mcp-commands-20260510123456-2"
|
|
assert "node" not in created
|
|
assert writes[0][2]["name"] == "add-mcp-commands-20260510123456-2"
|
|
|
|
|
|
def test_lineage_status_reports_generic_plan_and_card_coverage(mod):
|
|
linked_plan = node("/plans/linked", "planned", links=[{"path": "/decisions/source", "relation": "implements"}])
|
|
unlinked_plan = node("/plans/unlinked", "planned")
|
|
linked_card = node(
|
|
"/kanban/todo/linked",
|
|
"todo",
|
|
metadata={"column": "todo"},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
links=[{"path": "/plans/linked", "relation": "derived-from"}],
|
|
)
|
|
unlinked_card = node(
|
|
"/kanban/todo/unlinked",
|
|
"todo",
|
|
metadata={"column": "todo"},
|
|
tags=["kanban"],
|
|
node_type="kanban_card",
|
|
)
|
|
|
|
def fake_list(args):
|
|
if args.get("folder_path") == "/plans":
|
|
return {"items": [linked_plan, unlinked_plan]}
|
|
if args.get("q") == "/kanban":
|
|
return {"items": [linked_card, unlinked_card, node("/notes/not-kanban", "active")]}
|
|
return {"items": []}
|
|
|
|
mod.list_project_nodes = fake_list
|
|
result = mod.lineage_status({})
|
|
assert result["status"] == "warnings"
|
|
assert result["counts"]["plans"] == 2
|
|
assert result["counts"]["plans_with_source_record"] == 1
|
|
assert result["counts"]["kanban_cards"] == 2
|
|
assert result["counts"]["kanban_cards_with_origin_plan"] == 1
|
|
assert [warning["type"] for warning in result["warnings"]] == [
|
|
"plan_missing_source_record",
|
|
"kanban_card_missing_origin_plan",
|
|
]
|
|
|
|
|
|
def test_plan_to_kanban_creates_card_with_origin_plan_link(mod):
|
|
writes = []
|
|
plan = node(
|
|
"/plans/generic-work",
|
|
"planned",
|
|
metadata={"title": "Generic Work", "priority": "high", "owner": "codex"},
|
|
tags=["plan"],
|
|
content="Implement a generic record to plan to kanban flow.",
|
|
)
|
|
|
|
mod.get_node_by_path = lambda args, allow_agent=False: plan
|
|
mod.list_kanban_cards = lambda args: {"items": []}
|
|
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload)) or {
|
|
"id": "id-generic-work",
|
|
"project_id": "project-1",
|
|
"full_path": f"{payload.get('folder_path')}/{payload.get('name')}",
|
|
"folder_path": payload.get("folder_path"),
|
|
"name": payload.get("name"),
|
|
"type": payload.get("type"),
|
|
"metadata": payload.get("metadata", {}),
|
|
"tags": payload.get("tags", []),
|
|
"links": payload.get("links", []),
|
|
"content_text": payload.get("content_text", ""),
|
|
"updated_at": "2026-05-06T12:00:00Z",
|
|
}
|
|
mod.index_node = lambda node: None
|
|
|
|
created = mod.plan_to_kanban_card({"plan_path": "/plans/generic-work", "column": "todo"})
|
|
assert created["path"] == "/kanban/todo/generic-work"
|
|
assert created["source_plan_path"] == "/plans/generic-work"
|
|
assert writes[0][0] == "POST"
|
|
assert writes[0][2]["links"] == [{"path": "/plans/generic-work", "relation": "derived-from"}]
|
|
assert writes[0][2]["metadata"]["source_plan_path"] == "/plans/generic-work"
|
|
assert writes[0][2]["metadata"]["priority"] == "high"
|
|
assert writes[0][2]["metadata"]["owner"] == "codex"
|
|
|
|
|
|
def test_plan_lifecycle_dry_run_and_write(mod):
|
|
plan = node("/plans/example", "in_progress", tags=["plan"])
|
|
outcome = node("/implementation-notes/example", "completed", node_type="implementation-note")
|
|
writes = []
|
|
|
|
def fake_get(args, allow_agent=False):
|
|
if args["path"] == "/plans/example":
|
|
return plan
|
|
if args["path"] == "/implementation-notes/example":
|
|
return outcome
|
|
raise RuntimeError("HTTP 404")
|
|
|
|
mod.get_node_by_path = fake_get
|
|
mod.try_get_node_by_path = lambda path, allow_agent=False: outcome if path == "/implementation-notes/example" else None
|
|
mod.update_node_by_path = lambda args: writes.append(("update", args)) or {"node": {**plan, **args}}
|
|
mod.archive_node_by_path = lambda args: writes.append(("archive", args)) or {"archived": True}
|
|
|
|
dry = mod.complete_plan({"path": "/plans/example", "outcome_path": "/implementation-notes/example", "dry_run": True})
|
|
assert dry["changed"] is True
|
|
assert dry["dry_run"] is True
|
|
assert dry["status"] == "completed"
|
|
assert "archive node" in dry["planned_changes"]
|
|
assert writes == []
|
|
|
|
written = mod.complete_plan({"path": "/plans/example", "outcome_path": "/implementation-notes/example", "reason": "implemented"})
|
|
assert written["changed"] is True
|
|
assert written["archived"] is True
|
|
assert writes[0][0] == "update"
|
|
assert writes[0][1]["metadata"]["status"] == "completed"
|
|
assert writes[0][1]["metadata"]["outcome_path"] == "/implementation-notes/example"
|
|
assert "completed" in writes[0][1]["tags"]
|
|
assert "archived" in writes[0][1]["tags"]
|
|
assert writes[0][1]["links"] == [{"path": "/implementation-notes/example", "relation": "implemented-by"}]
|
|
assert writes[1][0] == "archive"
|
|
|
|
|
|
def test_plan_lifecycle_archived_idempotent(mod):
|
|
archived = node(
|
|
"/plans/example",
|
|
"completed",
|
|
metadata={"outcome_path": "/implementation-notes/example"},
|
|
tags=["plan", "completed", "archived"],
|
|
archived_at="2026-05-06T12:01:00Z",
|
|
)
|
|
archived["links"] = [{"path": "/implementation-notes/example", "relation": "implemented-by"}]
|
|
|
|
mod.get_node_by_path = lambda args, allow_agent=False: (_ for _ in ()).throw(RuntimeError("HTTP 404"))
|
|
mod.list_project_nodes = lambda args: {"items": [archived]}
|
|
result = mod.complete_plan({"path": "/plans/example", "outcome_path": "/implementation-notes/example"})
|
|
assert result["ok"] is True
|
|
assert result["changed"] is False
|
|
assert result["archived"] is True
|
|
|
|
|
|
def test_tool_registry(mod):
|
|
names = {tool["name"] for tool in mod.raw_tools()}
|
|
for name in [
|
|
"nap_plan_list_active",
|
|
"nap_lineage_status",
|
|
"nap_kanban_list",
|
|
"nap_kanban_get_pending",
|
|
"nap_kanban_pick_next",
|
|
"nap_kanban_start",
|
|
"nap_kanban_complete",
|
|
"nap_plan_complete",
|
|
"nap_plan_supersede",
|
|
"nap_plan_cancel",
|
|
"nap_plan_to_kanban",
|
|
]:
|
|
assert name in names
|
|
assert mod.tool_contract_metadata("nap_plan_complete")["dry_run_supported"] is True
|
|
assert mod.tool_contract_metadata("nap_plan_list_active")["side_effect"] == "none"
|
|
assert mod.tool_contract_metadata("nap_lineage_status")["side_effect"] == "none"
|
|
assert mod.tool_contract_metadata("nap_kanban_get_pending")["side_effect"] == "none"
|
|
assert mod.tool_contract_metadata("nap_plan_to_kanban")["side_effect"] == "remote-write"
|
|
assert mod.tool_contract_metadata("nap_kanban_start")["side_effect"] == "remote-write"
|
|
|
|
|
|
def run():
|
|
mod = load_module()
|
|
test_active_plan_listing(mod)
|
|
mod = load_module()
|
|
test_node_discovery_allows_large_requested_limit(mod)
|
|
mod = load_module()
|
|
test_node_by_path_returns_canonical_identity_uri(mod)
|
|
mod = load_module()
|
|
test_node_by_id_returns_canonical_identity_uri(mod)
|
|
mod = load_module()
|
|
test_node_by_uri_resolves_canonical_uri(mod)
|
|
mod = load_module()
|
|
test_node_get_accepts_node_id_without_path(mod)
|
|
mod = load_module()
|
|
test_node_get_rejects_path_reads(mod)
|
|
mod = load_module()
|
|
test_tools_include_node_by_path_descriptor(mod)
|
|
mod = load_module()
|
|
test_recent_memory_defaults_to_paths(mod)
|
|
mod = load_module()
|
|
test_find_defaults_to_paths_and_requests_compact_rest_view(mod)
|
|
mod = load_module()
|
|
test_title_view_requests_summary_rest_view_without_content(mod)
|
|
mod = load_module()
|
|
test_kanban_list_grouping_and_filters(mod)
|
|
mod = load_module()
|
|
test_kanban_dependency_summary_filters_ready_and_blocking(mod)
|
|
mod = load_module()
|
|
test_kanban_list_reads_beyond_first_source_page(mod)
|
|
mod = load_module()
|
|
test_kanban_workflow_reads_pending_and_pick_next(mod)
|
|
mod = load_module()
|
|
test_kanban_workflow_create_move_and_block(mod)
|
|
mod = load_module()
|
|
test_kanban_create_deduplicates_overlapping_titles(mod)
|
|
mod = load_module()
|
|
test_lineage_status_reports_generic_plan_and_card_coverage(mod)
|
|
mod = load_module()
|
|
test_plan_to_kanban_creates_card_with_origin_plan_link(mod)
|
|
mod = load_module()
|
|
test_plan_lifecycle_dry_run_and_write(mod)
|
|
mod = load_module()
|
|
test_plan_lifecycle_archived_idempotent(mod)
|
|
mod = load_module()
|
|
test_tool_registry(mod)
|
|
print("ok: MCP memory ergonomics helpers smoke passed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|