felhom.eu/scripts/hub-db-backup/test_hub_db_backup.py

#!/usr/bin/env python3
"""Tests for felhom-hub-db-backup and felhom-hub-db-restore-test (R-173).

No test reaches the hub, PBS or ep0: `kubectl` and `proxmox-backup-client` are fakes on PATH (the pod's
/data/snapshots is a temp dir; the PBS "server" is a temp dir). `sqlite3`, `date`, `shred` are the real tools.
Each test asserts the CONSEQUENCE: whether a push happened and whether the success signal (the file the alarm reads)
was written. Run: python3 scripts/hub-db-backup/test_hub_db_backup.py
"""
import json
import os
import shutil
import sqlite3
import stat
import subprocess
import tempfile
import time
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
PUSH = os.path.join(HERE, "felhom-hub-db-backup")
RESTORE = os.path.join(HERE, "felhom-hub-db-restore-test")

FAKE_KUBECTL = r'''#!/usr/bin/env python3
import os, subprocess, sys
a = sys.argv[1:]
pod = os.environ["FAKE_POD_DATA"]
i = a.index("--")
cmd = a[i + 1:]
def m(p): return p.replace("/data", pod, 1)
if cmd[:2] == ["sh", "-c"]:
    sys.exit(subprocess.call(["sh", "-c", cmd[2].replace("/data", pod)]))
if cmd[0] == "cat":
    if os.environ.get("FAKE_TRUNCATE"):
        data = open(m(cmd[1]), "rb").read()
        sys.stdout.buffer.write(data[: len(data) // 2]); sys.exit(0)
    sys.exit(subprocess.call(["cat", m(cmd[1])]))
sys.exit(97)
'''

FAKE_PBS = r'''#!/usr/bin/env python3
import json, os, shutil, sys, time
a = sys.argv[1:]
srv = os.environ["FAKE_PBS_DIR"]
open(os.path.join(srv, "calls.log"), "a").write(json.dumps({"argv": a, "pw": os.environ.get("PBS_PASSWORD_FILE", ""), "fp": os.environ.get("PBS_FINGERPRINT", "")}) + "\n")
if a[0] == "backup":
    if os.environ.get("FAKE_PBS_FAIL"): sys.exit(1)
    src = a[1].split(":", 1)[1]
    t = int(time.time())
    d = os.path.join(srv, "snaps", str(t)); os.makedirs(d, exist_ok=True)
    shutil.copy(os.path.join(src, "hub.db"), os.path.join(d, "hub.db"))
    sys.exit(0)
if a[0] == "snapshot" and a[1] == "list":
    base = os.path.join(srv, "snaps")
    out = [{"backup-type": "host", "backup-id": "dooplex-hub", "backup-time": int(x)} for x in (os.listdir(base) if os.path.isdir(base) else [])]
    print(json.dumps(out)); sys.exit(0)
if a[0] == "restore":
    if os.environ.get("FAKE_PBS_FAIL"): sys.exit(1)
    import calendar
    t = calendar.timegm(time.strptime(a[1].split("/")[-1], "%Y-%m-%dT%H:%M:%SZ"))
    os.makedirs(a[3], exist_ok=True)
    shutil.copy(os.path.join(srv, "snaps", str(t), "hub.db"), os.path.join(a[3], "hub.db"))
    sys.exit(0)
sys.exit(98)
'''


# The CI runner carries python3 and git only (no sqlite3 CLI). When the real `sqlite3` is absent, a stand-in built on
# Python's own sqlite3 module (the SAME SQLite library) answers exactly the calls the scripts make:
# `sqlite3 -readonly <db> '<sql>'`, one row per line, "|"-separated, exit 1 + stderr on an error. Said out loud below.
FAKE_SQLITE3 = r'''#!/usr/bin/env python3
import sqlite3, sys
a = [x for x in sys.argv[1:] if x != "-readonly"]
try:
    db = sqlite3.connect("file:" + a[0] + "?mode=ro", uri=True)
    for row in db.execute(a[1]):
        print("|".join("" if v is None else str(v) for v in row))
except Exception as e:
    print("Error: " + str(e), file=sys.stderr); sys.exit(1)
'''
REAL_SQLITE3 = shutil.which("sqlite3")
if not REAL_SQLITE3:
    print("test_hub_db_backup: no sqlite3 CLI here — using the Python-sqlite3 stand-in (same SQLite library)")


