"""Real-Postgres tests for lesson → rule links (milestone 440 step 1, #4630). 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() # ── #4631: "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 == []