nap-cli/tests/test_mcp_memory_ergonomics_helpers.py

#!/usr/bin/env python3
"""Smoke-test active plan, kanban, and plan lifecycle MCP helpers."""

import importlib.util
import pathlib
import sys


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


def node(path, status="planned", metadata=None, tags=None, content="body text", archived_at=None, node_type="plan"):
    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": [],
        "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_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
        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_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_kanban_list", "nap_plan_complete", "nap_plan_supersede", "nap_plan_cancel"]:
        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"


def run():
    mod = load_module()
    test_active_plan_listing(mod)
    mod = load_module()
    test_kanban_list_grouping_and_filters(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()