"""Real-Postgres tests for the rule-history write path (milestone 323 step 2). Every assertion here is about ORDERING or ABSENCE, which is why none of them can be made with mocks. What a version is for: it holds what the rule USED TO SAY. The edit destroys that, and nothing can recompute it — the rescoping of rule 79 had to be hand-copied into a task log to survive (#3237), which is not a process, it is a person remembering. Two of these pin a DELIBERATE ABSENCE. `note_versions` carries a 300-second interval gate and a 50-version cap, both defences against note autosave. Rules have no autosave, so here those guards would only ever discard a real edit. Nothing in the code says so on its own, and "make it consistent with the sibling" is a plausible-sounding change that would silently start dropping history — so the absence is a test, not a comment. """ import pytest import pytest_asyncio from sqlalchemy import select from scribe.models import async_session from scribe.models.rulebook import Rule, Rulebook from scribe.models.rule_version import RuleVersion from scribe.services import rule_versions as rv_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 = "rule_history_owner" @pytest_asyncio.fixture async def constraint(): """One rule carrying a check, with no history yet. CLEANED UP AT SETUP, NOT TEARDOWN, and that is forced. `update_rule` fires a detached `asyncio.create_task(upsert_rule_embedding(...))` that opens its own connection and UPDATEs the rule row. A teardown that deleted the rulebook would race it: the delete cascade-locks the rule the embedding task is writing, and Postgres kills one of them with a deadlock. Purging at setup instead runs on a fresh loop, after the previous test's loop closed and cancelled whatever it left in flight. """ async with async_session() as s: owner = await ensure_user(s, OWNER_USERNAME) uid = owner.id await s.commit() # A previous failed run would leave a book whose topic title collides. for book in (await s.execute( select(Rulebook).where(Rulebook.owner_user_id == uid) )).scalars().all(): await s.delete(book) await s.commit() book = await rulebooks_svc.create_rulebook(uid, "History fixtures") topic = await rulebooks_svc.create_topic(book.id, uid, "ci") rule = await rulebooks_svc.create_rule( topic.id, uid, "The runner has no bash", "Write every `run:` step in POSIX sh.", why="the image ships no bash", verify_with="read the workflow's shell setting", ) return {"uid": uid, "rule_id": rule.id} async def _versions(rule_id: int) -> list[RuleVersion]: async with async_session() as s: return list((await s.execute( select(RuleVersion).where(RuleVersion.rule_id == rule_id) .order_by(RuleVersion.id) )).scalars().all()) async def test_an_edit_leaves_the_text_it_replaced(constraint): """The headline. The version holds the OLD statement — the rule row already holds the new one, so a snapshot of the new text would record nothing that was not already there.""" await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], statement="Write every `run:` step in POSIX sh, and set shell: sh.", ) [version] = await _versions(constraint["rule_id"]) assert version.statement == "Write every `run:` step in POSIX sh." assert version.why == "the image ships no bash" async def test_the_actor_is_recorded(constraint): """`user_id` is who made the edit — the question an audit trail over a binding instruction is actually asked.""" await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], title="The runner ships no bash", ) [version] = await _versions(constraint["rule_id"]) assert version.user_id == constraint["uid"] async def test_an_edit_that_changes_no_text_writes_nothing(constraint): """Git semantics. `update_rule` is reached by callers that resend every field, so without the skip a form saved twice would file an identical snapshot and the history would stop reading as a list of edits.""" await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], statement="Write every `run:` step in POSIX sh.", title="The runner has no bash", ) assert await _versions(constraint["rule_id"]) == [] async def test_reordering_is_not_an_edit(constraint): """`order_index` goes through the same door but is not text. A run of "moved rule 4 above rule 3" snapshots would bury the edits somebody is looking for.""" await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], order_index=7, ) assert await _versions(constraint["rule_id"]) == [] async def test_the_snapshot_holds_the_CHECK_that_was_in_force(constraint): """The ordering trap. `update_rule` drops `verified_at` when `verify_with` changes, and rewriting the check is exactly the edit whose history matters most — so the snapshot has to be taken before the field loop, or it records the new check against the old wording.""" await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], verify_with="run `sh -c 'echo $0'` in a job step", ) [version] = await _versions(constraint["rule_id"]) assert version.verify_with == "read the workflow's shell setting" async with async_session() as s: rule = await s.get(Rule, constraint["rule_id"]) assert rule.verify_with == "run `sh -c 'echo $0'` in a job step" async def test_clearing_a_field_is_an_edit_worth_recording(constraint): """A rule that stops being a constraint loses its check entirely, and the version is then the only place the retired check exists. `clear` is a separate argument from the field loop — a snapshot wired to only one of the two would lose exactly this case.""" await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], clear=("verify_with",), ) [version] = await _versions(constraint["rule_id"]) assert version.verify_with == "read the workflow's shell setting" async with async_session() as s: rule = await s.get(Rule, constraint["rule_id"]) assert rule.verify_with is None async def test_rapid_successive_edits_are_all_kept(constraint): """DELIBERATE ABSENCE — no MIN_VERSION_INTERVAL_SECONDS. `note_versions` skips a snapshot taken within 300 seconds of the last one, because note autosave fires every 60 and would burn all 50 slots. Rules have no autosave: every one of these is a deliberate update_rule, and three edits in the same second are three edits. Porting the sibling's gate here would silently keep only the first. """ for n in (1, 2, 3): await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], statement=f"revision {n}", ) versions = await _versions(constraint["rule_id"]) assert [v.statement for v in versions] == [ "Write every `run:` step in POSIX sh.", "revision 1", "revision 2", ] async def test_the_service_carries_no_pruning_machinery(constraint): """DELIBERATE ABSENCE — no MAX_VERSIONS and no pruning DELETE. Asserted structurally because the alternative is 51 real edits to prove a negative. The cap exists in `note_versions` to bound autosave volume; a rule is edited a handful of times in its life, so a cap here could only ever discard the one edit somebody went looking for. """ assert not hasattr(rv_svc, "MAX_VERSIONS") assert not hasattr(rv_svc, "MIN_VERSION_INTERVAL_SECONDS") async def test_history_reads_newest_first(constraint): """The reader's question is "what did this just say?", so the most recent supersession has to be the first row.""" await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], statement="first change", ) await rulebooks_svc.update_rule( constraint["rule_id"], constraint["uid"], statement="second change", ) listed = await rv_svc.list_versions(constraint["rule_id"]) assert [v.statement for v in listed] == [ "first change", "Write every `run:` step in POSIX sh.", ]