tests/test_services_backup.py

@@ -188,8 +188,17 @@ def _stand_in(model):
     of a dict. Typed rather than a bare instance because the helpers call
     `.isoformat()` on the timestamps, which `None` does not have.
     """
+    import itertools
     import sqlalchemy as sa
 
+    # DISTINCT integers, not a constant 1. The import guard feeds this row's
+    # exported form to the builder, and several builders reject a self-edge —
+    # a supersession or a rule relation whose two ends are the same id is not
+    # a weaker claim, it is a row pointing at itself. With one value shared by
+    # every column those builders would refuse a legitimate stand-in and the
+    # guard would read as a fixture bug. Only the KEYS matter to either guard.
+    ints = itertools.count(1)
+
     row = model()
     for column in model.__table__.columns:
         t = column.type
@@ -200,7 +209,7 @@ def _stand_in(model):
         elif isinstance(t, sa.Boolean):
             value = False
         elif isinstance(t, sa.Integer):
-            value = 1
+            value = next(ints)
         elif isinstance(t, sa.ARRAY) or isinstance(getattr(t, "impl", None), sa.ARRAY):
             value = []
         elif isinstance(t, (sa.Text, sa.String)):
@@ -245,40 +254,179 @@ def test_every_column_is_exported_or_declared_excluded(table):
     )
 
 
-def test_the_usage_importer_restores_the_reading_project():
-    """The column guard above checks the EXPORT side only.
+def _import_guard_targets():
+    """table -> (model, export helper, import builder).
 
-    A column can be exported faithfully and then dropped on the way back in,
-    which restores a backup that reports success and has quietly lost a
-    dimension — #3182's failure mode, one direction over. There is no general
-    import-side guard yet; this covers the column #4196 added, by source
-    inspection, because the behavioural path needs Postgres.
-
-    It also pins the DEGRADE. `code_shape_events` skips a row whose project
-    will not map, because a shape event without its project says nothing. A
-    usage event is not like that: the project is optional by design and null
-    already means "not reported", so an unmappable one must restore as
-    unreported rather than vanish — dropping it would lose a real pull and
-    deflate the very pull-through this table exists to report.
+    Built from the export registry so the two cannot drift apart: a table with
+    a row helper and no builder shows up here as a KeyError with its own name
+    in it, rather than as a table nobody checks.
     """
-    import inspect
+    builders = {
+        "users": backup._build_user,
+        "projects": backup._build_project,
+        "milestones": backup._build_milestone,
+        "notes": backup._build_note,
+        "task_logs": backup._build_task_log,
+        "note_drafts": backup._build_note_draft,
+        "note_versions": backup._build_note_version,
+        "settings": backup._build_setting,
+        "rulebooks": backup._build_rulebook,
+        "rulebook_topics": backup._build_topic,
+        "rules": backup._build_rule,
+        "rule_versions": backup._build_rule_version,
+        "systems": backup._build_system,
+        "canonical_systems": backup._build_canonical_system,
+        "record_systems": backup._build_record_system,
+        "note_supersessions": backup._build_note_supersession,
+        "rule_relations": backup._build_rule_relation,
+        "note_usage_events": backup._build_usage_event,
+        "rule_usage_events": backup._build_rule_usage_event,
+        "retrieval_tuning_events": backup._build_retrieval_tuning_event,
+        "design_systems": backup._build_design_system,
+        "design_tokens": backup._build_design_token,
+        "repo_bindings": backup._build_repo_binding,
+        "code_shapes": backup._build_code_shape,
+        "code_shape_events": backup._build_code_shape_event,
+        "code_shape_uses": backup._build_code_shape_use,
+    }
+    return {
+        table: (model, helper, builders[table])
+        for table, (model, helper) in _column_guard_targets().items()
+    }
 
-    src = inspect.getsource(backup._restore_v2)
-    marker = 'for ev in data.get("note_usage_events", []):'
-    assert marker in src, "the usage import loop moved; this guard is blind"
-    block = src[src.index(marker):][:1200]
 
