bvandeusenandClaude Opus 5.5
914f31359c
CI & Build / Python lint (push) Successful in 5s
CI & Build / Plugin hooks (push) Successful in 17s
CI & Build / TypeScript typecheck (push) Successful in 57s
CI & Build / integration (push) Successful in 1m19s
CI & Build / Python tests (push) Successful in 2m2s
CI & Build / Build & push image (push) Successful in 15s
Docstrings, section headers and assertion messages drop the record number, milestone or step that introduced the behaviour, and incident accounts are restated as the present-tense failure shape. A dead pointer to a design doc that is not in the repo is removed. Renamed to the property they check: test_the_old_shape_would_be_caught -> test_a_rank_then_scope_query_is_caught test_a_ledger_repeat_no_longer_produces_that_case_at_all -> test_a_ledger_repeat_is_counted_not_withheld test_the_header_no_longer_promises_once_per_session -> test_the_header_explains_the_seen_marker test_the_portal_case_no_longer_stamps_one_canon_onto_a_whole_file -> test_one_canon_is_not_stamped_onto_a_whole_file test_the_repos_own_prose_no_longer_defines_with -> test_the_repos_own_prose_does_not_define_with test_the_original_regression_create_note_with_content -> test_the_real_server_refuses_content_on_create_note Fixture and example ids stay: "note 5" in a supersession chain, "rule 156" in an illustrative question, `#2275` as the hex-versus-citation case, `#3244` as the example of a bare number, fixed dates in date-filter tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
265 lines
12 KiB
Python
265 lines
12 KiB
Python
"""Tests for MCP process tools — patches the service layer."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from tests.helpers import FakeMCP, fake_note
|
|
|
|
|
|
pytestmark = pytest.mark.usefixtures("_bind_user")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_process_requires_title_and_body():
|
|
from scribe.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(title="Drift Audit", note_type="process")
|
|
with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
|
|
AsyncMock(return_value=None)), \
|
|
patch("scribe.services.notes.create_note",
|
|
AsyncMock(return_value=created)) as mock_create:
|
|
from scribe.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_create_process_blocks_a_near_duplicate():
|
|
"""The gate matters more for processes than for other kinds: each one becomes
|
|
a skill file that auto-surfaces, so two near-identical procedures don't just
|
|
bloat the corpus — they compete to be followed."""
|
|
from scribe.services.dedup import DuplicateMatch
|
|
|
|
# The real dataclass, not a MagicMock: a mock answers every attribute, so it
|
|
# would pass whatever field names this test happened to guess and prove
|
|
# nothing about the payload the tool actually returns.
|
|
match = DuplicateMatch(id=42, title="Drift Audit", similarity=0.94, reason="semantic")
|
|
with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
|
|
AsyncMock(return_value=match)), \
|
|
patch("scribe.services.notes.create_note", AsyncMock()) as mock_create:
|
|
from scribe.mcp.tools.processes import create_process
|
|
out = await create_process(title="Drift Audit", body="the prompt")
|
|
assert out["duplicate"] is True
|
|
assert out["existing_id"] == 42
|
|
mock_create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_process_force_bypasses_the_gate():
|
|
created = fake_note(title="Drift Audit", note_type="process")
|
|
with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
|
|
AsyncMock()) as find_mock, \
|
|
patch("scribe.services.notes.create_note",
|
|
AsyncMock(return_value=created)):
|
|
from scribe.mcp.tools.processes import create_process
|
|
await create_process(title="Drift Audit", body="the prompt", force=True)
|
|
find_mock.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_process_returns_body_and_candidates():
|
|
note = fake_note(id=7, title="Drift Audit", note_type="process")
|
|
with patch("scribe.services.notes.resolve_process",
|
|
AsyncMock(return_value=(note, [{"id": 9, "title": "Drift Audit Notes"}]))):
|
|
from scribe.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"}]
|
|
# The caller's own process carries no provenance marker — absence of the flag
|
|
# is what makes it read as theirs.
|
|
assert "shared" not in out
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_process_flags_another_users_process():
|
|
"""A shared Process must arrive labelled. get_process's contract is 'follow
|
|
the returned body', so an unlabelled one would put someone else's procedure
|
|
in charge of the session."""
|
|
note = fake_note(id=7, title="Drift Audit", user_id=9, note_type="process")
|
|
with patch("scribe.services.notes.resolve_process",
|
|
AsyncMock(return_value=(note, []))), \
|
|
patch("scribe.services.access.describe_provenance",
|
|
AsyncMock(return_value={"shared": True, "owner": "alex",
|
|
"permission": "viewer"})):
|
|
from scribe.mcp.tools.processes import get_process
|
|
out = await get_process("deploy")
|
|
assert out["shared"] is True
|
|
assert out["owner"] == "alex"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_process_not_found_raises():
|
|
with patch("scribe.services.notes.resolve_process",
|
|
AsyncMock(return_value=(None, []))):
|
|
from scribe.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():
|
|
# Resolves share-aware now, so the patch target is get_note_for_user, which
|
|
# returns (note, permission).
|
|
plain = fake_note(id=3, title="Drift Audit", note_type="note")
|
|
plain.deleted_at = None
|
|
with patch("scribe.services.notes.get_note_for_user",
|
|
AsyncMock(return_value=(plain, "owner"))):
|
|
from scribe.mcp.tools.processes import update_process
|
|
with pytest.raises(ValueError):
|
|
await update_process(process_id=3, title="x")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_process_refuses_a_read_only_share_with_the_reason():
|
|
"""An editor grant lets the holder edit someone else's process; a viewer
|
|
grant must be refused, and saying "not found" about a process the caller can
|
|
open would just send them looking for a missing id."""
|
|
theirs = fake_note(id=5, title="Drift Audit", user_id=9, note_type="process")
|
|
theirs.deleted_at = None
|
|
with patch("scribe.services.notes.get_note_for_user",
|
|
AsyncMock(return_value=(theirs, "viewer"))), \
|
|
patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)), \
|
|
patch("scribe.services.notes.update_note", AsyncMock()) as mock_update:
|
|
from scribe.mcp.tools.processes import update_process
|
|
with pytest.raises(ValueError, match="read-only"):
|
|
await update_process(process_id=5, title="x")
|
|
mock_update.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_process_trashes_it_recoverably():
|
|
proc = fake_note(id=4, title="Drift Audit", note_type="process")
|
|
proc.deleted_at = None
|
|
with patch("scribe.services.notes.get_note_for_user",
|
|
AsyncMock(return_value=(proc, "owner"))), \
|
|
patch("scribe.mcp.tools.processes.trash_svc.delete",
|
|
AsyncMock(return_value="batch-1")) as mock_delete:
|
|
from scribe.mcp.tools.processes import delete_process
|
|
out = await delete_process(process_id=4)
|
|
assert out["deleted_batch_id"] == "batch-1"
|
|
# Through the trash, not a hard delete — restorable like every other kind.
|
|
assert mock_delete.await_args.args[1] == "note"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_process_refuses_a_plain_note():
|
|
"""This tool is reached for by name. Letting it trash an ordinary note
|
|
because the id happened to resolve would be a destructive action taken on a
|
|
mistyped argument."""
|
|
plain = fake_note(id=3, title="Drift Audit", note_type="note")
|
|
plain.deleted_at = None
|
|
with patch("scribe.services.notes.get_note_for_user",
|
|
AsyncMock(return_value=(plain, "owner"))), \
|
|
patch("scribe.mcp.tools.processes.trash_svc.delete", AsyncMock()) as mock_delete:
|
|
from scribe.mcp.tools.processes import delete_process
|
|
with pytest.raises(ValueError):
|
|
await delete_process(process_id=3)
|
|
mock_delete.assert_not_awaited()
|
|
|
|
|
|
# ── the moments a procedure is for ───────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_process_stores_the_moments_it_is_for():
|
|
created = fake_note(title="Release", note_type="process",
|
|
data={"moments": ["work.deliver"]})
|
|
with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
|
|
AsyncMock(return_value=None)), \
|
|
patch("scribe.services.notes.create_note",
|
|
AsyncMock(return_value=created)) as mock_create:
|
|
from scribe.mcp.tools.processes import create_process
|
|
out = await create_process(title="Release", body="steps",
|
|
moments=["Work.Deliver", "work.deliver"])
|
|
assert mock_create.await_args.kwargs["data"] == {"moments": ["work.deliver"]}
|
|
assert out["moments"] == ["work.deliver"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_process_refuses_an_unknown_moment_before_writing():
|
|
"""A typo'd moment would be stored and never fire — a procedure that looks
|
|
attached and is not."""
|
|
with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
|
|
AsyncMock(return_value=None)), \
|
|
patch("scribe.services.notes.create_note", AsyncMock()) as mock_create:
|
|
from scribe.mcp.tools.processes import create_process
|
|
with pytest.raises(ValueError, match="unknown moment"):
|
|
await create_process(title="Release", body="steps", moments=["work.shipped"])
|
|
mock_create.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_process_without_moments_stores_no_data():
|
|
created = fake_note(title="Release", note_type="process")
|
|
with patch("scribe.mcp.tools.processes.dedup_svc.find_duplicate_note",
|
|
AsyncMock(return_value=None)), \
|
|
patch("scribe.services.notes.create_note",
|
|
AsyncMock(return_value=created)) as mock_create:
|
|
from scribe.mcp.tools.processes import create_process
|
|
out = await create_process(title="Release", body="steps")
|
|
assert mock_create.await_args.kwargs["data"] is None
|
|
assert out["moments"] == []
|
|
|
|
|
|
async def _update(moments, data):
|
|
proc = fake_note(id=5, title="Release", note_type="process", data=data)
|
|
with patch("scribe.services.notes.get_note_for_user",
|
|
AsyncMock(return_value=(proc, "owner"))), \
|
|
patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)), \
|
|
patch("scribe.services.notes.update_note",
|
|
AsyncMock(return_value=proc)) as mock_update:
|
|
from scribe.mcp.tools.processes import update_process
|
|
await update_process(process_id=5, moments=moments)
|
|
return mock_update.await_args.kwargs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_process_replaces_moments_and_keeps_the_rest_of_data():
|
|
sent = await _update(["work.verify"], {"moments": ["work.deliver"], "other": 1})
|
|
assert sent["data"] == {"moments": ["work.verify"], "other": 1}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_process_empty_moments_clears_them():
|
|
assert (await _update([], {"moments": ["work.deliver"]}))["data"] is None
|
|
assert (await _update([], {"moments": ["work.deliver"], "other": 1}))["data"] == {"other": 1}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_process_without_moments_leaves_data_alone():
|
|
assert "data" not in await _update(None, {"moments": ["work.deliver"]})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_process_says_which_moments_it_is_for():
|
|
note = fake_note(id=7, title="Release", note_type="process",
|
|
data={"moments": ["work.deliver"]})
|
|
with patch("scribe.services.notes.resolve_process", AsyncMock(return_value=(note, []))):
|
|
from scribe.mcp.tools.processes import get_process
|
|
out = await get_process("release")
|
|
assert out["moments"] == ["work.deliver"]
|
|
|
|
|
|
def test_register_attaches_every_tool_in_the_module():
|
|
"""Derived from the module rather than listed: a tool written but never
|
|
registered is invisible to an agent, and nothing else would notice."""
|
|
import inspect
|
|
|
|
from scribe.mcp.tools import processes
|
|
mcp = FakeMCP()
|
|
|
|
processes.register(mcp)
|
|
public = {
|
|
name for name, obj in vars(processes).items()
|
|
if inspect.iscoroutinefunction(obj) and not name.startswith("_")
|
|
}
|
|
assert set(mcp.names) == public
|