admin
4cee21acf7
gates / gates (push) Successful in 5m32s
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
440 lines
22 KiB
Python
440 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for felhom-dooplex-offsite, its restore test and felhom-backup-failmail (R-232).
|
|
|
|
No test reaches Gitea, k3s, PBS, ep0 or Resend: `kubectl`, `proxmox-backup-client` and `pg_restore` are fakes on PATH
|
|
(the Gitea pod's /data is a temp dir; the PBS "server" is a temp dir; the fake pg_restore accepts a file that starts
|
|
with PostgreSQL's custom-dump magic "PGDMP"). `git`, `tar`, `sha256sum`, `shred`, `find` are the real tools, so
|
|
`git fsck` runs on real repositories. Each test asserts the CONSEQUENCE: whether a push happened and whether the
|
|
success signal (the file the alarm reads) was written. Run: python3 scripts/dooplex-offsite/test_dooplex_offsite.py
|
|
"""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
PUSH = os.path.join(HERE, "felhom-dooplex-offsite")
|
|
RESTORE = os.path.join(HERE, "felhom-dooplex-offsite-restore-test")
|
|
FAILMAIL = os.path.join(HERE, "felhom-backup-failmail")
|
|
|
|
FAKE_KUBECTL = r'''#!/usr/bin/env python3
|
|
import os, subprocess, sys
|
|
a = sys.argv[1:]
|
|
pod = os.environ["FAKE_POD_DATA"]
|
|
cmd = a[a.index("--") + 1:]
|
|
if "vaultwarden-system" in a:
|
|
vw = os.environ["FAKE_VW_DATA"]
|
|
if cmd == ["/vaultwarden", "backup"]:
|
|
if os.environ.get("FAKE_VW_BACKUP_FAIL"):
|
|
print("[ERROR] backup failed: database is locked"); sys.exit(1)
|
|
name = "db_20261009_094401.sqlite3"
|
|
import shutil as sh; sh.copy(os.path.join(vw, "db.sqlite3"), os.path.join(vw, name))
|
|
print("[NOTICE] You are using a plain text `ADMIN_TOKEN` which is insecure.")
|
|
print("Backup to 'data/%s' was successful" % name); sys.exit(0)
|
|
if cmd[0] == "cat":
|
|
sys.exit(subprocess.call(["cat", vw + cmd[1][len("/data"):]]))
|
|
if cmd[:2] == ["rm", "-f"]:
|
|
p = vw + cmd[2][len("/data"):]
|
|
open(os.path.join(os.environ["FAKE_STATE"], "vw-removed"), "a").write(cmd[2] + "\n")
|
|
if os.path.exists(p): os.remove(p)
|
|
sys.exit(0)
|
|
if cmd[:2] == ["sh", "-c"]:
|
|
sys.exit(subprocess.call(["sh", "-c", cmd[2].replace("/data", vw)]))
|
|
if cmd[0] == "tar":
|
|
sys.exit(subprocess.call([vw if x == "/data" else x for x in cmd]))
|
|
sys.exit(96)
|
|
if cmd[0] == "tar":
|
|
cnt = os.path.join(os.environ["FAKE_STATE"], "tar-calls")
|
|
n = int(open(cnt).read()) if os.path.exists(cnt) else 0
|
|
open(cnt, "w").write(str(n + 1))
|
|
if n < int(os.environ.get("FAKE_TAR_FAILS", "0")):
|
|
sys.stderr.write("tar: file vanished\n"); sys.exit(1)
|
|
c = ["tar" if x == "tar" else (pod if x == "/data" else x) for x in cmd]
|
|
sys.exit(subprocess.call(c))
|
|
if cmd[0] == "find":
|
|
out = subprocess.run(["find", pod + cmd[1][len("/data"):]] + cmd[2:],
|
|
capture_output=True, text=True)
|
|
for line in out.stdout.splitlines():
|
|
print("/data" + line[len(pod):])
|
|
for extra in filter(None, os.environ.get("FAKE_EXTRA_REPOS", "").split(",")):
|
|
print("/data/git/repositories/admin/" + extra)
|
|
sys.exit(out.returncode)
|
|
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", "")}) + "\n")
|
|
base = os.path.join(srv, "snaps")
|
|
if a[0] == "backup":
|
|
if os.environ.get("FAKE_PBS_FAIL"): sys.exit(1)
|
|
src = a[1].split(":", 1)[1]
|
|
t = int(os.environ.get("FAKE_PBS_TIME", time.time()))
|
|
shutil.copytree(src, os.path.join(base, str(t)))
|
|
sys.exit(0)
|
|
if a[0] == "snapshot" and a[1] == "list":
|
|
out = [{"backup-type": "host", "backup-id": "dooplex-gitea", "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":
|
|
import calendar
|
|
t = calendar.timegm(time.strptime(a[1].split("/")[-1], "%Y-%m-%dT%H:%M:%SZ"))
|
|
shutil.copytree(os.path.join(base, str(t)), a[3])
|
|
sys.exit(0)
|
|
sys.exit(98)
|
|
'''
|
|
|
|
FAKE_PG_RESTORE = r'''#!/bin/sh
|
|
[ "$1" = "--list" ] || exit 97
|
|
head -c 5 "$2" | grep -q '^PGDMP' || { echo "pg_restore: error: input file does not appear to be a valid archive" >&2; exit 1; }
|
|
'''
|
|
|
|
|
|
# The CI runner has no sqlite3 CLI: a stand-in on Python's own sqlite3 module (the same SQLite library) answers the calls
|
|
# the scripts make — `sqlite3 -readonly <db> '<sql>'`, one row per line, exit 1 on an error.
|
|
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 = None if os.environ.get("FORCE_FAKE_SQLITE3") else shutil.which("sqlite3")
|
|
|
|
|
|
def make_vw_db(path, users=1, items=3):
|
|
import sqlite3
|
|
db = sqlite3.connect(path)
|
|
db.execute("CREATE TABLE users (uuid TEXT PRIMARY KEY, email TEXT)")
|
|
db.execute("CREATE TABLE ciphers (uuid TEXT PRIMARY KEY, data TEXT)")
|
|
for i in range(users): db.execute("INSERT INTO users VALUES (?, ?)", ("u%d" % i, "u%d@example.invalid" % i))
|
|
for i in range(items): db.execute("INSERT INTO ciphers VALUES (?, ?)", ("c%d" % i, "2.encrypted"))
|
|
db.commit(); db.close()
|
|
|
|
|
|
def git(*a, cwd=None):
|
|
env = dict(os.environ, GIT_AUTHOR_NAME="t", GIT_AUTHOR_EMAIL="t@t", GIT_COMMITTER_NAME="t", GIT_COMMITTER_EMAIL="t@t")
|
|
subprocess.run(["git", *a], cwd=cwd, check=True, capture_output=True, env=env)
|
|
|
|
|
|
class Base(unittest.TestCase):
|
|
def setUp(self):
|
|
self.t = tempfile.mkdtemp()
|
|
j = lambda *p: os.path.join(self.t, *p)
|
|
self.pod, self.pbs, self.state, self.text = j("pod"), j("pbs"), j("state"), j("textfile")
|
|
self.conf, self.tokens, self.dumps, self.secrets, self.bin = j("conf"), j("tokens"), j("dumps"), j("secrets"), j("bin")
|
|
for d in (self.pbs, self.state, self.text, self.conf, self.tokens, self.dumps, self.secrets, self.bin):
|
|
os.makedirs(d)
|
|
# the Gitea pod: two real repositories with a commit each, app.ini, the other paths
|
|
for name in ("felhom.eu", "felhom-agent"):
|
|
bare = j("pod", "git", "repositories", "admin", name + ".git")
|
|
os.makedirs(os.path.dirname(bare), exist_ok=True)
|
|
git("init", "-q", "--bare", bare)
|
|
work = j("work-" + name)
|
|
git("init", "-q", work)
|
|
open(os.path.join(work, "README"), "w").write(name + "\n")
|
|
git("add", "README", cwd=work); git("commit", "-q", "-m", "c", cwd=work)
|
|
git("push", "-q", bare, "HEAD:refs/heads/main", cwd=work)
|
|
for p in ("git/lfs", "gitea/conf", "gitea/attachments", "gitea/avatars", "gitea/repo-avatars", "gitea/jwt", "gitea/packages"):
|
|
os.makedirs(j("pod", p), exist_ok=True)
|
|
open(j("pod", "gitea/conf/app.ini"), "w").write("[database]\nDB_TYPE = postgres\n")
|
|
open(j("pod", "gitea/packages/blob"), "w").write("registry - must not be copied\n")
|
|
# a complete dump (00:00 local) and the nightly secrets export
|
|
self.now = int(time.time())
|
|
self.add_dump("20261008-220001", self.now - 1200)
|
|
self.add_secrets("20261008_031011", self.now - 21 * 3600)
|
|
for n in ("token-push", "token-restore"):
|
|
open(os.path.join(self.tokens, n), "w").write("x")
|
|
open(os.path.join(self.conf, "enc.key"), "w").write("{}")
|
|
open(os.path.join(self.conf, "env"), "w").write(
|
|
"PBS_REPOSITORY_PUSH='u!push@h:1:s'\nPBS_REPOSITORY_RESTORE='u!restore@h:1:s'\nPBS_FINGERPRINT='aa'\n")
|
|
self.vw = j("vw")
|
|
os.makedirs(self.vw)
|
|
make_vw_db(os.path.join(self.vw, "db.sqlite3"))
|
|
open(os.path.join(self.vw, "rsa_key.pem"), "w").write("-----TEST KEY-----\n")
|
|
fakes = [("kubectl", FAKE_KUBECTL), ("proxmox-backup-client", FAKE_PBS), ("pg_restore", FAKE_PG_RESTORE)]
|
|
if not REAL_SQLITE3:
|
|
fakes.append(("sqlite3", FAKE_SQLITE3))
|
|
for name, body in fakes:
|
|
p = os.path.join(self.bin, name)
|
|
open(p, "w").write(body); os.chmod(p, 0o755)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.t, ignore_errors=True)
|
|
|
|
def add_dump(self, name, mtime, complete=True, magic=b"PGDMP"):
|
|
d = os.path.join(self.dumps, name); os.makedirs(d)
|
|
open(os.path.join(d, "gitea.dump"), "wb").write(magic + b"\x01dump")
|
|
open(os.path.join(d, "globals.sql"), "w").write("-- roles\n")
|
|
if complete:
|
|
open(os.path.join(d, "SUCCESS"), "w").close(); os.utime(os.path.join(d, "SUCCESS"), (mtime, mtime))
|
|
|
|
def add_secrets(self, stamp, mtime):
|
|
for k, ext in (("secrets", "yaml.gpg"), ("configmaps", "yaml.gpg"), ("by-namespace", "tar.gz.gpg")):
|
|
p = os.path.join(self.secrets, "%s-%s.%s" % (k, stamp, ext))
|
|
open(p, "wb").write(b"gpg-" + k.encode()); os.utime(p, (mtime, mtime))
|
|
|
|
def env(self, **kw):
|
|
e = dict(os.environ, PATH=self.bin + ":" + os.environ["PATH"], FAKE_POD_DATA=self.pod, FAKE_PBS_DIR=self.pbs,
|
|
FAKE_STATE=self.t, FAKE_VW_DATA=self.vw, FELHOM_DXOFF_CONF=self.conf, FELHOM_DXOFF_TOKENS=self.tokens,
|
|
FELHOM_DXOFF_STATE=self.state, FELHOM_DXOFF_TEXTFILE_DIR=self.text, FELHOM_DXOFF_DUMPS=self.dumps,
|
|
FELHOM_DXOFF_SECRETS=self.secrets)
|
|
e.update({k: str(v) for k, v in kw.items()})
|
|
return e
|
|
|
|
def push(self, **kw):
|
|
return subprocess.run([PUSH], env=self.env(**kw), capture_output=True, text=True)
|
|
|
|
def restore(self, **kw):
|
|
return subprocess.run([RESTORE], env=self.env(**kw), capture_output=True, text=True)
|
|
|
|
def pushed(self):
|
|
d = os.path.join(self.pbs, "snaps")
|
|
return sorted(os.listdir(d)) if os.path.isdir(d) else []
|
|
|
|
def signal(self, name="felhom_dooplex_offsite.prom"):
|
|
return os.path.exists(os.path.join(self.text, name))
|
|
|
|
|
|
class Push(Base):
|
|
def test_happy_path_pushes_everything_but_the_registry_and_writes_the_signal(self):
|
|
r = self.push()
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertEqual(len(self.pushed()), 1)
|
|
snap = os.path.join(self.pbs, "snaps", self.pushed()[0])
|
|
self.assertTrue(os.path.isfile(os.path.join(snap, "gitea/git/repositories/admin/felhom.eu.git/HEAD")))
|
|
self.assertTrue(os.path.isfile(os.path.join(snap, "gitea/gitea/conf/app.ini")))
|
|
self.assertFalse(os.path.exists(os.path.join(snap, "gitea/gitea/packages")), "the registry must stay out")
|
|
self.assertEqual(open(os.path.join(snap, "db/DUMP-FOLDER")).read().strip(), "20261008-220001")
|
|
self.assertEqual(len(os.listdir(os.path.join(snap, "secrets"))), 3)
|
|
self.assertEqual(open(os.path.join(snap, "REPOS")).read().strip(), "2")
|
|
self.assertTrue(self.signal())
|
|
call = json.loads(open(os.path.join(self.pbs, "calls.log")).readline())
|
|
self.assertIn("--crypt-mode", call["argv"]); self.assertIn("encrypt", call["argv"])
|
|
self.assertTrue(call["pw"].endswith("token-push"))
|
|
self.assertFalse(os.path.exists(os.path.join(self.state, "stage")), "the stage must be removed")
|
|
|
|
def test_newest_incomplete_dump_is_skipped_for_the_complete_one(self):
|
|
self.add_dump("20261009-040001", self.now - 60, complete=False)
|
|
r = self.push()
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
snap = os.path.join(self.pbs, "snaps", self.pushed()[0])
|
|
self.assertEqual(open(os.path.join(snap, "db/DUMP-FOLDER")).read().strip(), "20261008-220001")
|
|
|
|
def test_stale_dump_refuses(self):
|
|
shutil.rmtree(self.dumps); os.makedirs(self.dumps)
|
|
self.add_dump("20261008-040001", self.now - 8 * 3600)
|
|
r = self.push()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("dump CronJob stopped", r.stderr)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_no_complete_dump_refuses(self):
|
|
shutil.rmtree(self.dumps); os.makedirs(self.dumps)
|
|
self.add_dump("20261009-040001", self.now, complete=False)
|
|
r = self.push()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_one_failed_file_copy_is_retried(self):
|
|
r = self.push(FAKE_TAR_FAILS=1, FELHOM_DXOFF_RETRY_SLEEP=0)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertTrue(self.signal())
|
|
|
|
def test_two_failed_file_copies_refuse(self):
|
|
r = self.push(FAKE_TAR_FAILS=2, FELHOM_DXOFF_RETRY_SLEEP=0)
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("failed twice", r.stderr)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_fewer_repositories_than_the_pod_lists_refuses(self):
|
|
r = self.push(FAKE_EXTRA_REPOS="ghost.git")
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("the pod lists 3", r.stderr)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_missing_app_ini_refuses(self):
|
|
os.remove(os.path.join(self.pod, "gitea/conf/app.ini"))
|
|
r = self.push()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_stale_secrets_refuse(self):
|
|
shutil.rmtree(self.secrets); os.makedirs(self.secrets)
|
|
self.add_secrets("20261007_031011", self.now - 31 * 3600)
|
|
r = self.push()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("secrets export", r.stderr)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_a_link_in_the_pods_archive_refuses(self):
|
|
os.symlink("/etc/passwd", os.path.join(self.pod, "gitea/avatars/evil"))
|
|
r = self.push()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("link(s)", r.stderr)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_vaultwarden_is_copied_and_its_backup_file_removed(self):
|
|
r = self.push()
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
snap = os.path.join(self.pbs, "snaps", self.pushed()[0])
|
|
self.assertTrue(os.path.isfile(os.path.join(snap, "vaultwarden/db.sqlite3")))
|
|
self.assertTrue(os.path.isfile(os.path.join(snap, "vaultwarden/rsa_key.pem")))
|
|
self.assertEqual(open(os.path.join(snap, "vaultwarden/USERS")).read().strip(), "1")
|
|
self.assertFalse(os.path.exists(os.path.join(self.vw, "db_20261009_094401.sqlite3")), "the backup file must be removed")
|
|
self.assertEqual(open(os.path.join(self.t, "vw-removed")).read().strip(), "/data/db_20261009_094401.sqlite3")
|
|
self.assertIn("vaultwarden_users 1", open(os.path.join(self.text, "felhom_dooplex_offsite.prom")).read())
|
|
|
|
def test_vaultwarden_backup_failing_refuses(self):
|
|
r = self.push(FAKE_VW_BACKUP_FAIL=1)
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("vaultwarden backup", r.stderr)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_vaultwarden_with_no_user_refuses(self):
|
|
os.remove(os.path.join(self.vw, "db.sqlite3"))
|
|
make_vw_db(os.path.join(self.vw, "db.sqlite3"), users=0)
|
|
r = self.push()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("no user", r.stderr)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
self.assertFalse(os.path.exists(os.path.join(self.vw, "db_20261009_094401.sqlite3")), "removed even on refusal")
|
|
|
|
def test_a_corrupt_vaultwarden_copy_refuses(self):
|
|
open(os.path.join(self.vw, "db.sqlite3"), "wb").write(b"SQLite format 3\x00" + b"\xff" * 4096)
|
|
r = self.push()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("Vaultwarden integrity_check", r.stderr)
|
|
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
|
|
|
|
def test_failed_push_writes_no_signal(self):
|
|
r = self.push(FAKE_PBS_FAIL=1)
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertFalse(self.signal())
|
|
|
|
|
|
class RestoreTest(Base):
|
|
def pushed_copy(self, **kw):
|
|
r = self.push(FAKE_PBS_TIME=self.now - 3600, **kw)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
return os.path.join(self.pbs, "snaps", self.pushed()[0])
|
|
|
|
def test_happy_path_writes_the_signal_with_the_readonly_token(self):
|
|
self.pushed_copy()
|
|
r = self.restore()
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertIn("2 repositories pass git fsck", r.stdout)
|
|
self.assertTrue(self.signal("felhom_dooplex_offsite_restore.prom"))
|
|
calls = [json.loads(x) for x in open(os.path.join(self.pbs, "calls.log"))]
|
|
self.assertTrue(all(c["pw"].endswith("token-restore") for c in calls if c["argv"][0] != "backup"))
|
|
|
|
def test_a_changed_file_fails_the_manifest(self):
|
|
snap = self.pushed_copy()
|
|
open(os.path.join(snap, "secrets", sorted(os.listdir(os.path.join(snap, "secrets")))[0]), "ab").write(b"x")
|
|
r = self.restore()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("MANIFEST", r.stderr)
|
|
self.assertFalse(self.signal("felhom_dooplex_offsite_restore.prom"))
|
|
|
|
def test_a_broken_repository_fails_git_fsck(self):
|
|
snap = self.pushed_copy()
|
|
repo = os.path.join(snap, "gitea/git/repositories/admin/felhom-agent.git")
|
|
objs = [os.path.join(dp, f) for dp, _, fs in os.walk(os.path.join(repo, "objects")) for f in fs
|
|
if len(os.path.basename(dp)) == 2]
|
|
os.remove(objs[0])
|
|
# keep the manifest honest about the removal, so ONLY git fsck can catch it
|
|
man = os.path.join(snap, "MANIFEST.sha256")
|
|
rel = "./" + os.path.relpath(objs[0], snap)
|
|
kept = [l for l in open(man).readlines() if not l.rstrip().endswith(rel)]
|
|
open(man, "w").writelines(kept)
|
|
r = self.restore()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("git fsck", r.stderr)
|
|
self.assertFalse(self.signal("felhom_dooplex_offsite_restore.prom"))
|
|
|
|
def test_git_never_reads_a_repositorys_own_config(self):
|
|
# The pod's `config` files are untrusted input to a root git. A config git would refuse to open
|
|
# (repositoryformatversion 99) proves git never read it: the old `git -C <repo> fsck` failed here (red-proof).
|
|
snap = self.pushed_copy()
|
|
cfg = os.path.join(snap, "gitea/git/repositories/admin/felhom.eu.git/config")
|
|
open(cfg, "w").write("[core]\n\trepositoryformatversion = 99\n\tbare = true\n\tfsmonitor = touch /nonexistent/x\n")
|
|
man = os.path.join(snap, "MANIFEST.sha256")
|
|
kept = [l for l in open(man).readlines() if not l.rstrip().endswith("felhom.eu.git/config")]
|
|
open(man, "w").writelines(kept)
|
|
r = self.restore()
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
|
|
def test_restore_reports_vaultwarden_rows_never_contents(self):
|
|
self.pushed_copy()
|
|
r = self.restore()
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertIn("Vaultwarden 1 user(s) / 3 item(s)", r.stdout)
|
|
self.assertNotIn("encrypted", r.stdout + r.stderr)
|
|
|
|
def test_restore_without_vaultwarden_fails(self):
|
|
snap = self.pushed_copy()
|
|
os.remove(os.path.join(snap, "vaultwarden/db.sqlite3"))
|
|
man = os.path.join(snap, "MANIFEST.sha256")
|
|
kept = [l for l in open(man).readlines() if "vaultwarden/db.sqlite3" not in l]
|
|
open(man, "w").writelines(kept)
|
|
r = self.restore()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("no Vaultwarden database", r.stderr)
|
|
self.assertFalse(self.signal("felhom_dooplex_offsite_restore.prom"))
|
|
|
|
def test_an_unreadable_dump_fails(self):
|
|
shutil.rmtree(self.dumps); os.makedirs(self.dumps)
|
|
self.add_dump("20261008-220001", self.now - 1200, magic=b"XXXXX")
|
|
self.pushed_copy()
|
|
r = self.restore()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("pg_restore", r.stderr)
|
|
self.assertFalse(self.signal("felhom_dooplex_offsite_restore.prom"))
|
|
|
|
def test_an_old_copy_fails(self):
|
|
r = self.push(FAKE_PBS_TIME=self.now - 51 * 3600)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
r = self.restore()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertIn("h old", r.stderr)
|
|
self.assertFalse(self.signal("felhom_dooplex_offsite_restore.prom"))
|
|
|
|
def test_no_copy_fails(self):
|
|
r = self.restore()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertFalse(self.signal("felhom_dooplex_offsite_restore.prom"))
|
|
|
|
|
|
class FailMail(unittest.TestCase):
|
|
def test_failmail_calls_notify_failure_with_the_unit_name(self):
|
|
t = tempfile.mkdtemp()
|
|
try:
|
|
cfg = os.path.join(t, "cfg.sh"); out = os.path.join(t, "out")
|
|
# Like the real backup-config.sh, it reads a variable that is unset (NOTIFY_WEBHOOK_URL is commented out
|
|
# there): the 2026-10-09 dry run found the mail script under `set -u` dying on exactly that line.
|
|
open(cfg, "w").write('notify_failure() { if [ -n "${NOTIFY_WEBHOOK_URL_UNSET_IN_TEST}" ]; then :; fi; '
|
|
'echo "$1" > %s; return 0; }\n' % out)
|
|
r = subprocess.run([FAILMAIL, "felhom-dooplex-offsite.service"], env=dict(os.environ, FELHOM_FAILMAIL_CONFIG=cfg),
|
|
capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertIn("felhom-dooplex-offsite.service failed", open(out).read())
|
|
finally:
|
|
shutil.rmtree(t)
|
|
|
|
def test_every_backup_unit_names_the_failmail(self):
|
|
units = [os.path.join(HERE, u) for u in ("felhom-dooplex-offsite.service", "felhom-dooplex-offsite-restore-test.service")]
|
|
units += [os.path.join(HERE, "..", "hub-db-backup", u) for u in ("felhom-hub-db-backup.service", "felhom-hub-db-restore-test.service")]
|
|
for u in units:
|
|
self.assertIn("OnFailure=felhom-backup-failmail@%n.service", open(u).read(), u)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|