FabledScribe/tests/test_services_backup.py

"""Unit tests for the v4 backup export contract.

CI runs pytest with no database, so these cover the parts that don't need one:
the version/coverage constants, the pure join-table row helpers, and the export
dict shape (via a mocked session). Full FK-remapping round-trip is exercised
manually against a real DB (export a backup, confirm rulebooks appear).
"""
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from scribe.services import backup


def test_backup_version_is_v6():
    """v6 added note_supersessions (#278). The bump is the point of the test —
    a payload section added without moving the version produces backups that
    are structurally different and indistinguishable by inspection."""
    assert backup.BACKUP_VERSION == 6


def test_not_included_lists_the_known_gaps():
    # The deferred tables must be surfaced explicitly, not silently dropped.
    for table in ("groups", "project_shares", "note_shares", "api_keys",
                  "note_embeddings", "retrieval_logs"):
        assert table in backup._NOT_INCLUDED


def test_every_table_is_either_backed_up_or_explicitly_excluded():
    """THE GUARD (#2293), and the only shape of test that catches an ABSENCE.

    A new table gets a model and a migration — both fail loudly if wrong — and
    then silently never gets a backup section. No error, no warning, and a
    restore that reports success. That is how `systems`, `record_systems`,
    `note_usage_events`, `design_systems`, `design_tokens` and `repo_bindings`
    all went missing, over five migrations, with nothing to notice.

    Extending the export fixes today. THIS fixes the next one: adding a table
    now fails here until someone either backs it up or states in
    `_NOT_INCLUDED` that it shouldn't be. Either is fine; silence is not.
    """
    from scribe.models import Base

    schema = set(Base.metadata.tables)
    accounted = set(backup._BACKED_UP) | set(backup._NOT_INCLUDED)

    unaccounted = schema - accounted
    assert not unaccounted, (
        f"{len(unaccounted)} table(s) are neither backed up nor explicitly "
        f"excluded: {sorted(unaccounted)}. Add each to backup._BACKED_UP (and "
        f"give it an export + restore section) or to backup._NOT_INCLUDED with "
        f"a reason in the comment above it."
    )

    # And the reverse: a name in either list that no longer exists is a lie the
    # guard would otherwise keep telling. This half is what caught "embeddings",
    # "invitations" and "password_resets" — three entries that named nothing.
    phantom = accounted - schema
    assert not phantom, (
        f"backup lists table(s) that are not in the schema: {sorted(phantom)}. "
        f"Renamed or dropped — fix the list rather than leaving it to read as "
        f"coverage."
    )


def test_join_table_row_helpers_are_pure():
    subs = [SimpleNamespace(project_id=1, rulebook_id=2)]
    rsup = [SimpleNamespace(project_id=1, rule_id=9)]
    tsup = [SimpleNamespace(project_id=1, topic_id=7)]
    assert backup._subscription_rows(subs) == [{"project_id": 1, "rulebook_id": 2}]
    assert backup._rule_suppression_rows(rsup) == [{"project_id": 1, "rule_id": 9}]
    assert backup._topic_suppression_rows(tsup) == [{"project_id": 1, "topic_id": 7}]


class _Result:
    def scalars(self):
        return self

    def all(self):
        return []


class _Session:
    async def execute(self, *a, **k):
        return _Result()

    async def get(self, *a, **k):
        return None


class _CM:
    async def __aenter__(self):
        return _Session()

    async def __aexit__(self, *a):
        return False


@pytest.mark.asyncio
async def test_export_full_backup_contains_every_declared_section():
    with patch("scribe.services.backup.async_session", lambda: _CM()):
        out = await backup.export_full_backup()

    assert out["version"] == backup.BACKUP_VERSION
    assert out["scope"] == "full"
    assert "api_keys" in out["_not_included"]
    # The sections v2 silently dropped, the six v5 added, and v6's
    # note_supersessions (all empty here).
    for key in ("rulebooks", "rulebook_topics", "rules",
                "rulebook_subscriptions", "rule_suppressions",
                "topic_suppressions",
                "systems", "record_systems", "design_systems",
                "design_tokens", "note_usage_events", "repo_bindings",
                "note_supersessions"):
        assert key in out, f"missing export section: {key}"
        assert out[key] == []


def test_supersession_rows_serialise_the_pair():
    """The row builder is a plain function precisely so it can be tested with
    no database — same reason as the other v5/v6 builders."""
    class _Row:
        def __init__(self, a, b):
            self.superseder_id, self.superseded_id = a, b

    assert backup._note_supersession_rows([_Row(9, 4), _Row(9, 5)]) == [
        {"superseder_id": 9, "superseded_id": 4},
        {"superseder_id": 9, "superseded_id": 5},
    ]