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>
508 lines
21 KiB
Python
508 lines
21 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
|
|
|
|
|
|
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 + 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"
|
|
|
|
|
|
# ── 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"
|
|
assert row["tier"] == "always_on"
|
|
|
|
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_unrecognised_tier_filter_raises_rather_than_narrowing():
|
|
"""_valid_tier's silent always_on fallback is right for a WRITE — a typo
|
|
should leave a rule binding. It is wrong for a FILTER, where the same
|
|
fallback would quietly answer a different question than the one asked and
|
|
return a short list that looks like good news."""
|
|
from scribe.services.rulebooks import rules_due_for_verification
|
|
|
|
with pytest.raises(ValueError, match="tier must be one of"):
|
|
await rules_due_for_verification(7, tier="occasionally")
|