bvandeusenandClaude Opus 5.5
0f5ef0be63
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / TypeScript typecheck (push) Successful in 56s
CI & Build / integration (push) Successful in 1m4s
CI & Build / Python tests (push) Successful in 1m55s
CI & Build / Build & push image (push) Successful in 44s
Step 2 of the Scribe -> Grimoire rename. - src/scribe -> src/grimoire, and every import, patch target, module path, hypercorn entry point and `python -m grimoire.restore` with it; pyproject, uv.lock and the frontend package are named grimoire. - Image git.fabledsword.com/bvandeusen/fabledgrimoire; CI's Postgres service runs as grimoire / grimoire_test. - Compose files default the DB user, database and (prod) volume to grimoire, each overridable from .env, so an install created as Scribe keeps its data by setting POSTGRES_USER, POSTGRES_DB and PGDATA_VOLUME. - Backups are written as grimoire-<ts>.tar with format grimoire-backup-tar. Archives named scribe-* with the scribe-backup-tar manifest still restore and are still counted at the destination (folder and S3), with tests. - Exports, the MCP auth realm, the ASGI scope key and the persistent logger carry the new name. Kept on purpose: the secret-sealing key label stays "scribe.secrets.v1", because every stored secret was derived under it. Plugin identity (plugin name, hooks, env vars, .scribe marker, skills) is step 3. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
354 lines
18 KiB
Python
354 lines
18 KiB
Python
"""MCP system tools + task/note issue wiring (service layer mocked)."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from tests.helpers import fake_note, fake_system, fake_task
|
|
|
|
|
|
# The name assessment both doors run before minting. A test
|
|
# that patches systems_svc wholesale must stub it, or the awaited MagicMock
|
|
# raises — this shape is the "nothing matched" answer.
|
|
_NO_MATCH = {"duplicate": None, "canonical": None}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_system_returns_dict():
|
|
with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value=_NO_MATCH)
|
|
svc.create_system = AsyncMock(return_value=fake_system(name="Reader"))
|
|
from grimoire.mcp.tools.systems import create_system
|
|
result = await create_system(project_id=5, name="Reader", description="pdf reader")
|
|
assert result["name"] == "Reader"
|
|
# An unmatched name is a project-specific area: created, no offer, no fuss.
|
|
assert "canonical_suggestion" not in result and "canonical_note" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_system_no_access_raises():
|
|
with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value=_NO_MATCH)
|
|
svc.create_system = AsyncMock(return_value=None)
|
|
from grimoire.mcp.tools.systems import create_system
|
|
with pytest.raises(ValueError):
|
|
await create_system(project_id=5, name="Reader")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_system_applies_an_exact_area_and_offers_a_similar_one():
|
|
"""The two bases must behave differently, and this is where it is decided.
|
|
|
|
`exact` differs from the catalog name only in spelling, so it is APPLIED —
|
|
that is the mechanical case the catalog exists to collapse. `overlap` is a
|
|
judgment call, so it is only OFFERED: applying it silently is how a
|
|
cross-project rule ends up surfacing in the wrong project.
|
|
"""
|
|
from grimoire.mcp.tools.systems import create_system
|
|
|
|
with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value={
|
|
"duplicate": None,
|
|
"canonical": {"id": 3, "name": "CI & Release", "basis": "exact"},
|
|
})
|
|
svc.create_system = AsyncMock(return_value=fake_system(name="CI and Release"))
|
|
exact = await create_system(project_id=5, name="CI and Release")
|
|
assert svc.create_system.await_args.kwargs["canonical_id"] == 3
|
|
assert "canonical_note" in exact and "canonical_suggestion" not in exact
|
|
|
|
with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value={
|
|
"duplicate": None,
|
|
"canonical": {"id": 3, "name": "CI & Release", "basis": "overlap", "score": 0.33},
|
|
})
|
|
svc.create_system = AsyncMock(return_value=fake_system(id=9, name="CI & runners"))
|
|
similar = await create_system(project_id=5, name="CI & runners")
|
|
assert svc.create_system.await_args.kwargs["canonical_id"] is None
|
|
assert similar["canonical_suggestion"]["id"] == 3
|
|
assert "map_system_to_canonical(9, 3)" in similar["canonical_suggestion"]["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_system_duplicate_names_the_existing_one_and_creates_nothing():
|
|
from grimoire.mcp.tools.systems import create_system
|
|
with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value={
|
|
"duplicate": {"id": 4, "name": "Reader"}, "canonical": None,
|
|
})
|
|
svc.create_system = AsyncMock()
|
|
result = await create_system(project_id=5, name="reader")
|
|
assert result["duplicate"] is True and result["existing_id"] == 4
|
|
assert "Reader" in result["message"]
|
|
svc.create_system.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_system_splits_records_by_kind():
|
|
issue = fake_task(id=10, task_kind="issue", milestone_id=None)
|
|
work = fake_task(id=11, milestone_id=None)
|
|
note = fake_note(id=12, status=None, milestone_id=None)
|
|
with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.get_system = AsyncMock(return_value=fake_system(id=3))
|
|
svc.list_records_for_system = AsyncMock(return_value=[issue, work, note])
|
|
from grimoire.mcp.tools.systems import get_system
|
|
result = await get_system(system_id=3)
|
|
assert [r["id"] for r in result["issues"]] == [10]
|
|
assert [r["id"] for r in result["tasks"]] == [11]
|
|
assert [r["id"] for r in result["notes"]] == [12]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_patterns_reach_the_service_from_both_tools():
|
|
"""Omitted means unchanged; an empty list is a value — it clears them."""
|
|
with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value=_NO_MATCH)
|
|
svc.create_system = AsyncMock(return_value=fake_system(name="Billing"))
|
|
svc.update_system = AsyncMock(return_value=fake_system(name="Billing"))
|
|
from grimoire.mcp.tools.systems import create_system, update_system
|
|
await create_system(project_id=5, name="Billing", path_patterns=["src/billing"])
|
|
assert svc.create_system.await_args.kwargs["path_patterns"] == ["src/billing"]
|
|
await update_system(system_id=1, name="Billing")
|
|
assert "path_patterns" not in svc.update_system.await_args.kwargs
|
|
await update_system(system_id=1, path_patterns=[])
|
|
assert svc.update_system.await_args.kwargs["path_patterns"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_system_not_found_raises():
|
|
with patch("grimoire.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.update_system = AsyncMock(return_value=None)
|
|
from grimoire.mcp.tools.systems import update_system
|
|
with pytest.raises(ValueError):
|
|
await update_system(system_id=99, name="x")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_issue_sets_kind_provenance_and_systems():
|
|
note = MagicMock(); note.id = 50; note.to_dict.return_value = {"id": 50, "task_kind": "issue"}
|
|
with patch("grimoire.mcp.tools.tasks.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.tasks.notes_svc") as notes_svc, \
|
|
patch("grimoire.mcp.tools.tasks.systems_svc") as systems_svc:
|
|
notes_svc.create_note = AsyncMock(return_value=note)
|
|
systems_svc.set_record_systems = AsyncMock(return_value=[2, 3])
|
|
systems_svc.list_record_systems = AsyncMock(return_value=[])
|
|
from grimoire.mcp.tools.tasks import create_task
|
|
result = await create_task(title="bug", kind="issue", system_ids=[2, 3], arose_from_id=9)
|
|
_, kwargs = notes_svc.create_note.call_args
|
|
assert kwargs["task_kind"] == "issue"
|
|
assert kwargs["arose_from_id"] == 9
|
|
systems_svc.set_record_systems.assert_awaited_once_with(1, 50, [2, 3])
|
|
assert result["id"] == 50
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_untagged_hint_names_the_projects_systems():
|
|
sys_a = MagicMock(); sys_a.id = 1; sys_a.name = "workers"
|
|
sys_b = MagicMock(); sys_b.id = 2; sys_b.name = "scrape-pipeline"
|
|
with patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
svc.list_systems = AsyncMock(return_value=[sys_a, sys_b])
|
|
from grimoire.mcp.tools.systems import untagged_systems_hint
|
|
hint = await untagged_systems_hint(1, 5)
|
|
assert "workers" in hint and "scrape-pipeline" in hint
|
|
assert "system_ids" in hint
|
|
assert "create_system" in hint
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_untagged_hint_zero_systems_prompts_first_create_and_fails_open():
|
|
from grimoire.mcp.tools.systems import untagged_systems_hint
|
|
with patch("grimoire.mcp.tools.systems.systems_svc") as svc, \
|
|
patch("grimoire.mcp.tools.systems.notes_svc") as notes:
|
|
# Zero Systems in a YOUNG project (below the bootstrap threshold):
|
|
# the mild question stays proportionate — it must prompt the FIRST
|
|
# create_system, not go silent.
|
|
svc.list_systems = AsyncMock(return_value=[])
|
|
notes.list_notes = AsyncMock(return_value=([], 3))
|
|
hint = await untagged_systems_hint(1, 5)
|
|
assert "no Systems yet" in hint and "create_system" in hint
|
|
with patch("grimoire.mcp.tools.systems.systems_svc") as svc:
|
|
# A hint must never break a create — DB failure degrades to no hint.
|
|
svc.list_systems = AsyncMock(side_effect=RuntimeError("db down"))
|
|
assert await untagged_systems_hint(1, 5) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_untagged_hint_escalates_in_a_mature_zero_systems_project():
|
|
"""In a mature project with zero Systems the generic question does not
|
|
convert. The escalated ask must carry the
|
|
project's OWN evidence — record count and recent titles — and demand a
|
|
concrete deliverable, because that is the property separating the nudges
|
|
that convert from the prose that doesn't."""
|
|
from grimoire.mcp.tools.systems import untagged_systems_hint
|
|
recent = [fake_note(title="Fix scrape retry backoff"), fake_note(title="Worker pool sizing")]
|
|
with patch("grimoire.mcp.tools.systems.systems_svc") as svc, \
|
|
patch("grimoire.mcp.tools.systems.notes_svc") as notes:
|
|
svc.list_systems = AsyncMock(return_value=[])
|
|
notes.list_notes = AsyncMock(return_value=(recent, 282))
|
|
# The standard names come from the GLOBAL catalog,
|
|
# not a module constant — so the ask reads them through the service.
|
|
# That the SEEDED vocabulary is these eight is migration 0087's
|
|
# business, asserted against a real database in the inception
|
|
# integration test; what belongs here is that whatever the catalog
|
|
# holds reaches the ask verbatim.
|
|
svc.standard_systems = AsyncMock(return_value=[
|
|
("CI & Release", "..."), ("Auth & Access", "..."),
|
|
])
|
|
hint = await untagged_systems_hint(1, 5)
|
|
assert "282 records" in hint
|
|
assert "Fix scrape retry backoff" in hint # the project's own evidence
|
|
assert "3-6" in hint and "create_system" in hint
|
|
# NOT an approval flow: the agent mints directly, and the
|
|
# standard cross-project names carry the consistency instead.
|
|
assert "without asking permission" in hint
|
|
assert "propose" not in hint and "confirmed" not in hint
|
|
assert "CI & Release" in hint and "Auth & Access" in hint
|
|
# The generic wording is REPLACED, not appended — two questions is noise.
|
|
assert "no Systems yet" not in hint
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bootstrap_ask_still_asks_when_the_catalog_is_unreachable():
|
|
"""The standard names are an AID to the question, not the question. If the
|
|
catalog read fails (or an install has an empty one), the ask must still
|
|
carry the project's evidence and demand the same deliverable — degrading to
|
|
a weaker nudge is acceptable, going silent is not."""
|
|
from grimoire.mcp.tools.systems import untagged_systems_hint
|
|
recent = [fake_note(title="Fix scrape retry backoff")]
|
|
with patch("grimoire.mcp.tools.systems.systems_svc") as svc, \
|
|
patch("grimoire.mcp.tools.systems.notes_svc") as notes:
|
|
svc.list_systems = AsyncMock(return_value=[])
|
|
notes.list_notes = AsyncMock(return_value=(recent, 282))
|
|
svc.standard_systems = AsyncMock(side_effect=RuntimeError("db down"))
|
|
hint = await untagged_systems_hint(1, 5)
|
|
assert hint is not None
|
|
assert "282 records" in hint and "create_system" in hint
|
|
assert "3-6" in hint and "without asking permission" in hint
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bootstrap_ask_stays_quiet_below_threshold_and_fails_open():
|
|
from grimoire.mcp.tools import systems as tools
|
|
with patch("grimoire.mcp.tools.systems.notes_svc") as notes:
|
|
notes.list_notes = AsyncMock(
|
|
return_value=([], tools._BOOTSTRAP_MIN_RECORDS - 1))
|
|
assert await tools.bootstrap_systems_ask(1, 5) is None
|
|
with patch("grimoire.mcp.tools.systems.notes_svc") as notes:
|
|
# The record count is decoration on a hint on a create — three layers
|
|
# deep, nothing there may break the call.
|
|
notes.list_notes = AsyncMock(side_effect=RuntimeError("db down"))
|
|
assert await tools.bootstrap_systems_ask(1, 5) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_populated_vocabulary_never_counts_records():
|
|
"""The bootstrap query is zero-state-only: once Systems exist the hint
|
|
must not spend a count query per untagged record."""
|
|
from grimoire.mcp.tools.systems import untagged_systems_hint
|
|
sys_a = MagicMock(); sys_a.id = 1; sys_a.name = "workers"
|
|
with patch("grimoire.mcp.tools.systems.systems_svc") as svc, \
|
|
patch("grimoire.mcp.tools.systems.notes_svc") as notes:
|
|
svc.list_systems = AsyncMock(return_value=[sys_a])
|
|
notes.list_notes = AsyncMock()
|
|
hint = await untagged_systems_hint(1, 5)
|
|
assert "workers" in hint
|
|
notes.list_notes.assert_not_awaited()
|
|
|
|
|
|
# The name gate's own cases (case/whitespace normalisation, exact-over-overlap,
|
|
# fail-open) moved to tests/test_services_systems.py with the logic itself —
|
|
# services/systems.assess_system_name, so both doors share one answer.
|
|
# What stays the TOOL's job — rendering a duplicate, applying an exact area,
|
|
# offering an overlap — is covered at the top of this file.
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_untagged_in_project_carries_the_question():
|
|
note = MagicMock(); note.id = 60; note.to_dict.return_value = {"id": 60}
|
|
sys_a = MagicMock(); sys_a.id = 4; sys_a.name = "plugin-hooks"
|
|
with patch("grimoire.mcp.tools.tasks.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.tasks.notes_svc") as notes_svc, \
|
|
patch("grimoire.mcp.tools.tasks.dedup_svc") as dedup_svc, \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as seam_svc:
|
|
notes_svc.create_note = AsyncMock(return_value=note)
|
|
dedup_svc.find_duplicate_note = AsyncMock(return_value=None)
|
|
seam_svc.list_record_systems = AsyncMock(return_value=[])
|
|
seam_svc.list_systems = AsyncMock(return_value=[sys_a])
|
|
from grimoire.mcp.tools.tasks import create_task
|
|
result = await create_task(title="untagged", project_id=5)
|
|
assert "plugin-hooks" in result["systems_hint"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_tagged_shows_systems_and_projectless_gets_neither():
|
|
note = MagicMock(); note.id = 61
|
|
# Fresh dict per call — a shared return_value dict lets the first create's
|
|
# mutation leak into the second create's response.
|
|
note.to_dict.side_effect = lambda: {"id": 61}
|
|
tagged_sys = MagicMock(); tagged_sys.to_dict.return_value = {"id": 4, "name": "plugin-hooks"}
|
|
with patch("grimoire.mcp.tools.tasks.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.tasks.notes_svc") as notes_svc, \
|
|
patch("grimoire.mcp.tools.tasks.dedup_svc") as dedup_svc, \
|
|
patch("grimoire.mcp.tools.tasks.systems_svc") as systems_svc, \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as seam_svc:
|
|
notes_svc.create_note = AsyncMock(return_value=note)
|
|
dedup_svc.find_duplicate_note = AsyncMock(return_value=None)
|
|
systems_svc.set_record_systems = AsyncMock()
|
|
seam_svc.list_record_systems = AsyncMock(return_value=[tagged_sys])
|
|
from grimoire.mcp.tools.tasks import create_task
|
|
tagged = await create_task(title="tagged", project_id=5, system_ids=[4])
|
|
seam_svc.list_record_systems = AsyncMock(return_value=[])
|
|
orphan = await create_task(title="orphan", project_id=0)
|
|
assert tagged["systems"] == [{"id": 4, "name": "plugin-hooks"}]
|
|
assert "systems_hint" not in tagged
|
|
assert "systems" not in orphan and "systems_hint" not in orphan
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_task_shows_the_records_systems_on_read():
|
|
"""The touching-a-System reflex needs the affiliation visible on READ —
|
|
a session opening the task it is about to work must see its areas."""
|
|
note = MagicMock()
|
|
note.id = 70; note.user_id = 1; note.project_id = 5
|
|
note.deleted_at = None; note.parent_id = None
|
|
note.to_dict.return_value = {"id": 70, "task_kind": "work"}
|
|
tagged_sys = MagicMock(); tagged_sys.to_dict.return_value = {"id": 2, "name": "exporter"}
|
|
with patch("grimoire.mcp.tools.tasks.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.tasks.notes_svc") as notes_svc, \
|
|
patch("grimoire.mcp.tools.tasks.access_svc") as access_svc, \
|
|
patch("grimoire.mcp.tools.tasks.record_pulled"), \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as seam_svc:
|
|
notes_svc.get_note_for_user = AsyncMock(return_value=(note, "owner"))
|
|
access_svc.describe_provenance = AsyncMock(return_value={})
|
|
seam_svc.list_record_systems = AsyncMock(return_value=[tagged_sys])
|
|
from grimoire.mcp.tools.tasks import get_task
|
|
result = await get_task(task_id=70)
|
|
assert result["systems"] == [{"id": 2, "name": "exporter"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_task_log_on_untagged_project_task_asks_the_question():
|
|
"""Logging work IS working in some area — the strongest moment to ask."""
|
|
log = MagicMock(); log.to_dict.return_value = {"id": 9, "task_id": 70}
|
|
task = MagicMock(); task.user_id = 1; task.project_id = 5
|
|
with patch("grimoire.mcp.tools.tasks.current_user_id", return_value=1), \
|
|
patch("grimoire.mcp.tools.tasks.task_logs_svc") as logs_svc, \
|
|
patch("grimoire.mcp.tools.tasks.notes_svc") as notes_svc, \
|
|
patch("grimoire.mcp.tools.systems.systems_svc") as seam_svc, \
|
|
patch("grimoire.mcp.tools.systems.notes_svc") as seam_notes:
|
|
logs_svc.create_log = AsyncMock(return_value=log)
|
|
notes_svc.get_note_for_user = AsyncMock(return_value=(task, "owner"))
|
|
seam_svc.list_record_systems = AsyncMock(return_value=[])
|
|
seam_svc.list_systems = AsyncMock(return_value=[])
|
|
# Young project: below the bootstrap threshold, so the mild
|
|
# question is the expected form here.
|
|
seam_notes.list_notes = AsyncMock(return_value=([], 2))
|
|
from grimoire.mcp.tools.tasks import add_task_log
|
|
result = await add_task_log(task_id=70, content="progress")
|
|
assert "no Systems yet" in result["systems_hint"]
|