"""Real-Postgres round trip for rule_versions (milestone 323 step 1). Two things here cannot be shown with mocks, and both are the kind that fail QUIETLY — a restore reports success and hands back history that is wrong. 1. **`rule_id` is remapped, not copied.** It is an id in the SOURCE database. A restore that writes it straight through succeeds and attaches every snapshot to whatever rule happens to hold that number here — an edit history filed against the wrong binding instruction, which is worse than no history at all. This is #3182's `arose_from_id` trap on a new table. 2. **A null actor does not drop the row.** `user_id` is SET NULL precisely so history outlives the account that wrote it, so the restore deliberately diverges from its `NoteVersion` sibling, which skips a version it cannot map to a user. Nothing about the code says which of the two shapes is intended; without this test, "make it match the sibling" reads as a tidy-up and silently deletes the record the FK was chosen to preserve. These drive the REAL `restore_full_backup`. A test that re-derived the remap would agree with whatever the product does, including nothing. """ import pytest import pytest_asyncio from sqlalchemy import select from scribe.models import async_session from scribe.models.rule_version import RuleVersion from scribe.models.rulebook import Rule, Rulebook, RulebookTopic from scribe.models.user import User from scribe.services import backup from tests.helpers import ensure_user pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")] OWNER_USERNAME = "rule_version_roundtrip_owner" RESTORED_USERNAME = "rule_version_roundtrip_restored" async def _purge_books(username: str) -> None: """user -> rulebook -> topic -> rule -> rule_version is ON DELETE CASCADE the whole way down, and no ORM relationships are configured, so dropping the books clears every row this file made under them.""" async with async_session() as s: users = (await s.execute( select(User).where(User.username == username) )).scalars().all() for user in users: books = (await s.execute( select(Rulebook).where(Rulebook.owner_user_id == user.id) )).scalars().all() for book in books: await s.delete(book) await s.commit() async def _purge_restored() -> None: """The restored user is minted by the restore itself, so it goes too.""" await _purge_books(RESTORED_USERNAME) async with async_session() as s: for user in (await s.execute( select(User).where(User.username == RESTORED_USERNAME) )).scalars().all(): await s.delete(user) await s.commit() @pytest_asyncio.fixture(autouse=True) async def _no_leftovers(): """The usernames are fixed, so a previous failed run would leave rows that make the fixtures below pick the wrong user — or hit `.one()` with two. SETUP ONLY, and that is not a stylistic choice. `_dispose_engine` is a usefixtures entry, so it sets up AFTER this autouse one and therefore tears down BEFORE it. Any database call here after a `yield` would open a fresh pooled connection that the closing loop then orphans, and the next test to touch Postgres dies on "Future attached to a different loop" — including tests in other files. Cleanup belongs in the fixtures below, whose teardowns run while the engine is still live. """ await _purge_restored() await _purge_books(OWNER_USERNAME) @pytest_asyncio.fixture async def source(): """One rule with two snapshots: one written by a user who still exists, one whose actor is already gone. Both are needed. With only the attributed version, dropping unmappable rows would pass; with only the orphaned one, so would dropping the actor from every row. """ async with async_session() as s: owner = await ensure_user(s, OWNER_USERNAME) uid = owner.id await s.commit() async with async_session() as s: book = Rulebook(owner_user_id=uid, title="Environment facts") s.add(book) await s.flush() topic = RulebookTopic(rulebook_id=book.id, title="ci") s.add(topic) await s.flush() rule = Rule( topic_id=topic.id, title="The runner has no bash", statement="Write every `run:` step in POSIX sh.", verify_with="read the workflow's shell setting", ) s.add(rule) await s.flush() s.add_all([ RuleVersion( rule_id=rule.id, user_id=uid, title="The runner has no bash", statement="Use sh.", why="the image ships no bash", verify_with="read the workflow's shell setting", tier="always_on", ), # The actor is already gone — what SET NULL leaves behind. RuleVersion( rule_id=rule.id, user_id=None, title="The runner has no bash", statement="Use POSIX sh in run steps.", tier="always_on", ), ]) await s.commit() book_id, rule_id = book.id, rule.id async with async_session() as s: user_rows = backup._user_rows( [(await s.execute(select(User).where(User.id == uid))).scalars().one()] ) book_rows = backup._rulebook_rows( [(await s.execute(select(Rulebook).where(Rulebook.id == book_id))) .scalars().one()] ) topic_rows = backup._topic_rows( (await s.execute( select(RulebookTopic).where(RulebookTopic.rulebook_id == book_id) )).scalars().all() ) rule_rows = backup._rule_rows( [(await s.execute(select(Rule).where(Rule.id == rule_id))).scalars().one()] ) version_rows = backup._rule_version_rows( (await s.execute( select(RuleVersion).where(RuleVersion.rule_id == rule_id) .order_by(RuleVersion.id) )).scalars().all() ) # The restore mints a NEW user from the payload, so the restored corpus is # separate from the source one — which is what makes the id assertion # below able to fail. user_rows[0]["username"] = RESTORED_USERNAME yield { "payload": { "version": backup.BACKUP_VERSION, "users": user_rows, "rulebooks": book_rows, "rulebook_topics": topic_rows, "rules": rule_rows, "rule_versions": version_rows, }, "source_rule_id": rule_id, "source_user_id": uid, } async with async_session() as s: book = await s.get(Rulebook, book_id) if book is not None: await s.delete(book) await s.commit() @pytest_asyncio.fixture async def restored(source): """Runs the real restore, then hands back the new rows. The restore mints a NEW user from the payload, so the restored corpus is entirely separate from the source one — which is what makes the id assertions below able to fail. """ await backup.restore_full_backup(source["payload"]) async with async_session() as s: user = (await s.execute( select(User).where(User.username == RESTORED_USERNAME) )).scalars().first() assert user is not None, "the payload's user was not restored" book = (await s.execute( select(Rulebook).where(Rulebook.owner_user_id == user.id) )).scalars().one() topic = (await s.execute( select(RulebookTopic).where(RulebookTopic.rulebook_id == book.id) )).scalars().one() rule = (await s.execute( select(Rule).where(Rule.topic_id == topic.id) )).scalars().one() versions = (await s.execute( select(RuleVersion).where(RuleVersion.rule_id == rule.id) .order_by(RuleVersion.id) )).scalars().all() yield {"user": user, "rule": rule, "versions": versions, "source": source} # Here rather than in the autouse fixture: this teardown still runs while # the engine is live. See _no_leftovers. await _purge_restored() async def test_both_snapshots_come_back(restored): """The count first: everything below reads the same on an empty list, so without this a restore that silently dropped both would look like a pass on the shape assertions.""" assert len(restored["versions"]) == 2 async def test_the_history_attaches_to_the_RESTORED_rule(restored): """#3182's trap. The source rule still exists and holds a different id, so a straight-through copy would file this history against it — or against whatever unrelated rule owns that number.""" new_rule_id = restored["rule"].id source_rule_id = restored["source"]["source_rule_id"] assert new_rule_id != source_rule_id, ( "the restore reused the source id, so this test cannot tell a remap " "from a copy — the fixture is not proving what it claims" ) assert {v.rule_id for v in restored["versions"]} == {new_rule_id} async def test_the_actor_is_remapped_to_the_restored_user(restored): """`user_id` is an id in the source database too — the same trap, on the column that answers "who changed this".""" attributed = [v for v in restored["versions"] if v.user_id is not None] assert len(attributed) == 1 assert attributed[0].user_id == restored["user"].id assert attributed[0].user_id != restored["source"]["source_user_id"] async def test_a_snapshot_with_no_actor_survives(restored): """The deliberate divergence from NoteVersion. `user_id` is SET NULL so that history outlives the account that wrote it; skipping the row on an unmappable user would throw away exactly what the FK preserves.""" orphaned = [v for v in restored["versions"] if v.user_id is None] assert len(orphaned) == 1, ( "the version whose actor was already gone did not come back. A rule's " "history is the only record of what it used to say — losing it " "because nobody can be credited is the wrong trade." ) assert orphaned[0].statement == "Use POSIX sh in run steps." async def test_the_text_survives(restored): """The whole point of the table: what the rule USED to say. A restore that kept the rows and lost their wording would preserve a changelog of empty entries.""" by_statement = {v.statement: v for v in restored["versions"]} assert set(by_statement) == {"Use sh.", "Use POSIX sh in run steps."} assert by_statement["Use sh."].why == "the image ships no bash" assert by_statement["Use sh."].verify_with == ( "read the workflow's shell setting" ) assert by_statement["Use sh."].tier == "always_on"