dsh/server/tests/test_backup.py

"""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"