bvandeusenandClaude Opus 5
ffb7a0fe38
CI & Build / Python lint (push) Successful in 5s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / integration (push) Successful in 21s
CI & Build / TypeScript typecheck (push) Successful in 33s
CI & Build / Python tests (push) Successful in 1m7s
CI & Build / Build & push image (push) Successful in 24s
MCP and REST both gain when_to_apply / tier / system_ids / arose_from_id on create and update, plus relate_rules / unrelate_rules for the typed edges, and get_rule now returns a rule's areas and relations alongside it. rule_detail() is a SERVICE function, not one per door. It started as a copy in each — identical, and the prior-art hook flagged it immediately, which is the same lesson rules_payload (#2858) already recorded: a second copy drifts. Both doors call the one seam, so create, update and get cannot disagree about what a rule looks like coming back. The authoring guidance lands in create_rule's docstring rather than in a rule, per rule 119 as the operator described it: this is behaviour every instance should inherit, not one operator's preference. It states the test — ONE RULE = ONE THING YOU COULD VIOLATE. Rules that FAIL TOGETHER get linked with relate_rules(kind="co_surfaces"), never merged into one row. — and names why merging loses: a merged rule cannot be cited, surfaced or suppressed a clause at a time, and it grows without limit because adding to it is always cheaper than adding a rule. create_project_rule says the same about "overrides", which is what FabledCurator's 85/86 should have been instead of near-copies that drift from their parent. The tier arg carries the test itself: can you name the trigger WITHOUT naming a system, an artifact type or a moment? If the honest answer is "whenever you are working", it is always_on. Tests: the applicable-rules cases fabricated raw tuples matching the old column lists, so they move to the entity shape via fake_rule; new cases pin rule_brief (a DATE not a stamp, the depth left to get_rule, no null keys) and that an unknown tier falls back to BINDING. fake_rule gains when_to_apply / tier / arose_from_id for the note-2109 reason the helper exists: unnamed, they would be truthy MagicMocks. The tool tests stub the new rule_detail seam — they are about argument forwarding and have no database. The module header's "Sixteen tools" had been wrong for two milestones; the registration count test is what actually catches that, so the header now says so instead of carrying a number. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
368 lines
15 KiB
Python
368 lines
15 KiB
Python
"""Tests for services/rulebooks.py — mocks async_session, no real DB.
|
|
|
|
Mirrors the pattern in tests/test_events_service.py.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from tests.helpers import fake_rule, fake_rulebook, fake_topic, make_mock_session
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _no_exclusions():
|
|
"""get_applicable_rules asks for the project's always-on exclusions
|
|
(milestone 297) through its own session; these mocked-session tests
|
|
script the rule queries only, so the exclusions lookup is stubbed empty."""
|
|
with patch("scribe.services.rulebooks.excluded_always_on_rulebooks", AsyncMock(return_value=[])):
|
|
yield
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rulebook_stores_to_db():
|
|
mock_session = make_mock_session()
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import create_rulebook
|
|
await create_rulebook(
|
|
user_id=7, title="FabledSword family", description="rules for the family",
|
|
)
|
|
assert mock_session.add.called
|
|
assert mock_session.commit.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_rulebooks_returns_owned_only():
|
|
rb = fake_rulebook(id=1)
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [rb]
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import list_rulebooks
|
|
results = await list_rulebooks(user_id=7)
|
|
assert len(results) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rulebook_returns_none_when_not_owner():
|
|
"""get_rulebook scopes by owner_user_id — wrong user gets None."""
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import get_rulebook
|
|
result = await get_rulebook(rulebook_id=1, user_id=99)
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rulebook_only_sets_provided_fields():
|
|
rb = fake_rulebook(id=1, title="old")
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = rb
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import update_rulebook
|
|
await update_rulebook(rulebook_id=1, user_id=7, title="new")
|
|
assert rb.title == "new"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_rulebook_calls_delete():
|
|
rb = fake_rulebook(id=1)
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = rb
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
mock_session.delete = AsyncMock()
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import delete_rulebook
|
|
await delete_rulebook(rulebook_id=1, user_id=7)
|
|
assert mock_session.delete.called
|
|
|
|
|
|
# ── Topic CRUD ───────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_topic_requires_owned_rulebook():
|
|
"""create_topic raises ValueError if the rulebook isn't owned by user."""
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import create_topic
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await create_topic(
|
|
rulebook_id=999, user_id=7, title="git-workflow",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_topics_returns_topics_for_owned_rulebook():
|
|
rb = fake_rulebook(id=1)
|
|
topic = fake_topic(id=10, rulebook_id=1, title="git-workflow")
|
|
|
|
# Two execute calls: ownership check, then topic select.
|
|
mock_session = make_mock_session()
|
|
rb_result = MagicMock()
|
|
rb_result.scalar_one_or_none.return_value = rb
|
|
topic_result = MagicMock()
|
|
topic_result.scalars.return_value.all.return_value = [topic]
|
|
mock_session.execute = AsyncMock(side_effect=[rb_result, topic_result])
|
|
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import list_topics
|
|
results = await list_topics(rulebook_id=1, user_id=7)
|
|
assert len(results) == 1
|
|
assert results[0].title == "git-workflow"
|
|
|
|
|
|
# ── Rule CRUD ───────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rule_requires_owned_topic():
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None # topic not found
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import create_rule
|
|
with pytest.raises(ValueError, match="topic .* not found"):
|
|
await create_rule(
|
|
topic_id=999, user_id=7, title="x", statement="y",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_rules_filters_by_topic_id():
|
|
"""list_rules(topic_id=X) returns rules in that topic, ownership-scoped."""
|
|
rule = fake_rule(id=1, topic_id=10)
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [rule]
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import list_rules
|
|
results = await list_rules(user_id=7, topic_id=10)
|
|
assert len(results) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rule_returns_none_when_not_owner():
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import get_rule
|
|
result = await get_rule(rule_id=1, user_id=99)
|
|
assert result is None
|
|
|
|
|
|
# ── Subscriptions + applicable_rules ────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subscribe_project_requires_owned_rulebook():
|
|
"""subscribe_project raises if user doesn't own the rulebook."""
|
|
mock_session = make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import subscribe_project
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await subscribe_project(
|
|
project_id=1, rulebook_id=999, user_id=7,
|
|
)
|
|
|
|
|
|
def _empty():
|
|
"""A MagicMock result whose .all() returns [] (or .scalars().all() returns [])."""
|
|
r = MagicMock()
|
|
r.all.return_value = []
|
|
r.scalars.return_value.all.return_value = []
|
|
return r
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_applicable_rules_returns_shape():
|
|
"""get_applicable_rules returns the full projection — including the
|
|
new suppression fields and rulebook/topic IDs on each rule."""
|
|
mock_session = make_mock_session()
|
|
sub_result = MagicMock()
|
|
sub_result.all.return_value = [(1, "FabledSword family")]
|
|
rules_result = MagicMock()
|
|
rules_result.all.return_value = [
|
|
# The query selects the ENTITY plus three labels, so rule_brief stays
|
|
# the one place deciding what a surfaced rule carries (note 3026).
|
|
(fake_rule(id=i, title=f"Rule {i}", statement=f"Statement {i}", topic_id=2),
|
|
"git-workflow", 1, "FabledSword family")
|
|
for i in range(50)
|
|
]
|
|
# Execute order: sub_q, suppressed_rules_q, suppressed_topics_q, rules_q, proj_rules_q
|
|
mock_session.execute = AsyncMock(side_effect=[
|
|
sub_result, _empty(), _empty(), rules_result, _empty(),
|
|
])
|
|
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import get_applicable_rules
|
|
result = await get_applicable_rules(project_id=3, user_id=7, limit=50)
|
|
|
|
assert "rules" in result
|
|
assert "project_rules" in result
|
|
assert "suppressed_rules" in result
|
|
assert "suppressed_topics" in result
|
|
assert "truncated" in result
|
|
assert "subscribed_rulebooks" in result
|
|
assert result["subscribed_rulebooks"] == [{"id": 1, "title": "FabledSword family"}]
|
|
assert len(result["rules"]) == 50
|
|
assert result["rules"][0]["topic_id"] == 2
|
|
assert result["rules"][0]["rulebook_id"] == 1
|
|
assert result["project_rules"] == []
|
|
assert result["suppressed_rules"] == []
|
|
assert result["suppressed_topics"] == []
|
|
assert result["truncated"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_applicable_rules_truncates_when_over_limit():
|
|
"""When limit+1 rows are returned, truncated=True and only `limit` returned."""
|
|
mock_session = make_mock_session()
|
|
sub_result = MagicMock()
|
|
sub_result.all.return_value = []
|
|
rules_result = MagicMock()
|
|
rules_result.all.return_value = [
|
|
(fake_rule(id=i, title=f"r{i}"), "topic", 1, "rb") for i in range(51)
|
|
]
|
|
mock_session.execute = AsyncMock(side_effect=[
|
|
sub_result, _empty(), _empty(), rules_result, _empty(),
|
|
])
|
|
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import get_applicable_rules
|
|
result = await get_applicable_rules(project_id=3, user_id=7, limit=50)
|
|
|
|
assert result["truncated"] is True
|
|
assert len(result["rules"]) == 50
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_applicable_rules_includes_project_scoped_rules():
|
|
"""Project-scoped rules surface in the project_rules field."""
|
|
mock_session = make_mock_session()
|
|
proj_rules_result = MagicMock()
|
|
proj_rules_result.all.return_value = [
|
|
(fake_rule(id=100, topic_id=None, project_id=3, title="Use alembic",
|
|
statement="Always run migrations via alembic, never raw SQL."),),
|
|
(fake_rule(id=101, topic_id=None, project_id=3, title="PR-bound",
|
|
statement="Land schema changes in their own PR."),),
|
|
]
|
|
mock_session.execute = AsyncMock(side_effect=[
|
|
_empty(), _empty(), _empty(), _empty(), proj_rules_result,
|
|
])
|
|
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import get_applicable_rules
|
|
result = await get_applicable_rules(project_id=3, user_id=7)
|
|
|
|
assert len(result["project_rules"]) == 2
|
|
assert result["project_rules"][0]["title"] == "Use alembic"
|
|
assert result["project_rules"][1]["id"] == 101
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_applicable_rules_surfaces_suppressed_with_context():
|
|
"""Suppressed rules and topics come back with full title + rulebook context
|
|
so the UI can render them without an extra round-trip."""
|
|
mock_session = make_mock_session()
|
|
suppressed_rules_result = MagicMock()
|
|
suppressed_rules_result.all.return_value = [
|
|
# (rule_id, title, topic_id, topic_title, rulebook_id, rulebook_title)
|
|
(17, "Old rule", 5, "old-topic", 1, "FabledSword family"),
|
|
]
|
|
suppressed_topics_result = MagicMock()
|
|
suppressed_topics_result.all.return_value = [
|
|
# (topic_id, topic_title, rulebook_id, rulebook_title)
|
|
(22, "design-system", 1, "FabledSword family"),
|
|
]
|
|
mock_session.execute = AsyncMock(side_effect=[
|
|
_empty(), suppressed_rules_result, suppressed_topics_result, _empty(), _empty(),
|
|
])
|
|
|
|
with patch("scribe.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.rulebooks import get_applicable_rules
|
|
result = await get_applicable_rules(project_id=3, user_id=7)
|
|
|
|
assert len(result["suppressed_rules"]) == 1
|
|
assert result["suppressed_rules"][0]["id"] == 17
|
|
assert result["suppressed_rules"][0]["rulebook_title"] == "FabledSword family"
|
|
assert len(result["suppressed_topics"]) == 1
|
|
assert result["suppressed_topics"][0]["title"] == "design-system"
|
|
|
|
|
|
# ── rule_brief + tier (milestone 307) ───────────────────────────────────
|
|
|
|
def test_rule_brief_carries_age_but_not_the_deep_fields():
|
|
"""The shape a SURFACED rule takes, and the reason it exists.
|
|
|
|
There were three hand-written copies of this dict and they had already
|
|
diverged — none carried the timestamps the model has always held, which is
|
|
why a rule written before the capability it duplicates was
|
|
indistinguishable at read time from one still doing work (note 3026).
|
|
"""
|
|
from scribe.services.rulebooks import rule_brief
|
|
|
|
out = rule_brief(fake_rule(
|
|
when_to_apply="before any git push",
|
|
updated_at=datetime(2026, 6, 1, 14, 30, tzinfo=timezone.utc),
|
|
))
|
|
assert out["when_to_apply"] == "before any git push"
|
|
assert out["tier"] == "always_on"
|
|
# A DATE, not a stamp: the question is "how old is this", and a full ISO
|
|
# string across the always-on set is ~2k characters of payload.
|
|
assert out["updated_at"] == "2026-06-01"
|
|
# The depth stays with get_rule — putting it in every listing is the bloat
|
|
# this milestone is about.
|
|
assert "why" not in out and "how_to_apply" not in out
|
|
|
|
|
|
def test_rule_brief_omits_keys_a_rule_has_no_value_for():
|
|
"""#2483: a null key reads as a capability the record has and isn't using,
|
|
which is a different claim from not having one."""
|
|
from scribe.services.rulebooks import rule_brief
|
|
|
|
out = rule_brief(fake_rule())
|
|
assert "when_to_apply" not in out
|
|
assert "arose_from_id" not in out
|
|
|
|
|
|
def test_an_unknown_tier_falls_back_to_binding():
|
|
"""The asymmetry that decides the direction: a rule that preloads when it
|
|
needn't costs context; a rule that quietly stops preloading costs the
|
|
behaviour it was written for. So a typo binds."""
|
|
from scribe.services.rulebooks import _valid_tier
|
|
|
|
assert _valid_tier("conditional") == "conditional"
|
|
assert _valid_tier("always_on") == "always_on"
|
|
assert _valid_tier("Conditional") == "always_on"
|
|
assert _valid_tier("") == "always_on"
|
|
assert _valid_tier("occasionally") == "always_on"
|