Files
FabledScribe/tests/test_services_rulebooks.py
T
bvandeusenandClaude Opus 5 0e10f6bb8a
CI & Build / Python lint (push) Failing after 3s
CI & Build / Plugin hooks (push) Failing after 12s
CI & Build / integration (push) Failing after 27s
CI & Build / TypeScript typecheck (push) Failing after 35s
CI & Build / Python tests (push) Failing after 37s
CI & Build / Build & push image (push) Skipped
feat(rules)!: retire the always-on tier — every rule arrives by retrieval (#394)
Milestone 394, steps 5-8. Operator: "remove the always on rule functionality
as the goal was to not have it at all since it didn't seem to work as
expected."

Unconditional preload had three failures the retrieval arms do not. It could
not be MEASURED — a resident rule is in the context whether or not it
mattered, so nothing distinguished "this governed the act" from "this was
scenery", and it was the one surface structurally exempt from the scoreboard
judging every other. It was SUMMARISED AWAY by compaction while the session
went on believing it held the rules. And it CROWDED OUT the few rules that
applied with the thirty that did not.

WHAT GOES

Schema (0100): rules.tier + ck_rules_tier, rule_versions.tier,
rulebooks.always_on, and project_rulebook_exclusions — a table recording a
project's opt-out of something that no longer binds it unasked.

Tools: list_always_on_rules, exclude_always_on_rulebook,
include_always_on_rulebook. Service: the same three plus rules_etag_for,
_valid_tier and the whole etag family. The SessionStart preload and the
write-path staleness arm go with them: nothing is resident, so nothing can
have drifted since a session loaded it.

THREE CALLS WORTH REVIEWING

enter_project got NARROWER, not wider. Its filter was `always_on OR
area-tagged`; dropping the tier arm leaves the deterministic half, so a
project with no canonical-tagged Systems gets no bulk rules and reaches them
by retrieval instead. Dropping the whole clause would have made that payload
bigger than the preload this milestone deletes.

Backups import tolerantly. A pre-394 archive carries tier, always_on and the
retired inception choice; none is read, and the exclusion key is DROPPED
rather than remapped, because restoring it would write data that
validate_inception now rejects as unknown.

The migration is irreversible in the way that matters and says so: downgrade
recreates the columns at their defaults and cannot restore which rules were
always-on. A value invented to fill a hole is not a measurement.

THE INSTRUCTION SURFACES SAY THE HARDER THING

Deleting "call list_always_on_rules()" is easy; replacing it is not, because
the new model asks a session to trust something it cannot see. All three
surfaces now say a session holds nothing, that rules arrive when work matches
them, and — the half that got dangerous — that "no rule arrived" means
"nothing matched", never "there is no rule". Under residency an empty session
was rare and suspicious; it is now the ordinary state of most turns, so
reading it as permission is wrong on nearly every turn rather than
occasionally. That is #3720's defect at session scale.

test_instruction_surfaces_agree is repointed rather than retired: its two
halves collapsed into one instruction, and it gains a guard that every
surface states what absence means. _INSTRUCTIONS is back at 1999/2000 —
the inception clause paid for the longer HOW line.

UI (rule 27, and the opportunity step 8 named)

The tier selector is gone, and what replaces it is the point: `when_to_apply`
is now the field that decides whether a rule is ever seen, so the editor
marks it required, warns while it is empty, and both rule lists badge a
trigger-less rule "never surfaces". A rule without one is not quiet, it is
unreachable.

TESTS

Two files deleted outright — test_rules_etag.py and test_inception_rules.py
tested subsystems that no longer exist. Elsewhere obsolete cases were removed
and the rest repointed. One deserves naming: the wiring test asserted the act
arms pass no `tier`, which had become an assertion that could not fail. It is
repointed onto `kind`, which does still exist and where the same claim is
live — a preference must reach a write exactly as a rule does.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011cPyzNnegXHr5iRMzzy5KJ
2026-09-11 16:22:17 -04:00

485 lines
20 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."""
if True:
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
def _no_edges():
"""Silence the three post-query lookups get_applicable_rules now makes.
They are separate service functions with their own coverage (and the real
wiring is proven against Postgres in test_integration_rule_surfacing), so
stubbing them here keeps each of these tests about the one projection it
was written to check — rather than about the order a mocked session's
execute() calls happen to arrive in.
"""
return (
patch("scribe.services.rulebooks.co_surfaced_partners",
AsyncMock(return_value=[])),
patch("scribe.services.rulebooks.list_rule_relations",
AsyncMock(return_value={})),
patch("scribe.services.rulebooks.list_rule_systems",
AsyncMock(return_value={})),
)
@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,
# project-areas_q (milestone 307), rules_q, proj_rules_q
mock_session.execute = AsyncMock(side_effect=[
sub_result, _empty(), _empty(), _empty(), rules_result, _empty(),
])
_p1, _p2, _p3 = _no_edges()
with patch("scribe.services.rulebooks.async_session") as mock_cls, _p1, _p2, _p3:
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(), _empty(), rules_result, _empty(),
])
_p1, _p2, _p3 = _no_edges()
with patch("scribe.services.rulebooks.async_session") as mock_cls, _p1, _p2, _p3:
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(), _empty(), proj_rules_result,
])
_p1, _p2, _p3 = _no_edges()
with patch("scribe.services.rulebooks.async_session") as mock_cls, _p1, _p2, _p3:
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=[
# sub_q, suppressed_rules_q, suppressed_topics_q, project-areas_q,
# rules_q, proj_rules_q
_empty(), suppressed_rules_result, suppressed_topics_result,
_empty(), _empty(), _empty(),
])
_p1, _p2, _p3 = _no_edges()
with patch("scribe.services.rulebooks.async_session") as mock_cls, _p1, _p2, _p3:
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 (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"
# 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
# ── verify_with / expires_when (milestone 312) ──────────────────────────
def test_a_rule_with_no_check_says_nothing_about_verification():
"""The empty case is the COMMON case, and it must stay silent.
Most rules are decisions: they have no truth value and there is nothing to
go and check. If a brief carried `last_verified` for those too, the signal
would be worthless — every rule would look like something someone ought to
be verifying, and the handful that genuinely rot would stop standing out.
"""
from scribe.services.rulebooks import last_verified_label, rule_brief
rule = fake_rule()
assert last_verified_label(rule) is None
assert "last_verified" not in rule_brief(rule)
def test_an_unverified_constraint_reads_never_rather_than_null():
"""#2483 again: a null key reads as a capability going unused. "never" is
a different and much stronger claim — this rule asserts a fact about
someone else's software and nobody has ever confirmed it."""
from scribe.services.rulebooks import last_verified_label, rule_brief
rule = fake_rule(verify_with="cat CI-runner/renovate/config.js")
assert last_verified_label(rule) == "never"
assert rule_brief(rule)["last_verified"] == "never"
def test_a_verified_constraint_reports_the_date_it_was_checked():
"""A date, not a stamp — the question is "how old is this", the same call
rule_brief makes for updated_at."""
from scribe.services.rulebooks import last_verified_label
rule = fake_rule(
verify_with="cat CI-runner/renovate/config.js",
verified_at=datetime(2026, 8, 27, 11, 46, tzinfo=timezone.utc),
)
assert last_verified_label(rule) == "2026-08-27"
def test_the_check_text_itself_never_enters_a_listing():
"""A listing says WHICH rules can rot, not how to test them. The check can
be a long command; multiplied across an always-on set it is the same bloat
`why` and `how_to_apply` are kept out of a brief to avoid."""
from scribe.services.rulebooks import rule_brief
out = rule_brief(fake_rule(
verify_with="a very long command " * 20,
expires_when="the runner learns a new shell",
))
assert "verify_with" not in out
assert "expires_when" not in out
# ── the sweep's row shape (milestone 312 step 3) ────────────────────────
def test_a_sweep_row_carries_the_check_in_full():
"""The OPPOSITE call from rule_brief, and deliberately so.
A listing omits the depth because nobody reading it wants to act on one
rule. A sweep row exists to be acted on — the reader is about to go and
run the check — so the text is the payload's point, not its bloat.
"""
from scribe.services.rulebooks import verification_row
row = verification_row(fake_rule(
verify_with="cat CI-runner/renovate/config.js",
expires_when="dependencyDashboardApproval is turned off",
when_to_apply="when a dependency bump is in play",
))
assert row["verify_with"] == "cat CI-runner/renovate/config.js"
assert row["expires_when"] == "dependencyDashboardApproval is turned off"
assert row["when_to_apply"] == "when a dependency bump is in play"
def test_never_verified_reports_no_day_count_rather_than_zero():
""""Never" is not "0 days ago" — the second reads as freshly checked.
Getting this wrong would invert the row's meaning for exactly the rules
that most need attention.
"""
from scribe.services.rulebooks import verification_row
row = verification_row(fake_rule(verify_with="read the workflow"))
assert row["last_verified"] == "never"
assert row["days_since_verified"] is None
def test_a_verified_row_counts_the_days():
from datetime import timedelta
from scribe.services.rulebooks import verification_row
row = verification_row(fake_rule(
verify_with="read the workflow",
verified_at=datetime.now(timezone.utc) - timedelta(days=74, hours=1),
))
assert row["days_since_verified"] == 74