def make_db(path, hosts=2, recovery=("enc:v1:abc", "enc:v1:def"), corrupt=False):
    db = sqlite3.connect(path)
    db.execute("CREATE TABLE hosts (host_id TEXT)")
    db.execute("CREATE TABLE host_recovery (host_id TEXT, secret TEXT)")
    db.executemany("INSERT INTO hosts VALUES (?)", [("h%d" % i,) for i in range(hosts)])
    db.executemany("INSERT INTO host_recovery VALUES ('h', ?)", [(r,) for r in recovery])
    db.execute("CREATE TABLE filler (x BLOB)")
    db.executemany("INSERT INTO filler VALUES (randomblob(3000))", [()] * 40)
    db.execute("CREATE INDEX filler_x ON filler(x)")
    db.commit(); db.close()
    if corrupt:  # overwrite a late page (index b-tree) so integrity_check reports errors but the file still opens
        size = os.path.getsize(path)
        with open(path, "r+b") as f:
            f.seek(size - 4096 + 100); f.write(b"\xff" * 2000)


def stamp(epoch):
    return time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(epoch))


class Base(unittest.TestCase):
    def setUp(self):
        self.t = tempfile.mkdtemp()
        j = lambda *p: os.path.join(self.t, *p)
        for d in ("bin", "pod/snapshots", "conf", "state", "textfile", "pbs"):
            os.makedirs(j(d), exist_ok=True)
        fakes = [("kubectl", FAKE_KUBECTL), ("proxmox-backup-client", FAKE_PBS)]
        if not REAL_SQLITE3:
            fakes.append(("sqlite3", FAKE_SQLITE3))
        for name, body in fakes:
            p = j("bin", name); open(p, "w").write(body); os.chmod(p, 0o755)
        open(j("conf", "env"), "w").write(
            "PBS_REPOSITORY_PUSH='dooplex-hub@pbs!push@127.0.0.1:18007:felhom-offsite'\n"
            "PBS_REPOSITORY_RESTORE='dooplex-hub@pbs!restore@127.0.0.1:18007:felhom-offsite'\n"
            "PBS_FINGERPRINT='aa:bb'\n")
        for f in ("token-push", "token-restore", "enc.key"):
            open(j("conf", f), "w").write("x")
        self.env = dict(os.environ, PATH=j("bin") + ":/usr/bin:/bin", FAKE_POD_DATA=j("pod"), FAKE_PBS_DIR=j("pbs"),
                        FELHOM_HUBBK_CONF=j("conf"), FELHOM_HUBBK_STATE=j("state"), FELHOM_HUBBK_TEXTFILE_DIR=j("textfile"))
        self.j = j

    def tearDown(self):
        shutil.rmtree(self.t, ignore_errors=True)

    def snapshot(self, age_s=600, **kw):
        p = self.j("pod", "snapshots", "hub-%s.db" % stamp(int(time.time()) - age_s))
        make_db(p, **kw)
        return p

    def run_script(self, script, **extra):
        env = dict(self.env, **extra)
        return subprocess.run([script], env=env, capture_output=True, text=True, timeout=60)

    def pushed(self):
        d = self.j("pbs", "snaps")
        return sorted(os.listdir(d)) if os.path.isdir(d) else []

    def signal(self, name):
        return os.path.exists(self.j("textfile", name))

    def calls(self):
        p = self.j("pbs", "calls.log")
        return [json.loads(l) for l in open(p)] if os.path.exists(p) else []


