bvandeusen
b255a0f90e
Renames src/fabledassistant -> src/scribe and all imports, plus the default DB name and DB user/password (fabled -> scribe) in config + compose. 952 refs / 154 files. Reverses the old 'internal name stays fabledassistant' convention. Code-only: live databases are still physically named 'fabledassistant'. Deployed environments must set POSTGRES_DB / POSTGRES_USER (or rename the DB) since the defaults now resolve to 'scribe'. Repo (FabledScribe), git host (fabledsword), MCP (fabled-git) and the image name (fabledscribe) are intentionally unchanged. ruff check src/ clean locally; CI (typecheck + pytest) is the gate. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
335 lines
12 KiB
Python
335 lines
12 KiB
Python
"""Tests for MCP rulebook tools — patches the service layer."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from scribe.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_rulebook(id=1, title="t"):
|
|
rb = MagicMock()
|
|
rb.id = id
|
|
rb.title = title
|
|
rb.to_dict.return_value = {"id": id, "title": title}
|
|
return rb
|
|
|
|
|
|
def _fake_topic(id=10, title="git"):
|
|
t = MagicMock()
|
|
t.id = id
|
|
t.title = title
|
|
t.to_dict.return_value = {"id": id, "title": title}
|
|
return t
|
|
|
|
|
|
def _fake_rule(id=100, title="r", statement="s"):
|
|
r = MagicMock()
|
|
r.id = id
|
|
r.title = title
|
|
r.topic_id = 10
|
|
r.statement = statement
|
|
r.to_dict.return_value = {"id": id, "title": title, "statement": statement}
|
|
return r
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_rulebooks_wraps_in_dict():
|
|
rows = [_fake_rulebook(id=1), _fake_rulebook(id=2)]
|
|
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)
|
|
topics = [_fake_topic(id=10), _fake_topic(id=11)]
|
|
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)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rule_passes_required_fields():
|
|
rule = _fake_rule()
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule", mock):
|
|
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_update_rule_only_sends_non_default_fields():
|
|
rule = _fake_rule()
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rule", mock):
|
|
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)
|
|
assert kwargs == {"statement": "new statement"}
|
|
|
|
|
|
@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()
|
|
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()
|
|
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_all_sixteen_tools():
|
|
"""register(mcp) should call mcp.tool(name=...) for all 16 tools."""
|
|
from scribe.mcp.tools.rulebooks import register
|
|
registered: list[str] = []
|
|
|
|
class FakeMCP:
|
|
def tool(self, name=None):
|
|
def decorator(fn):
|
|
registered.append(name)
|
|
return fn
|
|
return decorator
|
|
|
|
register(FakeMCP())
|
|
assert len(registered) == 22
|
|
# spot-check a few names
|
|
assert "list_rulebooks" in registered
|
|
assert "create_rule" in registered
|
|
assert "subscribe_project_to_rulebook" in registered
|
|
assert "list_always_on_rules" in registered
|
|
assert "create_project_rule" in registered
|
|
assert "suppress_rule_for_project" in registered
|
|
assert "unsuppress_rule_for_project" in registered
|
|
assert "suppress_topic_for_project" in registered
|
|
assert "unsuppress_topic_for_project" in registered
|
|
|
|
|
|
@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), _fake_rule(id=101)]
|
|
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()
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock):
|
|
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()
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock):
|
|
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()
|
|
mock = AsyncMock(return_value=rule)
|
|
with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_project_rule", mock):
|
|
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}
|