-    assert 'ev.get("project_id")' in block, (
-        "the usage importer drops project_id — a restore would report success "
-        "and come back without the reading project"
+def _everything_maps(row: dict) -> "backup._Maps":
+    """Id maps in which every id the row mentions resolves.
+
+    The guard is about which COLUMNS a builder sets, not about what it does
+    when a foreign key is missing — that is the skip/degrade question, which
+    the tests below ask directly. So every lookup succeeds here, and a builder
+    that returned None would be a bug in the fixture rather than a finding.
+    """
+    maps = backup._Maps()
+    ids = {v for v in row.values() if isinstance(v, int)} | {0, 1}
+    for name in ("users", "projects", "milestones", "notes", "rulebooks",
+                 "topics", "rules", "systems", "design_systems", "shapes"):
+        getattr(maps, name).update({i: i + 1000 for i in ids})
+    # `canonical_slug` only, never `slug`: a canonical_systems row is built
+    # exactly when its slug is NOT already known to the destination, so
+    # seeding it from the row's own slug would make that builder skip.
+    slug = row.get("canonical_slug")
+    if slug:
+        maps.canonical_by_slug[slug] = 7
+    return maps
+
+
+@pytest.mark.parametrize("table", sorted(_import_guard_targets()))
+def test_every_exported_column_is_imported_or_declared_excluded(table):
+    """THE COLUMN GUARD, THE OTHER WAY (#4197).
+
+    The export guard above makes a dropped column unexpressible on the way
+    OUT. Nothing watched the way back IN, and that is the worse half: an
+    export gap leaves an obviously thin backup, an import gap means holding a
+    complete, correct file and restoring an incomplete database from it, with
+    a success message.
+
+    It was one-sided because the code was — `_restore_v2` built every model
+    inline, so there was no per-table unit to hand a stand-in to. There is
+    now, and this composes the two halves end to end: export a stand-in row,
+    feed THAT dict to the builder, and read which columns the constructed
+    model actually received.
+
+    What it caught on the first run: `code_shapes` was exporting `reason_code`,
+    `recheck_at` and `diverges_from` and importing none of them. Every judged
+    shape would have restored with its verdict and without the code for why,
+    every recheck flag cleared, and every divergence pointer gone.
+    """
+    model, helper, builder = _import_guard_targets()[table]
+    [row] = helper([_stand_in(model)])
+
+    built = builder(row, _everything_maps(row))
+    assert built is not None, (
+        f"{table}: the builder skipped a row whose ids all resolve — "
+        "the guard fixture is wrong, or the builder is"
     )
-    assert "project_id_map" in block, (
-        "project_id must be re-mapped; a raw id points at whatever project "
-        "happens to hold that number in the destination install"
+
+    received = set(built.__dict__) - {"_sa_instance_state"}
+    columns = {c.name for c in model.__table__.columns}
+    missing = columns - received
+    declared = backup._IMPORT_COLUMN_EXCLUSIONS[table]
+
+    assert missing == declared, (
+        f"{table}: imported columns and _IMPORT_COLUMN_EXCLUSIONS disagree.\n"
+        f"  dropped but not declared: {sorted(missing - declared)}\n"
+        f"  declared but imported anyway: {sorted(declared - missing)}"
     )
-    assert "continue" not in block.split('project_id=')[1][:200], (
-        "an unmappable project must degrade to None, not skip the row"
+
+
+def test_the_import_guard_covers_every_table_the_export_guard_does():
+    """The two registries have to hold the same tables, or a table can be
+    guarded in one direction and silently unguarded in the other — which is
+    the state this whole pair of guards exists to end."""
+    assert set(_import_guard_targets()) == set(_column_guard_targets())
+    assert set(backup._IMPORT_COLUMN_EXCLUSIONS) == set(_column_guard_targets())
+
+
+def test_an_unmappable_project_degrades_a_usage_event_and_skips_a_shape_event():
+    """The skip-or-degrade choice is per table, and both answers are right.
+
+    `code_shape_events` SKIPS a row whose project will not map — a shape event
+    without its project says nothing. `note_usage_events` DEGRADES to None —
+    the project is optional by design, null already means "not reported", and
+    dropping the row would lose a real pull and deflate the very pull-through
+    the table exists to report.
+
+    Until #4197 this was asserted by reading the source of `_restore_v2` with
+    `inspect.getsource`, because there was no unit to call. Now there is.
+    """
+    maps = backup._Maps()
+    maps.notes[5] = 55
+    maps.users[9] = 99
+    maps.shapes[3] = 33
+    # project 7 is deliberately absent from maps.projects
+
+    event = backup._build_usage_event(
+        {"note_id": 5, "user_id": 9, "project_id": 7,
+         "event": "pulled", "source": "search"},
+        maps,
     )
+    assert event is not None, "an unmappable project must not drop a real pull"
+    assert event.project_id is None
+    assert event.note_id == 55
+
+    shape_event = backup._build_code_shape_event(
+        {"shape_id": 3, "project_id": 7, "path": "a.py", "symbol": "f"}, maps,
+    )
+    assert shape_event is None, (
+        "a shape event whose project did not map says nothing and must be "
+        "skipped, not restored project-less"
+    )
+
+
+def test_a_shapes_divergence_pointer_is_remapped_not_carried():
+    """`diverges_from` is a FK to notes.id, like `snippet_id`. Carrying the
+    source id would point at whatever snippet took that number in the
+    destination — wrong rather than missing, and nothing downstream could
+    tell. It was dropped entirely until #4197; restoring it raw would have
+    been the worse fix."""
+    maps = backup._Maps()
+    maps.projects[1] = 11
+    maps.notes[4] = 44
+
+    shape = backup._build_code_shape(
+        {"project_id": 1, "status": "exempt", "diverges_from": 4,
+         "reason_code": "scoped-css", "recheck_at": "2026-01-01T00:00:00+00:00"},
+        maps,
+    )
+    assert shape.diverges_from == 44, "not re-mapped through the note map"
+    assert shape.reason_code == "scoped-css"
+    assert shape.recheck_at is not None
+
+    # A pointer whose target did not survive lands NULL rather than dangling.
+    orphan = backup._build_code_shape(
+        {"project_id": 1, "status": "exempt", "diverges_from": 999}, maps,
+    )
+    assert orphan.diverges_from is None
 
 
 def test_the_column_guard_covers_every_table_with_a_row_helper():