"""Real-Postgres tests for moment proposals and judgments (milestone 458 step 7). What the step promises, against the real tables: - the pass reads only rules nobody has answered for, and any answer — a proposal, a mount, "no moment fits" — takes a rule off that list; - a proposal mounts nothing; a confirm mounts, a reject is kept, and a judged pair is never proposed again by either source; - an edit that takes a moment off a rule is a rejection the signal respects; - the signal counts SESSIONS, not opens, and asks once the bar is crossed; - only the owner can propose, judge, or feed the signal. """ import pytest import pytest_asyncio from sqlalchemy import select from scribe.models import async_session from scribe.models.rule_moment_judgment import RuleMomentJudgment from scribe.models.rulebook import Rulebook from scribe.models.user import User from scribe.services import rule_moment_judgments as judgments_svc from scribe.services import rulebooks as rulebooks_svc from tests.helpers import ensure_user pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")] OWNER_USERNAME = "moment_judgments_owner" STRANGER_USERNAME = "moment_judgments_stranger" async def _purge_books(username: str) -> None: """At SETUP, for the reason test_integration_rule_moments gives.""" async with async_session() as s: for user in (await s.execute( select(User).where(User.username == username) )).scalars().all(): for book in (await s.execute( select(Rulebook).where(Rulebook.owner_user_id == user.id) )).scalars().all(): await s.delete(book) await s.commit() @pytest_asyncio.fixture async def world(): for name in (OWNER_USERNAME, STRANGER_USERNAME): await _purge_books(name) async with async_session() as s: owner = await ensure_user(s, OWNER_USERNAME) stranger = await ensure_user(s, STRANGER_USERNAME) await s.commit() uid, sid = owner.id, stranger.id book = await rulebooks_svc.create_rulebook(uid, "Judgment fixtures") topic = await rulebooks_svc.create_topic(book.id, uid, "verification") done = await rulebooks_svc.create_rule( topic.id, uid, "Done means delivered and checked", "Work is done when it is delivered and its automated checks pass.", when_to_apply="about to tell the operator a piece of work is finished", ) style = await rulebooks_svc.create_rule( topic.id, uid, "Dates are written day first", "Every date shown to a person is written day, month, year.", when_to_apply="formatting a date for display", ) return {"uid": uid, "sid": sid, "done": done, "style": style} async def _row(rule_id: int, moment: str) -> RuleMomentJudgment | None: async with async_session() as s: return (await s.execute( select(RuleMomentJudgment).where( RuleMomentJudgment.rule_id == rule_id, RuleMomentJudgment.moment == moment, ) )).scalar_one_or_none() async def _unjudged_ids(uid: int) -> set[int]: return {r["id"] for r in (await judgments_svc.unjudged_rules(uid, limit=100))["rules"]} async def test_any_answer_takes_a_rule_off_the_pass(world): uid, done, style = world["uid"], world["done"], world["style"] assert {done.id, style.id} <= await _unjudged_ids(uid) out = await judgments_svc.propose(uid, [ {"rule_id": done.id, "moments": ["work.finish", "reply.report"], "why": "about finishing"}, {"rule_id": style.id, "none": "about how a date looks, whenever one is shown"}, ]) assert out["proposed"] == 2 and out["no_moment"] == 1 and not out["refused"] assert not ({done.id, style.id} & await _unjudged_ids(uid)) # A proposal mounted nothing. assert (await rulebooks_svc.list_rule_moments([done.id])).get(done.id) is None async def test_a_confirm_mounts_and_a_reject_is_never_proposed_again(world): uid, done = world["uid"], world["done"] await judgments_svc.propose(uid, [ {"rule_id": done.id, "moments": ["work.finish", "work.plan"], "why": "w"}, ]) pending = await judgments_svc.pending(uid, rule_id=done.id) assert [p["moment"] for p in pending["rules"][0]["proposals"]] == ["work.finish", "work.plan"] out = await judgments_svc.judge(uid, [ {"rule_id": done.id, "moment": "work.finish", "verdict": "confirm", "note": "yes"}, {"rule_id": done.id, "moment": "work.plan", "verdict": "reject", "note": "too early"}, ]) assert not out["refused"] assert (await rulebooks_svc.list_rule_moments([done.id]))[done.id] == ["work.finish"] confirmed = await _row(done.id, "work.finish") assert (confirmed.state, confirmed.source, confirmed.note) == ("confirmed", "pass", "yes") assert (await _row(done.id, "work.plan")).state == "rejected" assert (await judgments_svc.pending(uid))["total"] == 0 again = await judgments_svc.propose(uid, [ {"rule_id": done.id, "moments": ["work.plan", "work.finish"], "why": "w"}, ]) assert again["proposed"] == 0 assert {s["moment"] for s in again["skipped"]} == {"work.plan", "work.finish"} async def test_an_edit_that_unmounts_is_a_rejection_the_signal_respects(world): uid, done = world["uid"], world["done"] await rulebooks_svc.set_rule_moments(done.id, uid, ["work.deliver"]) await rulebooks_svc.set_rule_moments(done.id, uid, []) row = await _row(done.id, "work.deliver") assert (row.state, row.source) == ("rejected", "edit") for session in ("a", "b", "c", "d"): assert await judgments_svc.co_occurred( uid, done.id, ["work.deliver"], situation=session, ) == "" assert (await _row(done.id, "work.deliver")).evidence is None async def test_the_signal_counts_sessions_and_asks_once_the_bar_is_crossed(world): uid, done = world["uid"], world["done"] # Ubiquitous moments are not evidence; a repeat in one session counts once. assert await judgments_svc.co_occurred(uid, done.id, ["work.run", "work.change"], situation="s1") == "" assert await _row(done.id, "work.run") is None for situation in ("s1", "s1", "s2"): assert await judgments_svc.co_occurred(uid, done.id, ["work.verify"], situation=situation) == "" line = await judgments_svc.co_occurred(uid, done.id, ["work.verify"], situation="s3") assert f"#{done.id}" in line and "`work.verify`" in line and "3 distinct sessions" in line assert "judge_rule_moments" in line row = await _row(done.id, "work.verify") assert (row.state, row.source) == ("suggested", "signal") # Asked once; the cooldown keeps the next open quiet. assert await judgments_svc.co_occurred(uid, done.id, ["work.verify"], situation="s4") == "" # And it waits for the operator like any proposal. pending = await judgments_svc.pending(uid, rule_id=done.id) [proposal] = pending["rules"][0]["proposals"] assert proposal["source"] == "signal" and proposal["evidence"]["situations"] == 4 async def test_a_mount_overturns_no_moment_fits(world): uid, style = world["uid"], world["style"] await judgments_svc.propose(uid, [{"rule_id": style.id, "none": "about how a date looks"}]) assert (await _row(style.id, "")).state == "confirmed" await judgments_svc.judge(uid, [ {"rule_id": style.id, "moment": "reply.report", "verdict": "confirm", "note": "dates in reports"}, ]) assert (await _row(style.id, "")).state == "rejected" async def test_only_the_owner_can_propose_judge_or_feed_the_signal(world): sid, done = world["sid"], world["done"] out = await judgments_svc.propose(sid, [{"rule_id": done.id, "moments": ["work.finish"]}]) assert out["proposed"] == 0 and out["refused"] judged = await judgments_svc.judge(sid, [ {"rule_id": done.id, "moment": "work.finish", "verdict": "confirm"}, ]) assert judged["refused"] and not judged["judged"] for situation in ("a", "b", "c"): assert await judgments_svc.co_occurred(sid, done.id, ["work.verify"], situation=situation) == "" assert await _row(done.id, "work.verify") is None assert done.id not in await _unjudged_ids(sid) assert (await judgments_svc.pending(sid))["total"] == 0 async def test_the_open_resolves_acts_through_the_installs_mappings(world): """`opened_after` reads the acts as delivery does: a push is a deliver.""" uid, done = world["uid"], world["done"] acts = [ {"tool_name": "Bash", "tool_input": {"command": "git push origin dev"}}, {"tool_name": "Bash", "tool_input": {"command": "ls"}}, "not an act", ] out = await judgments_svc.opened_after(uid, done.id, acts, session_id="s1") assert "work.deliver" in out["moments"] assert "work.run" not in out["moments"] assert (await _row(done.id, "work.deliver")).source == "signal"