#!/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)
'''
# ssh-keygen stand-in (the CI runner may lack OpenSSH): `ssh-keygen -y -P '' -f <key>` prints the public line a
# test key file carries after "PUB:" — the same contract as the real tool (derive the public half from the private).
FAKE_SSH_KEYGEN = r'''#!/bin/sh
[ "$1" = "-y" ] || exit 97
f=""; while [ $# -gt 0 ]; do [ "$1" = "-f" ] && f=$2; shift; done
sed -n 's/^PUB://p' "$f" | grep . || { echo "Load key: invalid format" >&2; exit 255; }
'''
SIGNING_KEYS = ("felhom-op-operational", "felhom-rec-recovery", "felhom_op_ed25519")


def make_signing(d):
    os.makedirs(d, exist_ok=True)
    for i, k in enumerate(SIGNING_KEYS):
        pub = "ssh-ed25519 AAAATEST%d %s" % (i, k)
        open(os.path.join(d, k), "w").write("-----BEGIN TEST KEY-----\nPUB:%s\n" % pub)
        open(os.path.join(d, k + ".pub"), "w").write(pub + "\n")


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")
        self.signing = j("signing")
        make_signing(self.signing)
        fakes = [("kubectl", FAKE_KUBECTL), ("proxmox-backup-client", FAKE_PBS), ("pg_restore", FAKE_PG_RESTORE),
                 ("ssh-keygen", FAKE_SSH_KEYGEN)]
        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_SIGNING=self.signing, 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_signing_keys_ride_the_copy(self):
        r = self.push()
        self.assertEqual(r.returncode, 0, r.stderr)
        snap = os.path.join(self.pbs, "snaps", self.pushed()[0])
        for k in SIGNING_KEYS:
            self.assertTrue(os.path.isfile(os.path.join(snap, "signing", k)), k)
            self.assertTrue(os.path.isfile(os.path.join(snap, "signing", k + ".pub")), k)

    def test_a_missing_signing_key_refuses(self):
        os.remove(os.path.join(self.signing, "felhom-rec-recovery"))
        r = self.push()
        self.assertNotEqual(r.returncode, 0)
        self.assertIn("signing key file felhom-rec-recovery missing", 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_checks_every_signing_key_opens_and_matches(self):
        self.pushed_copy()
        r = self.restore()
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertIn("3 signing key(s) match", r.stdout)
        self.assertNotIn("BEGIN TEST KEY", r.stdout + r.stderr)

    def test_restore_refuses_a_signing_key_that_does_not_match(self):
        snap = self.pushed_copy()
        p = os.path.join(snap, "signing", "felhom-op-operational.pub")
        open(p, "w").write("ssh-ed25519 AAAAOTHER someone-else\n")
        man = os.path.join(snap, "MANIFEST.sha256")
        kept = [l for l in open(man).readlines() if not l.rstrip().endswith("felhom-op-operational.pub")]
        open(man, "w").writelines(kept)
        r = self.restore()
        self.assertNotEqual(r.returncode, 0)
        self.assertIn("does not match its public half", r.stderr)
        self.assertFalse(self.signal("felhom_dooplex_offsite_restore.prom"))

    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)