"""A staging refresh keeps every row a person wrote on the site (R2 task T7). The record that motivated it: Adam's three comments of 11 Sep on `todo:9`, `todo:11` and `todo:13` existed only on staging, and a restore of production's dump over staging would have replaced `discussion_comments` wholesale. Nothing would have recovered them. These tests run the export and apply halves on an in-memory SQLite: `apply --check` writes nothing and lists every client-written table; a comment whose id production has since used is put back under a fresh id; a ranking changed on staging beats production's older row; an author whose account is gone keeps the typed name with an empty link; a second apply changes nothing. """ from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import create_engine, event, func, select from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app import cli_refresh as R from app.models import ( Account, Base, ClientTodo, ClientUpload, DiscussionComment, DiscussionItem, FeaturePriority, OwnerItemState, QuoteRequest, QuoteSelection, ) from app.services.discussion import CLIENT_WRITTEN_TABLES TABLES = [Account.__table__] + [Base.metadata.tables[n] for n in CLIENT_WRITTEN_TABLES] T0 = datetime(2026, 9, 11, 1, 0, tzinfo=UTC) def fresh(): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) @event.listens_for(engine, "connect") def _fk(conn, _): conn.execute("PRAGMA foreign_keys=ON") Base.metadata.create_all(engine, tables=TABLES) return sessionmaker(bind=engine, expire_on_commit=False, future=True)() def seed_people(db, *usernames: str) -> dict[str, int]: for u in usernames: db.add(Account(username=u, display_name=u.title(), email=f"{u}@example.com")) db.commit() return {a.username: a.id for a in db.scalars(select(Account))} def seed_staging(db): """What staging held: the seeded todo, Adam's comments, his ranking, rian's decision.""" ids = seed_people(db, "rian", "adam", "mark") db.add(ClientTodo(id=9, owner="adam", title="Airport write-ups", status="done", completed_by="Adam", completed_by_id=ids["adam"], completed_at=T0, sort=90)) db.add(DiscussionItem(id=3, title="Awards rules", sort_order=3)) db.add_all([ DiscussionComment(id=214, feature_key="todo:13", author="Adam", author_id=ids["adam"], body="videos sent", created_at=T0, updated_at=T0), DiscussionComment(id=215, feature_key="todo:11", author="Adam", author_id=ids["adam"], body="categories by the 13th", created_at=T0, updated_at=T0), DiscussionComment(id=216, item_id=3, author="Mark", author_id=ids["mark"], body="on the card", created_at=T0, updated_at=T0), ]) db.add(FeaturePriority(id=1, feature_key="price-tracker", priority="next", author="Adam", author_id=ids["adam"], created_at=T0, updated_at=T0 + timedelta(days=1))) db.add(QuoteSelection(id=1, item_key="guides", included=True, author="Adam", author_id=ids["adam"], created_at=T0, updated_at=T0)) db.add(QuoteRequest(id=1, author="Adam", author_id=ids["adam"], items="[]", total_usd=1500, created_at=T0)) db.add(ClientUpload(id=1, todo_id=9, original_name="lhr.docx", stored_name="abc.docx", content_type="x", kind="document", bytes=3, uploaded_by="Adam", uploaded_by_id=ids["adam"], created_at=T0)) db.add(OwnerItemState(item_id="decide-x", status="done", decision="yes", acted_by="rian", acted_by_id=ids["rian"], acted_at=T0, updated_at=T0)) db.commit() return ids def seed_production(db): """What the restored dump holds: rian and adam only (mark's account is gone, and adam's id differs), production's own comment under id 214, an older ranking, the same todo still open.""" ids = seed_people(db, "adam", "rian") # adam=1, rian=2: the ids differ from staging's db.add(ClientTodo(id=9, owner="adam", title="Airport write-ups", status="open", sort=90)) db.add(DiscussionItem(id=3, title="Awards rules", sort_order=3)) db.add(DiscussionComment(id=214, feature_key="structure:urls", author="Rian", author_id=ids["rian"], body="written on production", created_at=T0 + timedelta(hours=2), updated_at=T0 + timedelta(hours=2))) db.add(FeaturePriority(id=1, feature_key="price-tracker", priority="later", author="Adam", author_id=ids["adam"], created_at=T0, updated_at=T0)) db.add(OwnerItemState(item_id="decide-x", status="open", acted_by="rian", acted_by_id=ids["rian"], acted_at=T0 + timedelta(days=2), updated_at=T0 + timedelta(days=2))) db.commit() return ids def counts(db) -> dict[str, int]: return {n: db.scalar(select(func.count()).select_from(Base.metadata.tables[n])) for n in CLIENT_WRITTEN_TABLES} def test_every_client_written_table_has_a_spec_and_nothing_else_does(): assert set(R.SPECS) == set(CLIENT_WRITTEN_TABLES) for name in CLIENT_WRITTEN_TABLES: t = Base.metadata.tables[name] spec = R.SPECS[name] assert spec.pk in t.c for col in spec.natural or (): assert col in t.c, (name, col) for col, target in spec.fks.items(): assert col in t.c, (name, col) assert target == "accounts" or CLIENT_WRITTEN_TABLES.index(target) < CLIENT_WRITTEN_TABLES.index(name), \ f"{name}.{col} points at {target}, which must be applied first" def test_export_carries_every_table_and_the_accounts_map(): db = fresh() seed_staging(db) payload = R.export(db) assert set(payload["tables"]) == set(CLIENT_WRITTEN_TABLES) assert payload["counts"]["discussion_comments"] == 3 assert {a["username"] for a in payload["accounts"].values()} == {"rian", "adam", "mark"} # Datetimes travel as ISO strings, aware. assert payload["tables"]["discussion_comments"][0]["created_at"].endswith("+00:00") def test_check_writes_nothing_and_lists_every_table(): staging = fresh() seed_staging(staging) payload = R.export(staging) prod = fresh() seed_production(prod) before = counts(prod) report = R.apply(prod, payload, check=True) assert counts(prod) == before assert set(report.tables) == set(CLIENT_WRITTEN_TABLES) lines = report.lines() assert lines[0] == "[check, nothing written]" assert all(any(line.startswith(f"{name}:") for line in lines) for name in CLIENT_WRITTEN_TABLES) assert report.tables["discussion_comments"] == { "exported": 3, "restored": 2, "same": 0, "updated": 0, "kept": 0, "renumbered": 1, "unresolvable": 0, "unlinked": 1, } def test_apply_keeps_adams_comments_and_decides_conflicts_by_time(): staging = fresh() seed_staging(staging) payload = R.export(staging) prod = fresh() ids = seed_production(prod) report = R.apply(prod, payload) assert report.unresolvable == 0 comments = {c.body: c for c in prod.scalars(select(DiscussionComment))} # Adam's two todo comments are back, linked to production's adam (a different id than staging's). assert {b for b in comments} == {"videos sent", "categories by the 13th", "on the card", "written on production"} assert comments["videos sent"].author_id == ids["adam"] # Production's own comment 214 stayed under 214; staging's 214 came back under a fresh id. assert comments["written on production"].id == 214 assert comments["videos sent"].id == 217 # above both the table's and the export's highest id assert {comments["categories by the 13th"].id, comments["on the card"].id} == {215, 216} # Mark's account is gone on production: his comment keeps the typed name with an empty link. assert comments["on the card"].author == "Mark" and comments["on the card"].author_id is None assert comments["on the card"].item_id == 3 # The ranking Adam changed on staging (newer updated_at) beat production's older row, in place. fp = prod.scalar(select(FeaturePriority).where(FeaturePriority.feature_key == "price-tracker")) assert fp.priority == "next" and fp.id == 1 # rian's decision was taken later on production: production's row is kept. st = prod.get(OwnerItemState, "decide-x") assert st.status == "open" and report.tables["owner_item_states"]["kept"] == 1 # The todo completed on staging beats the open one restored from production. todo = prod.get(ClientTodo, 9) assert todo.status == "done" and todo.completed_by_id == ids["adam"] assert prod.scalar(select(ClientUpload)).todo_id == 9 assert report.tables["quote_requests"]["restored"] == 1 and report.tables["quote_selections"]["restored"] == 1 # A second apply of the same export changes nothing: every row is already there. again = R.apply(prod, payload) assert all(t["restored"] == 0 and t["updated"] == 0 and t["renumbered"] == 0 for t in again.tables.values()) assert again.tables["discussion_comments"]["same"] == 3 # two under their own ids, one found by content assert again.tables["discussion_comments"]["renumbered"] == 0 assert prod.scalar(select(func.count()).select_from(DiscussionComment)) == 4 def test_apply_refuses_an_export_from_other_code(): db = fresh() with pytest.raises(ValueError, match="format"): R.apply(db, {"format": 99, "tables": {}}, check=True) with pytest.raises(ValueError, match="lacks"): R.apply(db, {"format": R.FORMAT, "tables": {"discussion_comments": []}}, check=True) with pytest.raises(ValueError, match="does not know"): R.apply(db, {"format": R.FORMAT, "tables": {n: [] for n in CLIENT_WRITTEN_TABLES} | {"product_variants": []}}, check=True)