feat(backup): v3 backup covers rulebooks, rules, events + join tables

The v2 backup silently dropped the entire rulebook system (rulebooks, topics,
rules), the project subscription/suppression join tables, and events — so a
'full' backup wasn't. v3 adds all of them with FK re-mapping on restore, and a
_not_included field that names the still-deferred tables (ACL groups/shares,
api_keys, embeddings, transient/operational) so the gap is explicit, not silent.

restore_full_backup routes v2 and v3 through one path; v3-only sections are
guarded by data.get so a v2 payload still restores cleanly.

Tests: version/coverage constants, pure join-table row helpers, and the export
contract via a mocked session (CI has no DB; full round-trip is a manual check).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-09 23:52:14 -04:00
parent 9924f873b9
commit d6c8470ab2
2 changed files with 428 additions and 10 deletions
+72
View File
@@ -0,0 +1,72 @@
"""Unit tests for the v3 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/events appear).
"""
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from scribe.services import backup
def test_backup_version_is_v3():
assert backup.BACKUP_VERSION == 3
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", "embeddings"):
assert table in backup._NOT_INCLUDED
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_v3_sections():
with patch("scribe.services.backup.async_session", lambda: _CM()):
out = await backup.export_full_backup()
assert out["version"] == 3
assert out["scope"] == "full"
assert "api_keys" in out["_not_included"]
# The sections v2 silently dropped must now be present (empty here).
for key in ("rulebooks", "rulebook_topics", "rules",
"rulebook_subscriptions", "rule_suppressions",
"topic_suppressions", "events"):
assert key in out, f"missing v3 section: {key}"
assert out[key] == []