bvandeusenandClaude Opus 5
410d616c22
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 10s
CI & Build / TypeScript typecheck (push) Successful in 34s
CI & Build / integration (push) Successful in 32s
CI & Build / Python tests (push) Successful in 1m8s
CI & Build / Build & push image (push) Successful in 35s
The query the last two steps were storage for. `rules_due_for_verification` returns every rule carrying a `verify_with`, ordered by `verified_at` ASC NULLS FIRST, each row carrying the check IN FULL — the opposite call from rule_brief, because the reader is about to go and run it. NULLS FIRST is the ordering this turns on. Postgres sorts NULLs last on an ASC ordering, which would put the rules nobody has ever confirmed BEHIND every rule someone once looked at. Exactly backwards: a claim with no evidence at all outranks an old one. Rules with no check never appear, and that is the property that keeps the list worth reading. Most rules are decisions — no truth value, nothing to go and check. If they appeared here the sweep would be the rulebook. `mark_rule_verified(rule_id, still_true)` closes the loop, asymmetrically: passing writes a stamp, FAILING WRITES NOTHING. There is no "verified false" state because a rule whose check failed is not in a special condition, it is wrong — and recording the failure as a flag would let it sit there being false with the sweep satisfied that someone had looked. So it stays at the top until someone corrects or retires it, and the response says so. An unrecognised `tier` filter raises rather than falling back. _valid_tier's silent always_on default is right for a WRITE — a typo should leave a rule binding — and wrong for a FILTER, where the same fallback quietly answers a different question and returns a short list that reads as good news. Deliberately NOT filterable by project: a project reaches rules through project scope, subscriptions, always-on rulebooks and exclusions, and a filter missing one of those paths would UNDER-report — the exact failure this surface exists to prevent. Said so in the docstring rather than shipping a half-correct filter. Ownership-scoped like every other rule read (owned rulebook, or owned project), in ONE statement with an OR across the XOR rather than two queries merged in Python, so the ordering is the database's and cannot disagree with itself. Note that rules have no sharing ACL in this schema — no rule_shares, no rulebook_shares — so there is no wider set for access.py to consult here. Also fixes a test title that had been lying for ten tools: "all sixteen tools" asserted 26. The number now lives only in the assertion. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
436 lines
18 KiB
Python
436 lines
18 KiB
Python
"""Tests for MCP rulebook tools — patches the service layer."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from tests.helpers import FakeMCP, fake_rule, fake_rulebook, fake_topic
|
|
|
|
|
|
pytestmark = pytest.mark.usefixtures("_bind_user")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_rulebooks_wraps_in_dict():
|
|
rows = [fake_rulebook(id=1, title="t"), fake_rulebook(id=2, title="t")]
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.list_rulebooks",
|
|
AsyncMock(return_value=rows),
|
|
):
|
|
from scribe.mcp.tools.rulebooks import list_rulebooks
|
|
out = await list_rulebooks()
|
|
assert len(out["rulebooks"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rulebook_includes_topics():
|
|
rb = fake_rulebook(id=1, title="t")
|
|
topics = [fake_topic(id=10, title="git"), fake_topic(id=11, title="git")]
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.get_rulebook",
|
|
AsyncMock(return_value=rb),
|
|
), patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.list_topics",
|
|
AsyncMock(return_value=topics),
|
|
):
|
|
from scribe.mcp.tools.rulebooks import get_rulebook
|
|
out = await get_rulebook(rulebook_id=1)
|
|
assert out["id"] == 1
|
|
assert len(out["topics"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rulebook_raises_when_not_found():
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.get_rulebook",
|
|
AsyncMock(return_value=None),
|
|
):
|
|
from scribe.mcp.tools.rulebooks import get_rulebook
|
|
with pytest.raises(ValueError, match="rulebook 999 not found"):
|
|
await get_rulebook(rulebook_id=999)
|
|
|
|
|
|
def _plain_detail():
|
|
"""Stub the rule_detail seam these tool tests are not about.
|
|
|
|
create/update/get_rule now return through services.rulebooks.rule_detail,
|
|
which reads the rule's areas and edges from the database. These are unit
|
|
tests with no database, and what they assert is that the TOOL forwards the
|
|
right arguments — so the seam is stubbed to the plain record, the same way
|
|
they already stub the create/update calls themselves.
|
|
"""
|
|
async def _detail(_uid, rule, _system_ids=None):
|
|
return rule.to_dict()
|
|
return patch("scribe.mcp.tools.rulebooks.rulebooks_svc.rule_detail", _detail)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rule_passes_required_fields():
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule", mock), _plain_detail():
|
|
from scribe.mcp.tools.rulebooks import create_rule
|
|
await create_rule(
|
|
topic_id=10, title="dev is home", statement="Work directly on dev",
|
|
)
|
|
kwargs = mock.call_args.kwargs
|
|
assert kwargs["user_id"] == 7
|
|
assert kwargs["topic_id"] == 10
|
|
assert kwargs["statement"] == "Work directly on dev"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rule_blocked_by_duplicate_gate():
|
|
from scribe.services.dedup import DuplicateMatch
|
|
dup = DuplicateMatch(id=47, title="dev is home", similarity=1.0, reason="title")
|
|
create_mock = AsyncMock()
|
|
with patch("scribe.mcp.tools.rulebooks.dedup_svc.find_duplicate_rule",
|
|
AsyncMock(return_value=dup)), \
|
|
patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule", create_mock):
|
|
from scribe.mcp.tools.rulebooks import create_rule
|
|
out = await create_rule(topic_id=10, title="dev is home", statement="x")
|
|
assert out["duplicate"] is True
|
|
assert out["existing_id"] == 47
|
|
assert "update_rule" in out["message"]
|
|
create_mock.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rule_force_bypasses_duplicate_gate():
|
|
find_mock = AsyncMock()
|
|
with patch("scribe.mcp.tools.rulebooks.dedup_svc.find_duplicate_rule", find_mock), \
|
|
patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule",
|
|
AsyncMock(return_value=fake_rule(id=5, title="r", statement="s", topic_id=10))), \
|
|
_plain_detail():
|
|
from scribe.mcp.tools.rulebooks import create_rule
|
|
out = await create_rule(topic_id=10, title="dev is home", statement="x", force=True)
|
|
assert out["id"] == 5
|
|
find_mock.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rule_only_sends_non_default_fields():
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rule", mock), _plain_detail():
|
|
from scribe.mcp.tools.rulebooks import update_rule
|
|
await update_rule(rule_id=1, statement="new statement")
|
|
args, kwargs = mock.call_args
|
|
assert args == (1, 7)
|
|
# `clear` is always forwarded — an empty tuple is "clear nothing", which is
|
|
# a value, not an absent argument. Everything the caller left at its
|
|
# default stays out: that is the property this test pins.
|
|
assert kwargs == {"statement": "new statement", "clear": ()}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rule_forwards_the_fields_named_for_clearing():
|
|
"""Naming a field is the only way to empty it through this door.
|
|
|
|
"" means "leave unchanged" here, so a caller has no value that means
|
|
"remove it" — which is what makes an explicit list necessary and what
|
|
stops a partial update from wiping the fields it did not mention.
|
|
"""
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rule", mock), _plain_detail():
|
|
from scribe.mcp.tools.rulebooks import update_rule
|
|
await update_rule(rule_id=1, clear_fields=["verify_with"])
|
|
_args, kwargs = mock.call_args
|
|
assert kwargs == {"clear": ["verify_with"]}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rule_sends_the_check_fields_when_given():
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rule", mock), _plain_detail():
|
|
from scribe.mcp.tools.rulebooks import update_rule
|
|
await update_rule(
|
|
rule_id=1,
|
|
verify_with="cat CI-runner/renovate/config.js",
|
|
expires_when="approval is turned off",
|
|
)
|
|
_args, kwargs = mock.call_args
|
|
assert kwargs == {
|
|
"verify_with": "cat CI-runner/renovate/config.js",
|
|
"expires_when": "approval is turned off",
|
|
"clear": (),
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_rule_without_confirmed_returns_warning():
|
|
"""delete_rule with confirmed=False returns a preview, not an action."""
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.get_rule",
|
|
AsyncMock(return_value=rule),
|
|
), patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.delete_rule",
|
|
AsyncMock(),
|
|
) as mock_delete:
|
|
from scribe.mcp.tools.rulebooks import delete_rule
|
|
out = await delete_rule(rule_id=1, confirmed=False)
|
|
assert out.get("confirmed_required") is True
|
|
assert "confirmed=True" in out.get("warning", "")
|
|
assert not mock_delete.called # service NOT called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_rule_with_confirmed_soft_deletes():
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
mock_delete = AsyncMock(return_value="batch-1")
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.get_rule",
|
|
AsyncMock(return_value=rule),
|
|
), patch(
|
|
"scribe.mcp.tools.rulebooks.trash_svc.delete",
|
|
mock_delete,
|
|
):
|
|
from scribe.mcp.tools.rulebooks import delete_rule
|
|
out = await delete_rule(rule_id=1, confirmed=True)
|
|
assert out["deleted"] == 1
|
|
assert out["deleted_batch_id"] == "batch-1"
|
|
assert mock_delete.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subscribe_project_to_rulebook_calls_service():
|
|
mock = AsyncMock()
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.subscribe_project", mock,
|
|
):
|
|
from scribe.mcp.tools.rulebooks import subscribe_project_to_rulebook
|
|
out = await subscribe_project_to_rulebook(project_id=3, rulebook_id=1)
|
|
assert out["subscribed"] is True
|
|
assert mock.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unsubscribe_project_from_rulebook_calls_service():
|
|
mock = AsyncMock()
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.unsubscribe_project", mock,
|
|
):
|
|
from scribe.mcp.tools.rulebooks import unsubscribe_project_from_rulebook
|
|
out = await unsubscribe_project_from_rulebook(project_id=3, rulebook_id=1)
|
|
assert out["subscribed"] is False
|
|
assert mock.called
|
|
|
|
|
|
def test_register_attaches_every_tool():
|
|
"""Every tool in the module reaches the server.
|
|
|
|
The count is the guard: a function added to the module but left out of
|
|
register()'s tuple is invisible to callers and raises nothing. The name
|
|
said "sixteen" for ten tools' worth of growth — the number lives in the
|
|
assertion, not the title, so it cannot drift again.
|
|
"""
|
|
from scribe.mcp.tools.rulebooks import register
|
|
mcp = FakeMCP()
|
|
|
|
register(mcp)
|
|
# 26 through milestone 307, +2 for the staleness sweep (milestone 312).
|
|
assert len(mcp.names) == 28
|
|
# spot-check a few names
|
|
assert "list_rulebooks" in mcp.names
|
|
assert "create_rule" in mcp.names
|
|
assert "subscribe_project_to_rulebook" in mcp.names
|
|
assert "list_always_on_rules" in mcp.names
|
|
# milestone 297: a project's opt-out of a whole always-on rulebook
|
|
assert "exclude_always_on_rulebook" in mcp.names
|
|
assert "include_always_on_rulebook" in mcp.names
|
|
assert "create_project_rule" in mcp.names
|
|
assert "suppress_rule_for_project" in mcp.names
|
|
# milestone 312: the sweep, and the stamp that answers it
|
|
assert "rules_due_for_verification" in mcp.names
|
|
assert "mark_rule_verified" in mcp.names
|
|
assert "unsuppress_rule_for_project" in mcp.names
|
|
assert "suppress_topic_for_project" in mcp.names
|
|
assert "unsuppress_topic_for_project" in mcp.names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_always_on_rules_returns_empty_when_no_always_on_rulebooks():
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.list_always_on_rules",
|
|
AsyncMock(return_value=[]),
|
|
):
|
|
from scribe.mcp.tools.rulebooks import list_always_on_rules
|
|
out = await list_always_on_rules()
|
|
assert out == {"rules": [], "total": 0}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_always_on_rules_projects_each_rule():
|
|
rules = [fake_rule(id=100, title="r", statement="s", topic_id=10), fake_rule(id=101, title="r", statement="s", topic_id=10)]
|
|
with patch(
|
|
"scribe.mcp.tools.rulebooks.rulebooks_svc.list_always_on_rules",
|
|
AsyncMock(return_value=rules),
|
|
):
|
|
from scribe.mcp.tools.rulebooks import list_always_on_rules
|
|
out = await list_always_on_rules()
|
|
assert out["total"] == 2
|
|
assert {r["id"] for r in out["rules"]} == {100, 101}
|
|
assert all("topic_id" in r for r in out["rules"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rulebook_forwards_always_on_when_set():
|
|
rb = fake_rulebook(id=1, title="t")
|
|
mock = AsyncMock(return_value=rb)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rulebook", mock):
|
|
from scribe.mcp.tools.rulebooks import update_rulebook
|
|
await update_rulebook(rulebook_id=1, always_on=True)
|
|
kwargs = mock.call_args.kwargs
|
|
assert kwargs.get("always_on") is True
|
|
assert "title" not in kwargs
|
|
assert "description" not in kwargs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rulebook_omits_always_on_when_none():
|
|
rb = fake_rulebook(id=1, title="t")
|
|
mock = AsyncMock(return_value=rb)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rulebook", mock):
|
|
from scribe.mcp.tools.rulebooks import update_rulebook
|
|
await update_rulebook(rulebook_id=1, title="new title")
|
|
kwargs = mock.call_args.kwargs
|
|
assert "always_on" not in kwargs
|
|
assert kwargs["title"] == "new title"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_project_rule_passes_required_fields():
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock), _plain_detail():
|
|
from scribe.mcp.tools.rulebooks import create_project_rule
|
|
await create_project_rule(
|
|
project_id=42,
|
|
statement="Always run migrations through alembic, not raw SQL.",
|
|
why="audit trail",
|
|
)
|
|
kwargs = mock.call_args.kwargs
|
|
assert kwargs["user_id"] == 7
|
|
assert kwargs["project_id"] == 42
|
|
assert kwargs["statement"].startswith("Always run migrations")
|
|
assert kwargs["why"] == "audit trail"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_project_rule_derives_title_from_statement():
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock), _plain_detail():
|
|
from scribe.mcp.tools.rulebooks import create_project_rule
|
|
await create_project_rule(
|
|
project_id=42,
|
|
statement="Avoid auto-generated docstrings. Reviewers find them noise.",
|
|
)
|
|
kwargs = mock.call_args.kwargs
|
|
# Title should be derived from the first sentence, capped at 50 chars
|
|
assert kwargs["title"] == "Avoid auto-generated docstrings"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_project_rule_uses_explicit_title_when_given():
|
|
rule = fake_rule(id=100, title="r", statement="s", topic_id=10)
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock), _plain_detail():
|
|
from scribe.mcp.tools.rulebooks import create_project_rule
|
|
await create_project_rule(
|
|
project_id=42,
|
|
statement="anything",
|
|
title="no auto-docstrings",
|
|
)
|
|
kwargs = mock.call_args.kwargs
|
|
assert kwargs["title"] == "no auto-docstrings"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_suppress_rule_for_project_passes_through():
|
|
mock = AsyncMock(return_value=None)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.suppress_rule_for_project", mock):
|
|
from scribe.mcp.tools.rulebooks import suppress_rule_for_project
|
|
out = await suppress_rule_for_project(project_id=3, rule_id=17)
|
|
kwargs = mock.call_args.kwargs
|
|
assert kwargs == {"project_id": 3, "rule_id": 17, "user_id": 7}
|
|
assert out == {"project_id": 3, "rule_id": 17, "suppressed": True}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unsuppress_rule_for_project_passes_through():
|
|
mock = AsyncMock(return_value=None)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.unsuppress_rule_for_project", mock):
|
|
from scribe.mcp.tools.rulebooks import unsuppress_rule_for_project
|
|
out = await unsuppress_rule_for_project(project_id=3, rule_id=17)
|
|
kwargs = mock.call_args.kwargs
|
|
assert kwargs == {"project_id": 3, "rule_id": 17, "user_id": 7}
|
|
assert out == {"project_id": 3, "rule_id": 17, "suppressed": False}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_suppress_topic_for_project_passes_through():
|
|
mock = AsyncMock(return_value=None)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.suppress_topic_for_project", mock):
|
|
from scribe.mcp.tools.rulebooks import suppress_topic_for_project
|
|
out = await suppress_topic_for_project(project_id=3, topic_id=22)
|
|
kwargs = mock.call_args.kwargs
|
|
assert kwargs == {"project_id": 3, "topic_id": 22, "user_id": 7}
|
|
assert out == {"project_id": 3, "topic_id": 22, "suppressed": True}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unsuppress_topic_for_project_passes_through():
|
|
mock = AsyncMock(return_value=None)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.unsuppress_topic_for_project", mock):
|
|
from scribe.mcp.tools.rulebooks import unsuppress_topic_for_project
|
|
out = await unsuppress_topic_for_project(project_id=3, topic_id=22)
|
|
kwargs = mock.call_args.kwargs
|
|
assert kwargs == {"project_id": 3, "topic_id": 22, "user_id": 7}
|
|
assert out == {"project_id": 3, "topic_id": 22, "suppressed": False}
|
|
|
|
|
|
# ── Typed edges between rules (milestone 307) ───────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_relate_rules_forwards_the_kind_and_the_why():
|
|
"""The edge exists so a shape stops being merged into one row. The `note`
|
|
travels with it for the same reason a rule carries `why`: whoever later
|
|
decides whether the edge still holds needs the reasoning."""
|
|
relation = MagicMock()
|
|
relation.to_dict.return_value = {"id": 9, "kind": "co_surfaces"}
|
|
mock = AsyncMock(return_value=relation)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.add_rule_relation", mock):
|
|
from scribe.mcp.tools.rulebooks import relate_rules
|
|
out = await relate_rules(
|
|
from_rule_id=46, to_rule_id=144, kind="co_surfaces",
|
|
note="a stale channel tag and an unparseable version both read as "
|
|
"no update available",
|
|
)
|
|
assert out["id"] == 9
|
|
args = mock.call_args.args
|
|
assert args[0] == 7 and args[1] == 46 and args[2] == 144
|
|
assert args[3] == "co_surfaces"
|
|
assert "no update available" in args[4]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_relate_rules_raises_when_either_end_is_not_yours():
|
|
"""The service returns None when it cannot see both rules — a one-sided
|
|
edge would surface a rule the caller has no business reading."""
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.add_rule_relation",
|
|
AsyncMock(return_value=None)):
|
|
from scribe.mcp.tools.rulebooks import relate_rules
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await relate_rules(from_rule_id=1, to_rule_id=2, kind="overrides")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unrelate_rules_raises_when_the_edge_is_gone():
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.remove_rule_relation",
|
|
AsyncMock(return_value=False)):
|
|
from scribe.mcp.tools.rulebooks import unrelate_rules
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await unrelate_rules(relation_id=99)
|