feat(processes): MCP create/list/get/update_process tools

Task 2 of #582. New mcp/tools/processes.py mirrors entities.py — tools wrap
notes_svc directly. get_process is the fire mechanism (returns the full prompt
via resolve_process; surfaces other_matches on an ambiguous name). Registered
in register_all.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-02 22:27:41 -04:00
parent 1babe59843
commit 7b5a75989a
3 changed files with 184 additions and 1 deletions
+91
View File
@@ -0,0 +1,91 @@
"""Tests for MCP process tools — patches the service layer."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fabledassistant.mcp._context import _user_id_ctx
@pytest.fixture(autouse=True)
def _bind_user():
token = _user_id_ctx.set(7)
yield
_user_id_ctx.reset(token)
def _fake_note(id=1, title="Drift Audit", note_type="process"):
n = MagicMock()
n.id = id
n.title = title
n.note_type = note_type
n.to_dict.return_value = {"id": id, "title": title, "note_type": note_type}
return n
@pytest.mark.asyncio
async def test_create_process_requires_title_and_body():
from fabledassistant.mcp.tools.processes import create_process
with pytest.raises(ValueError):
await create_process(title="", body="something")
with pytest.raises(ValueError):
await create_process(title="X", body=" ")
@pytest.mark.asyncio
async def test_create_process_sets_note_type():
created = _fake_note()
with patch("fabledassistant.services.notes.create_note",
AsyncMock(return_value=created)) as mock_create:
from fabledassistant.mcp.tools.processes import create_process
out = await create_process(title="Drift Audit", body="the prompt", tags=["audit"])
assert out["note_type"] == "process"
# the service was asked to create a process
assert mock_create.await_args.kwargs["note_type"] == "process"
assert mock_create.await_args.kwargs["title"] == "Drift Audit"
@pytest.mark.asyncio
async def test_get_process_returns_body_and_candidates():
note = _fake_note(id=7)
with patch("fabledassistant.services.notes.resolve_process",
AsyncMock(return_value=(note, [{"id": 9, "title": "Drift Audit Notes"}]))):
from fabledassistant.mcp.tools.processes import get_process
out = await get_process("drift")
assert out["id"] == 7
assert out["other_matches"] == [{"id": 9, "title": "Drift Audit Notes"}]
@pytest.mark.asyncio
async def test_get_process_not_found_raises():
with patch("fabledassistant.services.notes.resolve_process",
AsyncMock(return_value=(None, []))):
from fabledassistant.mcp.tools.processes import get_process
with pytest.raises(ValueError):
await get_process("missing")
@pytest.mark.asyncio
async def test_update_process_rejects_non_process_note():
plain = _fake_note(id=3, note_type="note")
with patch("fabledassistant.services.notes.get_note",
AsyncMock(return_value=plain)):
from fabledassistant.mcp.tools.processes import update_process
with pytest.raises(ValueError):
await update_process(process_id=3, title="x")
def test_register_attaches_four_tools():
from fabledassistant.mcp.tools import processes
names: list[str] = []
class FakeMcp:
def tool(self, name):
names.append(name)
def deco(fn):
return fn
return deco
processes.register(FakeMcp())
assert set(names) == {
"list_processes", "create_process", "get_process", "update_process",
}