class Push(Base):
    def test_happy_path_pushes_encrypted_to_operator_and_writes_signal(self):
        self.snapshot()
        r = self.run_script(PUSH)
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertEqual(len(self.pushed()), 1)
        self.assertTrue(self.signal("felhom_hub_db_backup.prom"))
        c = [x for x in self.calls() if x["argv"][0] == "backup"][0]
        a = c["argv"]
        for flag in ("--ns", "--backup-id", "--crypt-mode", "--keyfile", "--repository"):
            self.assertIn(flag, a, "push without %s" % flag)
        for flag, val in (("--ns", "operator"), ("--backup-id", "dooplex-hub"), ("--crypt-mode", "encrypt")):
            self.assertEqual(a[a.index(flag) + 1], val)
        self.assertTrue(a[a.index("--keyfile") + 1].endswith("/enc.key"))
        self.assertIn("!push@", a[a.index("--repository") + 1])
        self.assertTrue(c["pw"].endswith("/token-push"))
        self.assertNotIn("x", " ".join(a).split())  # the token's value never on the command line
        self.assertEqual(os.listdir(self.j("state", "stage")), [], "the staged plaintext copy is left behind")
        txt = open(self.j("textfile", "felhom_hub_db_backup.prom")).read()
        self.assertIn("felhom_hub_db_backup_last_success_timestamp_seconds ", txt)

    def test_corrupt_copy_is_never_pushed(self):
        self.snapshot(corrupt=True)
        r = self.run_script(PUSH)
        self.assertNotEqual(r.returncode, 0)
        self.assertIn("integrity_check", r.stderr)
        self.assertEqual(self.pushed(), [])
        self.assertFalse(self.signal("felhom_hub_db_backup.prom"))

    def test_stale_snapshot_is_not_pushed_again(self):
        self.snapshot(age_s=27 * 3600)
        r = self.run_script(PUSH)
        self.assertNotEqual(r.returncode, 0)
        self.assertIn("stopped snapshotting", r.stderr)
        self.assertEqual(self.pushed(), [])
        self.assertFalse(self.signal("felhom_hub_db_backup.prom"))

    def test_truncated_copy_is_not_pushed(self):
        self.snapshot()
        r = self.run_script(PUSH, FAKE_TRUNCATE="1")
        self.assertNotEqual(r.returncode, 0)
        # the SIZE guard must be the one that refuses (half a file usually fails integrity_check too, which would
        # mask a missing size check — red-proof P4's first run did exactly that)
        self.assertIn("bytes, the pod's file is", r.stderr)
        self.assertEqual(self.pushed(), [])
        self.assertFalse(self.signal("felhom_hub_db_backup.prom"))

    def test_failed_push_writes_no_signal(self):
        self.snapshot()
        r = self.run_script(PUSH, FAKE_PBS_FAIL="1")
        self.assertNotEqual(r.returncode, 0)
        self.assertFalse(self.signal("felhom_hub_db_backup.prom"))
        self.assertEqual(os.listdir(self.j("state", "stage")), [])

    def test_no_snapshot_fails(self):
        r = self.run_script(PUSH)
        self.assertNotEqual(r.returncode, 0)
        self.assertFalse(self.signal("felhom_hub_db_backup.prom"))

    def test_empty_hosts_is_not_pushed(self):
        self.snapshot(hosts=0)
        r = self.run_script(PUSH)
        self.assertNotEqual(r.returncode, 0)
        self.assertEqual(self.pushed(), [])

    def test_newest_snapshot_is_the_one_pushed(self):
        self.snapshot(age_s=25 * 3600, hosts=1)
        self.snapshot(age_s=600, hosts=3)
        self.assertEqual(self.run_script(PUSH).returncode, 0)
        db = os.path.join(self.j("pbs", "snaps"), self.pushed()[0], "hub.db")
        self.assertEqual(sqlite3.connect(db).execute("SELECT COUNT(*) FROM hosts").fetchone()[0], 3)


class RestoreTest(Base):
    def push_one(self, **kw):
        self.snapshot(**kw)
        r = self.run_script(PUSH)
        self.assertEqual(r.returncode, 0, r.stderr)

    def test_happy_path_writes_signal_with_the_read_only_token(self):
        self.push_one()
        r = self.run_script(RESTORE)
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertTrue(self.signal("felhom_hub_db_restore.prom"))
        for c in self.calls():
            if c["argv"][0] in ("restore", "snapshot"):
                self.assertTrue(c["pw"].endswith("/token-restore"), c)
                self.assertIn("!restore@", c["argv"][c["argv"].index("--repository") + 1])
        self.assertEqual([x for x in os.listdir(self.j("state")) if x.startswith("restore.")], [], "restored copy left behind")

    def test_readable_console_password_fails(self):
        self.push_one(recovery=("enc:v1:abc", "hunter2"))
        r = self.run_script(RESTORE)
        self.assertNotEqual(r.returncode, 0)
        self.assertIn("stored readable", r.stderr)
        self.assertNotIn("hunter2", r.stderr + r.stdout)
        self.assertFalse(self.signal("felhom_hub_db_restore.prom"))

    def test_no_copy_on_ep0_fails(self):
        r = self.run_script(RESTORE)
        self.assertNotEqual(r.returncode, 0)
        self.assertFalse(self.signal("felhom_hub_db_restore.prom"))

    def test_old_copy_fails(self):
        self.push_one()
        r = self.run_script(RESTORE, FELHOM_HUBBK_NOW=str(int(time.time()) + 51 * 3600))
        self.assertNotEqual(r.returncode, 0)
        self.assertFalse(self.signal("felhom_hub_db_restore.prom"))

    def test_failed_restore_fails(self):
        self.push_one()
        r = self.run_script(RESTORE, FAKE_PBS_FAIL="1")
        self.assertNotEqual(r.returncode, 0)
        self.assertFalse(self.signal("felhom_hub_db_restore.prom"))


class Units(unittest.TestCase):
    def read(self, n):
        return open(os.path.join(HERE, n)).read()

    def test_schedules(self):
        self.assertIn("OnCalendar=*-*-* 02:30:00", self.read("felhom-hub-db-backup.timer"))
        self.assertIn("OnCalendar=Sun *-*-* 04:30:00", self.read("felhom-hub-db-restore-test.timer"))

    def test_units_run_the_installed_scripts(self):
        self.assertIn("ExecStart=/usr/local/sbin/felhom-hub-db-backup\n", self.read("felhom-hub-db-backup.service"))
        self.assertIn("ExecStart=/usr/local/sbin/felhom-hub-db-restore-test\n", self.read("felhom-hub-db-restore-test.service"))


if __name__ == "__main__":
    unittest.main(verbosity=2)