201 lines
7.0 KiB
Python
201 lines
7.0 KiB
Python
"""Backup export/verify/import, including the round-trip data guarantees."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tarfile
|
|
|
|
import pytest
|
|
|
|
from dsh_sync.backup import (FORMAT_VERSION, export_backup, import_backup,
|
|
verify_backup, _redact)
|
|
from dsh_sync.config import Config
|
|
from dsh_sync.db import connect
|
|
|
|
|
|
@pytest.fixture()
|
|
def cfg(client):
|
|
"""The config the `client` fixture is actually using."""
|
|
return client.app.state.cfg
|
|
|
|
|
|
def _seed(conn):
|
|
"""A small but referentially complete dataset.
|
|
|
|
Ids start high: the app's bootstrap admin already occupies id 1 (and the
|
|
conftest client builds the app before this runs), so anything lower collides.
|
|
"""
|
|
conn.execute("DELETE FROM tokens")
|
|
conn.execute("DELETE FROM audit_log")
|
|
conn.execute("DELETE FROM users")
|
|
conn.execute("DELETE FROM settings")
|
|
conn.execute("INSERT INTO users (id, username, role, created_at) "
|
|
"VALUES (500, 'admin', 'admin', 't')")
|
|
conn.execute("INSERT INTO users (id, username, role, created_at) "
|
|
"VALUES (501, 'alice', 'member', 't')")
|
|
conn.execute("INSERT INTO tokens (token_hash, user_id, name, created_at) "
|
|
"VALUES ('h1', 501, 'laptop', 't')")
|
|
conn.execute("INSERT INTO settings (scope, scope_id, key, value_json, version, "
|
|
"updated_by, updated_at) VALUES ('team', 0, 'k', '\"v\"', 3, 500, 't')")
|
|
conn.execute("INSERT INTO plugins (id, name, description, latest_version, updated_at) "
|
|
"VALUES ('demo', 'Demo', 'd', '1.0.0', 't')")
|
|
conn.execute("INSERT INTO plugin_versions (plugin_id, version, channel, sha256, "
|
|
"size_bytes, manifest_json, min_harness_version, artifact_path, "
|
|
"published_by, published_at) "
|
|
"VALUES ('demo', '1.0.0', 'stable', 'abc', 10, '{}', '0.0.0', "
|
|
"'demo/1.0.0.tar.gz', 1, 't')")
|
|
conn.execute("INSERT INTO audit_log (actor, action, detail_json, created_at) "
|
|
"VALUES (1, 'x', '{}', 't')")
|
|
conn.commit()
|
|
|
|
|
|
def test_export_writes_manifest_and_tables(cfg, tmp_path):
|
|
conn = connect(cfg)
|
|
try:
|
|
_seed(conn)
|
|
finally:
|
|
conn.close()
|
|
|
|
archive = export_backup(cfg, tmp_path / "b.tar.gz")
|
|
assert archive.is_file()
|
|
|
|
with tarfile.open(archive) as tf:
|
|
names = set(tf.getnames())
|
|
assert "manifest.json" in names
|
|
for table in ("users", "tokens", "settings", "plugins", "plugin_versions", "audit_log"):
|
|
assert f"db/{table}.json" in names
|
|
|
|
|
|
def test_verify_detects_a_tampered_member(cfg, tmp_path):
|
|
"""A bit-flipped payload must be caught by the manifest hashes."""
|
|
conn = connect(cfg)
|
|
try:
|
|
_seed(conn)
|
|
finally:
|
|
conn.close()
|
|
archive = export_backup(cfg, tmp_path / "b.tar.gz")
|
|
|
|
# Rewrite db/users.json with different content, keeping the old manifest.
|
|
with tarfile.open(archive) as tf:
|
|
members = {m.name: tf.extractfile(m).read() for m in tf.getmembers()}
|
|
members["db/users.json"] = json.dumps([{"id": 99, "username": "evil"}]).encode()
|
|
with tarfile.open(archive, "w:gz") as tf:
|
|
for name, blob in members.items():
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(blob)
|
|
tf.addfile(info, __import__("io").BytesIO(blob))
|
|
|
|
with pytest.raises(ValueError, match="corrupt"):
|
|
verify_backup(archive)
|
|
|
|
|
|
def test_verify_rejects_future_format(cfg, tmp_path):
|
|
archive = export_backup(cfg, tmp_path / "b.tar.gz")
|
|
with tarfile.open(archive) as tf:
|
|
members = {m.name: tf.extractfile(m).read() for m in tf.getmembers()}
|
|
m = json.loads(members["manifest.json"])
|
|
m["format_version"] = FORMAT_VERSION + 1
|
|
members["manifest.json"] = json.dumps(m).encode()
|
|
with tarfile.open(archive, "w:gz") as tf:
|
|
for name, blob in members.items():
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(blob)
|
|
tf.addfile(info, __import__("io").BytesIO(blob))
|
|
|
|
with pytest.raises(ValueError, match="unsupported backup format"):
|
|
verify_backup(archive)
|
|
|
|
|
|
def test_import_refuses_to_clobber_without_force(cfg, tmp_path):
|
|
conn = connect(cfg)
|
|
try:
|
|
_seed(conn)
|
|
finally:
|
|
conn.close()
|
|
archive = export_backup(cfg, tmp_path / "b.tar.gz")
|
|
|
|
with pytest.raises(ValueError, match="already holds"):
|
|
import_backup(cfg, archive, force=False)
|
|
|
|
|
|
def test_round_trip_restores_rows_and_links(cfg, tmp_path):
|
|
conn = connect(cfg)
|
|
try:
|
|
_seed(conn)
|
|
finally:
|
|
conn.close()
|
|
archive = export_backup(cfg, tmp_path / "b.tar.gz")
|
|
|
|
# Destroy the data, then restore it.
|
|
conn = connect(cfg)
|
|
try:
|
|
for t in ("tokens", "settings", "plugin_versions", "plugins", "audit_log", "users"):
|
|
conn.execute(f"DELETE FROM {t}")
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
out = import_backup(cfg, archive, force=True)
|
|
assert out["restored_rows"]["users"] == 2
|
|
|
|
conn = connect(cfg)
|
|
try:
|
|
users = {r["id"]: r["username"] for r in conn.execute("SELECT id, username FROM users")}
|
|
assert users == {500: "admin", 501: "alice"}
|
|
# The foreign key must still point at alice, not at whatever id reuse
|
|
# would have produced.
|
|
tok = conn.execute("SELECT user_id FROM tokens WHERE token_hash = 'h1'").fetchone()
|
|
assert tok["user_id"] == 501
|
|
ver = conn.execute("SELECT version FROM settings WHERE key = 'k'").fetchone()
|
|
assert ver["version"] == 3
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_restore_leaves_sequences_usable(cfg, tmp_path):
|
|
"""Explicit-id inserts leave BIGSERIAL behind; the next insert must not collide."""
|
|
conn = connect(cfg)
|
|
try:
|
|
_seed(conn)
|
|
finally:
|
|
conn.close()
|
|
archive = export_backup(cfg, tmp_path / "b.tar.gz")
|
|
import_backup(cfg, archive, force=True)
|
|
|
|
conn = connect(cfg)
|
|
try:
|
|
cur = conn.execute(
|
|
"INSERT INTO users (username, role, created_at) "
|
|
"VALUES ('fresh', 'member', 't') RETURNING id")
|
|
new_id = cur.fetchone()["id"]
|
|
conn.commit()
|
|
assert new_id > 2, f"sequence not advanced: got {new_id}"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_export_includes_plugin_artifacts(cfg, tmp_path):
|
|
conn = connect(cfg)
|
|
try:
|
|
_seed(conn)
|
|
finally:
|
|
conn.close()
|
|
art = cfg.plugins_dir / "demo" / "1.0.0.tar.gz"
|
|
art.parent.mkdir(parents=True, exist_ok=True)
|
|
art.write_bytes(b"plugin-bytes")
|
|
|
|
archive = export_backup(cfg, tmp_path / "b.tar.gz")
|
|
with tarfile.open(archive) as tf:
|
|
assert "plugins/demo/1.0.0.tar.gz" in tf.getnames()
|
|
|
|
# And it comes back on restore.
|
|
art.unlink()
|
|
import_backup(cfg, archive, force=True)
|
|
assert art.read_bytes() == b"plugin-bytes"
|
|
|
|
|
|
def test_redact_hides_password():
|
|
assert _redact("postgresql://dsh:s3cret@127.0.0.1:5432/db") == \
|
|
"postgresql://dsh:***@127.0.0.1:5432/db"
|
|
assert _redact("postgresql://dsh@127.0.0.1:5432/db") == \
|
|
"postgresql://dsh@127.0.0.1:5432/db"
|