bvandeusenandClaude Opus 5.5
fa41e006a7
Eight more test files state their properties without issue, milestone and rule numbers or dated regressions. The version tests describe what folding the channel into the version breaks instead of the date it did; the surface registry and settings-defaults tests state that floors move rather than quoting the decision that made them move. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
362 lines
16 KiB
Python
362 lines
16 KiB
Python
"""Real-Postgres tests for lesson → rule links.
|
|
|
|
What a mock cannot show: that the CHECK holds the three states, that deleting
|
|
either end takes the link with it, that a rule the caller does not own cannot
|
|
be linked (and nothing is half-written when one id is bad), and that each side
|
|
reads only what its reader may see.
|
|
"""
|
|
import uuid
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.lesson_rule_link import LessonRuleLink
|
|
from scribe.models.note import Note
|
|
from scribe.models.project import Project
|
|
from scribe.models.rulebook import Rule
|
|
from scribe.services import lesson_rules as links_svc
|
|
from scribe.services import lessons as lessons_svc
|
|
from scribe.services import rulebooks as rulebooks_svc
|
|
from tests.helpers import ensure_user
|
|
|
|
pytestmark = [
|
|
pytest.mark.integration,
|
|
pytest.mark.usefixtures("_dispose_engine", "_no_embedding"),
|
|
]
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _no_reindex():
|
|
"""Rule writes detach an embedding refresh that outlives the test's loop."""
|
|
with patch("scribe.services.rulebooks._refresh_rule_embedding", MagicMock()):
|
|
yield
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def world():
|
|
"""An owner with a lesson and two rules; a stranger with a rule of their own."""
|
|
tag = uuid.uuid4().hex[:8]
|
|
async with async_session() as s:
|
|
owner = await ensure_user(s, f"lrl_owner_{tag}")
|
|
stranger = await ensure_user(s, f"lrl_stranger_{tag}")
|
|
mine = Project(user_id=owner.id, title="Mine")
|
|
theirs = Project(user_id=stranger.id, title="Theirs")
|
|
s.add_all([mine, theirs])
|
|
await s.flush()
|
|
ids = {"owner": owner.id, "stranger": stranger.id,
|
|
"mine": mine.id, "theirs": theirs.id}
|
|
await s.commit()
|
|
|
|
owner = ids["owner"]
|
|
r1 = await rulebooks_svc.create_project_rule(
|
|
ids["mine"], owner, "Read the job log first", "Before waiting longer.",
|
|
when_to_apply="a CI run has overrun its usual duration",
|
|
)
|
|
r2 = await rulebooks_svc.create_project_rule(
|
|
ids["mine"], owner, "Probe the system itself", "Not a proxy for it.",
|
|
when_to_apply="about to state what version is deployed",
|
|
)
|
|
foreign = await rulebooks_svc.create_project_rule(
|
|
ids["theirs"], ids["stranger"], "Their rule", "Not yours.",
|
|
when_to_apply="something only they do",
|
|
)
|
|
lesson = await lessons_svc.create_lesson(
|
|
owner, what="An overrun run usually failed early",
|
|
when_to_apply="a CI run is still in_progress far past its usual time",
|
|
insight="Read the log; the failure is often minutes old.",
|
|
project_id=ids["mine"],
|
|
)
|
|
ids.update(r1=r1.id, r2=r2.id, foreign=foreign.id, lesson=lesson.id)
|
|
return ids
|
|
|
|
|
|
async def _states(lesson_id: int) -> dict[int, str]:
|
|
async with async_session() as s:
|
|
rows = (await s.execute(
|
|
select(LessonRuleLink).where(LessonRuleLink.lesson_id == lesson_id)
|
|
)).scalars().all()
|
|
return {r.rule_id: r.state for r in rows}
|
|
|
|
|
|
async def test_naming_rules_confirms_them_and_both_sides_read_the_link(world):
|
|
owner = world["owner"]
|
|
await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"], world["r2"]])
|
|
|
|
assert await _states(world["lesson"]) == {world["r1"]: "confirmed", world["r2"]: "confirmed"}
|
|
rules = (await links_svc.rules_for_lessons(owner, [world["lesson"]]))[world["lesson"]]
|
|
assert {r["id"] for r in rules} == {world["r1"], world["r2"]}
|
|
lessons = await links_svc.lessons_for_rule(owner, world["r1"])
|
|
assert [(l["id"], l["state"]) for l in lessons] == [(world["lesson"], "confirmed")]
|
|
|
|
|
|
async def test_a_rule_left_out_of_the_set_is_rejected_not_forgotten(world):
|
|
owner = world["owner"]
|
|
await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"], world["r2"]])
|
|
await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"]])
|
|
|
|
assert await _states(world["lesson"]) == {world["r1"]: "confirmed", world["r2"]: "rejected"}
|
|
rules = (await links_svc.rules_for_lessons(owner, [world["lesson"]]))[world["lesson"]]
|
|
# Confirmed reads first; the rejection keeps its reason.
|
|
assert [r["state"] for r in rules] == ["confirmed", "rejected"]
|
|
assert rules[1]["note"] == links_svc._REMOVED_NOTE
|
|
|
|
|
|
async def test_judge_moves_a_pair_both_ways_and_refuses_a_nonsense_verdict(world):
|
|
owner = world["owner"]
|
|
out = await links_svc.judge_link(owner, world["lesson"], world["r1"], "reject", "different failure")
|
|
assert (out["state"], out["note"]) == ("rejected", "different failure")
|
|
assert out["judged_at"] is not None
|
|
out = await links_svc.judge_link(owner, world["lesson"], world["r1"], "confirm")
|
|
assert out["state"] == "confirmed"
|
|
with pytest.raises(ValueError):
|
|
await links_svc.judge_link(owner, world["lesson"], world["r1"], "maybe")
|
|
|
|
|
|
async def test_a_rule_the_caller_cannot_read_links_nothing_at_all(world):
|
|
"""All-or-nothing: one unreadable id refuses the whole set, so the readable
|
|
one beside it is not linked either."""
|
|
with pytest.raises(ValueError):
|
|
await links_svc.set_lesson_rules(
|
|
world["owner"], world["lesson"], [world["r1"], world["foreign"]],
|
|
)
|
|
assert await _states(world["lesson"]) == {}
|
|
|
|
|
|
async def test_a_stranger_cannot_link_someone_elses_lesson(world):
|
|
with pytest.raises((ValueError, PermissionError)):
|
|
await links_svc.set_lesson_rules(world["stranger"], world["lesson"], [world["foreign"]])
|
|
assert await _states(world["lesson"]) == {}
|
|
|
|
|
|
async def test_deleting_either_end_takes_the_link_with_it(world):
|
|
owner = world["owner"]
|
|
await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"], world["r2"]])
|
|
async with async_session() as s:
|
|
await s.execute(delete(Rule).where(Rule.id == world["r1"]))
|
|
await s.commit()
|
|
assert await _states(world["lesson"]) == {world["r2"]: "confirmed"}
|
|
async with async_session() as s:
|
|
await s.execute(delete(Note).where(Note.id == world["lesson"]))
|
|
await s.commit()
|
|
assert await _states(world["lesson"]) == {}
|
|
|
|
|
|
async def test_the_check_holds_the_three_states(world):
|
|
async with async_session() as s:
|
|
s.add(LessonRuleLink(lesson_id=world["lesson"], rule_id=world["r1"], state="maybe"))
|
|
with pytest.raises(IntegrityError):
|
|
await s.commit()
|
|
|
|
|
|
async def test_one_row_per_pair(world):
|
|
async with async_session() as s:
|
|
s.add_all([
|
|
LessonRuleLink(lesson_id=world["lesson"], rule_id=world["r1"], state="suggested"),
|
|
LessonRuleLink(lesson_id=world["lesson"], rule_id=world["r1"], state="confirmed"),
|
|
])
|
|
with pytest.raises(IntegrityError):
|
|
await s.commit()
|
|
|
|
|
|
# ── "no rule fits" and the unjudged list ────────────────────────────────────
|
|
|
|
|
|
async def _judgment(lesson_id: int) -> dict:
|
|
return (await links_svc.judgments_for_lessons([lesson_id]))[lesson_id]
|
|
|
|
|
|
async def _unjudged_ids(user_id: int) -> set[int]:
|
|
items, _total = await links_svc.list_unjudged(user_id, limit=100)
|
|
return {i["id"] for i in items}
|
|
|
|
|
|
async def test_a_new_lesson_is_unjudged_until_it_is_answered_either_way(world):
|
|
owner, lesson = world["owner"], world["lesson"]
|
|
assert (await _judgment(lesson))["rule_judgment"] == "unjudged"
|
|
assert lesson in await _unjudged_ids(owner)
|
|
|
|
await links_svc.set_no_rule(owner, lesson, "specific to this CI host")
|
|
judged = await _judgment(lesson)
|
|
assert judged["rule_judgment"] == "no_rule"
|
|
assert judged["no_rule"]["why"] == "specific to this CI host"
|
|
assert lesson not in await _unjudged_ids(owner)
|
|
|
|
# Naming a rule overturns the answer — both cannot stand.
|
|
await links_svc.set_lesson_rules(owner, lesson, [world["r1"]])
|
|
judged = await _judgment(lesson)
|
|
assert judged == {"rule_judgment": "linked"}
|
|
assert lesson not in await _unjudged_ids(owner)
|
|
|
|
|
|
async def test_no_rule_rejects_a_confirmed_link_with_its_reason(world):
|
|
owner, lesson = world["owner"], world["lesson"]
|
|
await links_svc.set_lesson_rules(owner, lesson, [world["r1"]])
|
|
await links_svc.set_no_rule(owner, lesson, "the log was not the issue")
|
|
|
|
assert await _states(lesson) == {world["r1"]: "rejected"}
|
|
rules = (await links_svc.rules_for_lessons(owner, [lesson]))[lesson]
|
|
assert rules[0]["note"].endswith("the log was not the issue")
|
|
assert (await _judgment(lesson))["rule_judgment"] == "no_rule"
|
|
|
|
|
|
async def test_a_confirmation_by_judgment_also_clears_no_rule(world):
|
|
owner, lesson = world["owner"], world["lesson"]
|
|
await links_svc.set_no_rule(owner, lesson, "nothing yet")
|
|
await links_svc.judge_link(owner, lesson, world["r2"], "confirm")
|
|
assert (await _judgment(lesson))["rule_judgment"] == "linked"
|
|
|
|
|
|
async def test_a_rejection_alone_leaves_the_lesson_unjudged(world):
|
|
""""Not that rule" says nothing about whether another one fits."""
|
|
owner, lesson = world["owner"], world["lesson"]
|
|
await links_svc.judge_link(owner, lesson, world["r1"], "reject", "different failure")
|
|
assert (await _judgment(lesson))["rule_judgment"] == "unjudged"
|
|
assert lesson in await _unjudged_ids(owner)
|
|
|
|
|
|
async def test_no_rule_needs_a_reason_and_a_writer(world):
|
|
with pytest.raises(ValueError):
|
|
await links_svc.set_no_rule(world["owner"], world["lesson"], " ")
|
|
with pytest.raises((ValueError, PermissionError)):
|
|
await links_svc.set_no_rule(world["stranger"], world["lesson"], "mine now")
|
|
assert (await _judgment(world["lesson"]))["rule_judgment"] == "unjudged"
|
|
|
|
|
|
async def test_the_unjudged_list_is_scoped_to_the_reader(world):
|
|
assert world["lesson"] not in await _unjudged_ids(world["stranger"])
|
|
|
|
|
|
async def test_deleting_the_lesson_takes_its_answer_with_it(world):
|
|
from scribe.models.lesson_rule_link import LessonNoRule
|
|
|
|
await links_svc.set_no_rule(world["owner"], world["lesson"], "alone")
|
|
async with async_session() as s:
|
|
await s.execute(delete(Note).where(Note.id == world["lesson"]))
|
|
await s.commit()
|
|
left = (await s.execute(
|
|
select(LessonNoRule).where(LessonNoRule.lesson_id == world["lesson"])
|
|
)).scalars().all()
|
|
assert left == []
|
|
|
|
|
|
# ── soft links from co-surfacing ────────────────────────────────────────────
|
|
|
|
|
|
async def _link(lesson_id: int, rule_id: int):
|
|
async with async_session() as s:
|
|
return (await s.execute(
|
|
select(LessonRuleLink).where(
|
|
LessonRuleLink.lesson_id == lesson_id, LessonRuleLink.rule_id == rule_id,
|
|
)
|
|
)).scalar_one_or_none()
|
|
|
|
|
|
async def _arrive(world, prompt: str, project: int | None = None, rules=None) -> str:
|
|
return await links_svc.co_surfaced(
|
|
world["owner"], [world["lesson"]], rules or [world["r1"]],
|
|
arm="p", situation=prompt, project_id=project,
|
|
)
|
|
|
|
|
|
async def test_distinct_situations_reach_the_bar_and_the_proposal_fires_once(world):
|
|
assert await _arrive(world, "the CI run is still going after twenty minutes") == ""
|
|
assert await _arrive(world, "deploy job hanging on the image build step") == ""
|
|
line = await _arrive(world, "integration lane overran its usual duration", world["mine"])
|
|
assert f"judge_lesson_link({world['lesson']}, {world['r1']}" in line
|
|
|
|
link = await _link(world["lesson"], world["r1"])
|
|
assert link.state == "suggested"
|
|
assert links_svc.evidence_summary(link.evidence)["situations"] == 3
|
|
# Within the cooldown a fourth situation is counted but not asked again.
|
|
assert await _arrive(world, "pipeline stuck waiting on a runner for ages") == ""
|
|
link = await _link(world["lesson"], world["r1"])
|
|
assert links_svc.evidence_summary(link.evidence)["situations"] == 4
|
|
|
|
|
|
async def test_the_same_prompt_repeated_never_reaches_the_bar(world):
|
|
for _ in range(5):
|
|
assert await _arrive(world, "is the CI run still going") == ""
|
|
ev = links_svc.evidence_summary((await _link(world["lesson"], world["r1"])).evidence)
|
|
assert ev == {"situations": 1, "projects": 0, "co_surfaced": 5}
|
|
|
|
|
|
async def test_a_judged_pair_gathers_nothing_and_is_never_proposed(world):
|
|
owner = world["owner"]
|
|
await links_svc.judge_link(owner, world["lesson"], world["r1"], "reject", "different failure")
|
|
await links_svc.judge_link(owner, world["lesson"], world["r2"], "confirm")
|
|
for prompt in ("first situation here", "second situation there", "third situation elsewhere"):
|
|
assert await _arrive(world, prompt, rules=[world["r1"], world["r2"]]) == ""
|
|
assert (await _link(world["lesson"], world["r1"])).evidence is None
|
|
assert (await _link(world["lesson"], world["r2"])).evidence is None
|
|
|
|
|
|
async def test_a_rule_the_reader_does_not_own_is_not_paired(world):
|
|
for prompt in ("first situation here", "second situation there", "third situation elsewhere"):
|
|
assert await _arrive(world, prompt, rules=[world["foreign"]]) == ""
|
|
assert await _link(world["lesson"], world["foreign"]) is None
|
|
|
|
|
|
async def test_a_suggestion_shows_its_evidence_and_leaves_the_lesson_unjudged(world):
|
|
await _arrive(world, "the CI run is still going after twenty minutes", world["mine"])
|
|
rules = (await links_svc.rules_for_lessons(world["owner"], [world["lesson"]]))[world["lesson"]]
|
|
assert rules[0]["state"] == "suggested"
|
|
assert rules[0]["evidence"] == {"situations": 1, "projects": 1, "co_surfaced": 1}
|
|
assert (await _judgment(world["lesson"]))["rule_judgment"] == "unjudged"
|
|
|
|
|
|
async def test_confirming_a_suggestion_keeps_what_it_rested_on(world):
|
|
for prompt in ("first situation here", "second situation there", "third situation elsewhere"):
|
|
await _arrive(world, prompt)
|
|
out = await links_svc.judge_link(world["owner"], world["lesson"], world["r1"], "confirm", "same failure class")
|
|
assert out["state"] == "confirmed"
|
|
assert len(out["evidence"]["situations"]) == 3
|
|
|
|
|
|
# ── which links can carry a rule ────────────────────────────────────────────
|
|
|
|
|
|
async def test_only_a_confirmed_link_can_carry_its_rule(world):
|
|
"""A suggested link never expands: it would manufacture the co-arrival
|
|
that co-surfacing counts, and prove itself. A rejected one obviously not."""
|
|
owner, lesson = world["owner"], world["lesson"]
|
|
await links_svc.co_surfaced(owner, [lesson], [world["r1"]], arm="p", situation="one ask here")
|
|
await links_svc.judge_link(owner, lesson, world["r2"], "reject", "not this")
|
|
assert await links_svc.confirmed_lessons(owner) == set()
|
|
assert await links_svc.confirmed_rules_in_scope(owner, [lesson], world["mine"]) == {}
|
|
|
|
await links_svc.judge_link(owner, lesson, world["r1"], "confirm", "same failure class")
|
|
assert lesson in await links_svc.confirmed_lessons(owner)
|
|
rules = await links_svc.confirmed_rules_in_scope(owner, [lesson], world["mine"])
|
|
assert [r.id for r in rules[lesson]] == [world["r1"]]
|
|
|
|
|
|
async def test_a_project_rule_stays_in_its_project_even_through_a_lesson(world):
|
|
"""Rule scope, kept on the side door: the lesson surfaces
|
|
everywhere, its project rule only in its project."""
|
|
owner, lesson = world["owner"], world["lesson"]
|
|
await links_svc.set_lesson_rules(owner, lesson, [world["r1"]])
|
|
assert await links_svc.confirmed_rules_in_scope(owner, [lesson], world["mine"])
|
|
assert await links_svc.confirmed_rules_in_scope(owner, [lesson], world["theirs"]) == {}
|
|
assert await links_svc.confirmed_rules_in_scope(owner, [lesson], None) == {}
|
|
|
|
|
|
async def test_a_stranger_reaches_no_rule_through_someone_elses_link(world):
|
|
await links_svc.set_lesson_rules(world["owner"], world["lesson"], [world["r1"]])
|
|
assert await links_svc.confirmed_lessons(world["stranger"]) == set()
|
|
|
|
|
|
# ── which lessons count toward convergence ──────────────────────────────────
|
|
|
|
|
|
async def test_only_answered_lessons_are_no_rule_lessons(world):
|
|
owner = world["owner"]
|
|
other = await lessons_svc.create_lesson(
|
|
owner, what="Another overrun", when_to_apply="a deploy job overran",
|
|
)
|
|
await links_svc.set_no_rule(owner, world["lesson"], "specific to this host")
|
|
assert await links_svc._no_rule_ids([world["lesson"], other.id]) == {world["lesson"]}
|