PacketStation/tests/server/test_backup_scripts.py

"""The VM backup and restore scripts (scripts/backup) and the VM kit's shell parts (deploy/vm).

No PostgreSQL, no systemd and no root: fake `pg_dump`, `pg_dumpall`, `pg_restore` and `psql` sit in a
folder given as PG_BINDIR, a fake user switch stands in for setpriv (PG_AS), and a fake `systemd-run`
prints what ps-run would start. Every fake records its calls. Runs on Linux (CI, docker-1) and on
Windows when a Git Bash or Cygwin bash with cygpath is installed; skipped otherwise.
"""

from __future__ import annotations

import hashlib
import os
import re
import shutil
import stat
import subprocess
import sys
from pathlib import Path

import pytest

REPO = Path(__file__).resolve().parents[2]
BACKUP = REPO / "scripts" / "backup" / "ps-backup.sh"
RESTORE = REPO / "scripts" / "backup" / "ps-restore.sh"
VM = REPO / "deploy" / "vm"
UNITS = VM / "systemd"
WINDOWS = sys.platform == "win32"

# One script, written under each tool's name. It dispatches on its own name.
FAKE_PG = r"""#!/usr/bin/env bash
tool="$(basename "$0")"
printf '%s %s\n' "$tool" "$*" >> "$FAKE_LOG"
case "$tool" in
  asuser)
    exec "$@" ;;
  pg_dump)
    db=""
    while [[ $# -gt 0 ]]; do [[ "$1" == "-d" ]] && db="$2"; shift; done
    [[ "$db" == "${FAKE_DUMP_FAIL_DB:-none}" ]] && { echo "pg_dump: error: fake" >&2; exit 1; }
    [[ -n "${FAKE_DUMP_EMPTY:-}" ]] && exit 0
    printf 'PGDMP fake dump of %s\n' "$db"
    exit 0 ;;
  pg_dumpall)
    printf 'CREATE ROLE ps_app;\nALTER ROLE ps_app WITH LOGIN;\n'
    exit 0 ;;
  pg_restore)
    input="$(cat)"
    [[ -n "${FAKE_RESTORE_FAIL:-}" ]] && { echo "pg_restore: error: input file is too short" >&2; exit 1; }
    [[ "$input" == PGDMP* ]] || { echo "pg_restore: error: not a dump" >&2; exit 1; }
    if [[ " $* " == *" --list "* ]]; then
      printf ';\n; Archive created at 2026-10-21\n;\n215; 1259 16390 TABLE public packets ps_migrate\n'
      printf '3801; 0 16390 TABLE DATA public packets ps_migrate\n'
    fi
    exit 0 ;;
  psql)
    sql="$(cat)"
    printf '%s\n' "$sql" | sed 's/^/  sql: /' >> "$FAKE_LOG"
    case "$sql" in
      *"FROM pg_database"*) echo "${FAKE_DB_EXISTS:-0}" ;;
      *"FROM pg_roles"*) echo "${FAKE_ROLE_EXISTS:-1}" ;;
      *"FROM assets"*) cat "$FAKE_ASSET_ROWS" ;;
      *"relkind"*) printf 'packets\t3\n' ;;
      *"CREATE ROLE"*) printf '%s\n' "${FAKE_GLOBALS_ERR:-}" >&2 ;;
    esac
    exit 0 ;;
esac
echo "fake: unexpected tool $tool" >&2
exit 99
"""

FAKE_SYSTEMD_RUN = r"""#!/usr/bin/env bash
for a in "$@"; do printf '%s\n' "$a"; done
"""

FAKE_PACKETSTATION = r"""#!/usr/bin/env bash
printf 'args=%s\n' "$*"
for v in PG_APP_PASSWORD_FILE PG_MIGRATE_PASSWORD_FILE VISUAL_PASSWORD_FILE MFILES_PASSWORD_FILE \
         SESSION_KEY_FILE SEAL_KEY_FILE STATIC_DIR PS_BUILD WORKER_BEAT_FILE TZ; do
  printf '%s=%s\n' "$v" "${!v-<unset>}"
done
printf 'MFILES_PASSWORD=%s\n' "${MFILES_PASSWORD-<unset>}"
"""


def _find_bash() -> str | None:
    if not WINDOWS:
        return shutil.which("bash")
    candidates = [
        r"C:\Program Files\Git\bin\bash.exe",
        r"C:\Program Files\Git\usr\bin\bash.exe",
        shutil.which("bash") or "",
    ]
    for c in candidates:
        if not c or not Path(c).exists() or "system32" in c.lower():
            continue
        try:
            out = subprocess.run(
                [c, "-c", "cygpath -u 'C:\\' && command -v find stat mktemp sha256sum"],
                capture_output=True,
                text=True,
                timeout=30,
            )
        except OSError:
            continue
        if out.returncode == 0:
            return c
    return None


BASH = _find_bash()
needs_bash = pytest.mark.skipif(BASH is None, reason="no usable bash (Linux bash, or Git Bash with cygpath)")


def _posix(p: Path) -> str:
    if not WINDOWS:
        return str(p)
    assert BASH is not None
    out = subprocess.run([BASH, "-c", 'cygpath -u "$1"', "_", str(p)], capture_output=True, text=True, check=True)
    return out.stdout.strip()


def _write_exec(path: Path, text: str) -> None:
    path.write_text(text, encoding="utf-8", newline="\n")
    path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)


@pytest.fixture()
def fake(tmp_path: Path) -> dict[str, object]:
    bindir = tmp_path / "pgbin"
    bindir.mkdir()
    for tool in ("asuser", "pg_dump", "pg_dumpall", "pg_restore", "psql"):
        _write_exec(bindir / tool, FAKE_PG)
    log = tmp_path / "calls.log"
    log.touch()
    env = {k: v for k, v in os.environ.items() if not k.startswith("FAKE_")}
    env["FAKE_LOG"] = _posix(log)
    env["PG_BINDIR"] = _posix(bindir)
    env["PG_AS"] = _posix(bindir / "asuser")
    for k in ("KEEP", "PS_BACKUP_CONFIG"):
        env.pop(k, None)
    return {"env": env, "log": log, "tmp": tmp_path}


def _run(script: Path, fake: dict[str, object], *args: str, **extra_env: str) -> subprocess.CompletedProcess[str]:
    env = dict(fake["env"])  # type: ignore[call-overload]
    env.update(extra_env)
    assert BASH is not None
    return subprocess.run([BASH, _posix(script), *args], capture_output=True, text=True, env=env, timeout=120)


def run_backup(fake: dict[str, object], *args: str, **extra_env: str) -> subprocess.CompletedProcess[str]:
    return _run(BACKUP, fake, *args, **extra_env)


def run_restore(fake: dict[str, object], *args: str, **extra_env: str) -> subprocess.CompletedProcess[str]:
    return _run(RESTORE, fake, *args, **extra_env)


def stack_args(name: str, folder: Path, database: str = "packetstation") -> list[str]:
    return ["--stack", name, database, _posix(folder)]


def dumps(folder: Path) -> list[str]:
    return sorted(p.name for p in folder.glob("ps-*.dump"))


def no_partials(folder: Path) -> bool:
    return not [p for p in folder.iterdir() if p.name.endswith(".partial")]


MARKER = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z (ps-\d{8}T\d{6}Z\.dump)(.*)$")


# ---------------------------------------------------------------- ps-backup


@needs_bash
def test_success_writes_dump_globals_and_marker(fake: dict[str, object]) -> None:
    folder = fake["tmp"] / "backups"  # type: ignore[operator]
    res = run_backup(fake, *stack_args("test", folder))
    assert res.returncode == 0, res.stderr
    names = dumps(folder)
    assert len(names) == 1
    dump = folder / names[0]
    assert dump.read_text().startswith("PGDMP")
    globals_file = folder / names[0].replace(".dump", ".globals.sql")
    assert "CREATE ROLE" in globals_file.read_text()
    m = MARKER.match((folder / "last-success").read_text().strip())
    assert m and m.group(1) == names[0]
    assert not (folder / "last-failure").exists()
    assert no_partials(folder)
    if not WINDOWS:
        assert stat.S_IMODE(dump.stat().st_mode) == 0o600
        assert stat.S_IMODE(globals_file.stat().st_mode) == 0o600
        assert stat.S_IMODE((folder / "last-success").stat().st_mode) == 0o644
    calls = fake["log"].read_text()  # type: ignore[attr-defined]
    # pg_dump and pg_dumpall go through the user switch; pg_restore --list needs no database and no switch.
    assert re.search(r"^asuser \S*/pg_dump -d packetstation -Fc$", calls, re.M), calls
    assert re.search(r"^asuser \S*/pg_dumpall --globals-only --no-role-passwords$", calls, re.M), calls
    assert re.search(r"^pg_restore --list$", calls, re.M), calls
    assert "docker" not in calls


@needs_bash
def test_rotation_keeps_seven(fake: dict[str, object]) -> None:
    folder = fake["tmp"] / "backups"  # type: ignore[operator]
    folder.mkdir()
    old = [f"ps-202601{d:02d}T020000Z" for d in range(1, 9)]  # 8 older dumps
    for base in old:
        (folder / f"{base}.dump").write_text("PGDMP old")
        (folder / f"{base}.globals.sql").write_text("-- old")
    (folder / "ps-20251231T020000Z.globals.sql").write_text("-- orphan globals")
    res = run_backup(fake, *stack_args("test", folder))
    assert res.returncode == 0, res.stderr
    names = dumps(folder)
    assert len(names) == 7
    # The newest six old ones and today's stay; the two oldest go with their globals.
    assert names[:6] == [f"{b}.dump" for b in old[2:]]
    assert not (folder / f"{old[0]}.globals.sql").exists()
    assert not (folder / f"{old[1]}.globals.sql").exists()
    assert not (folder / "ps-20251231T020000Z.globals.sql").exists()
    assert sorted(p.name for p in folder.glob("*.globals.sql")) == [n.replace(".dump", ".globals.sql") for n in names]


@needs_bash
def test_keep_is_configurable(fake: dict[str, object]) -> None:
    folder = fake["tmp"] / "backups"  # type: ignore[operator]
    folder.mkdir()
    for d in range(1, 5):
        (folder / f"ps-202601{d:02d}T020000Z.dump").write_text("PGDMP old")
    res = run_backup(fake, "--keep", "2", *stack_args("test", folder))
    assert res.returncode == 0, res.stderr
    assert len(dumps(folder)) == 2
    assert run_backup(fake, "--keep", "0", *stack_args("test", folder)).returncode == 2


@needs_bash
def test_failed_restore_list_writes_last_failure(fake: dict[str, object]) -> None:
    folder = fake["tmp"] / "backups"  # type: ignore[operator]
    folder.mkdir()
    (folder / "ps-20260101T020000Z.dump").write_text("PGDMP old")
    (folder / "last-success").write_text("2026-01-01T02:00:05Z ps-20260101T020000Z.dump\n")
    res = run_backup(fake, *stack_args("test", folder), FAKE_RESTORE_FAIL="1")
    assert res.returncode != 0
    failure = (folder / "last-failure").read_text().strip()
    m = MARKER.match(failure)
    assert m, failure
    assert "pg_restore --list" in m.group(2)
    # Nothing new kept, nothing half-written, the old success left alone.
    assert dumps(folder) == ["ps-20260101T020000Z.dump"]
    assert not list(folder.glob("ps-*.globals.sql"))
    assert no_partials(folder)
    assert (folder / "last-success").read_text().startswith("2026-01-01")
    assert "FAILED" in res.stderr


@needs_bash
def test_failed_pg_dump_and_empty_dump(fake: dict[str, object]) -> None:
    folder = fake["tmp"] / "backups"  # type: ignore[operator]
    res = run_backup(fake, *stack_args("test", folder, database="nosuchdb"), FAKE_DUMP_FAIL_DB="nosuchdb")
    assert res.returncode != 0
    assert "pg_dump failed" in (folder / "last-failure").read_text()
    assert dumps(folder) == []
    assert no_partials(folder)

    res = run_backup(fake, *stack_args("test", folder), FAKE_DUMP_EMPTY="1")
    assert res.returncode != 0
    assert "pg_dump wrote nothing" in (folder / "last-failure").read_text()
    assert not (folder / "last-success").exists()


@needs_bash
def test_bad_database_name_is_refused(fake: dict[str, object]) -> None:
    folder = fake["tmp"] / "backups"  # type: ignore[operator]
    res = run_backup(fake, *stack_args("test", folder, database="x;rm -rf"))
    assert res.returncode != 0
    assert "not a plain name" in (folder / "last-failure").read_text()
    assert "pg_dump" not in fake["log"].read_text()  # type: ignore[attr-defined]


@needs_bash
def test_one_stack_fails_the_other_still_runs(fake: dict[str, object]) -> None:
    good = fake["tmp"] / "good"  # type: ignore[operator]
    bad = fake["tmp"] / "bad"  # type: ignore[operator]
    res = run_backup(
        fake,
        *stack_args("bad", bad, database="baddb"),
        *stack_args("good", good),
        FAKE_DUMP_FAIL_DB="baddb",
    )
    assert res.returncode == 1
    assert (bad / "last-failure").exists() and not (bad / "last-success").exists()
    assert (good / "last-success").exists() and not (good / "last-failure").exists()
    assert len(dumps(good)) == 1


@needs_bash
def test_config_file_and_stack_selection(fake: dict[str, object]) -> None:
    tmp: Path = fake["tmp"]  # type: ignore[assignment]
    a, b = tmp / "a", tmp / "b"
    conf = tmp / "backup.conf"
    conf.write_text(
        f"# name database folder\nalpha  packetstation  {_posix(a)}\n\nbeta   packetstation  {_posix(b)}\n",
        encoding="utf-8",
        newline="\n",
    )
    res = run_backup(fake, "--config", _posix(conf), "beta")
    assert res.returncode == 0, res.stderr
    assert not a.exists()
    assert len(dumps(b)) == 1

    res = run_backup(fake, "--config", _posix(conf), "gamma")
    assert res.returncode == 2

    res = run_backup(fake, "--config", _posix(tmp / "missing.conf"))
    assert res.returncode == 2

    # The old docker-1 format (with a container column) is refused, not misread.
    conf.write_text(f"test project:packetstation-test packetstation {_posix(a)}\n", encoding="utf-8", newline="\n")
    res = run_backup(fake, "--config", _posix(conf))
    assert res.returncode == 2
    assert "name database folder" in res.stderr


# ---------------------------------------------------------------- ps-restore


@needs_bash
def test_restore_into_a_fresh_database(fake: dict[str, object]) -> None:
    tmp: Path = fake["tmp"]  # type: ignore[assignment]
    dump = tmp / "ps-20261030T080000Z.dump"
    dump.write_text("PGDMP fake")
    res = run_restore(fake, "--dump", _posix(dump), "--db", "ps_restored")
    assert res.returncode == 0, res.stderr
    calls = fake["log"].read_text()  # type: ignore[attr-defined]
    assert "CREATE DATABASE %I OWNER %I ENCODING %L LOCALE %L TEMPLATE template0" in calls
    assert "-v name=ps_restored -v owner=ps_migrate -v loc=en_US.UTF-8" in calls
    assert "GRANT CONNECT, TEMPORARY" in calls
    assert re.search(r"^asuser \S*/pg_restore -d ps_restored --exit-on-error --single-transaction$", calls, re.M)
    assert "packets\t3" in res.stdout
    assert "DROP DATABASE" not in calls


@needs_bash
def test_restore_guards(fake: dict[str, object]) -> None:
    tmp: Path = fake["tmp"]  # type: ignore[assignment]
    dump = tmp / "ps-1.dump"
    dump.write_text("PGDMP fake")

    res = run_restore(fake, "--dump", _posix(dump), "--db", "ps_restored", FAKE_DB_EXISTS="1")
    assert res.returncode == 1
    assert "exists. Give --replace" in res.stderr

    res = run_restore(fake, "--dump", _posix(dump), "--db", "packetstation", "--replace", FAKE_DB_EXISTS="1")
    assert res.returncode == 1
    assert "refused for packetstation" in res.stderr

    res = run_restore(fake, "--dump", _posix(dump), "--db", "Bad-Name")
    assert res.returncode == 1

    hashed = tmp / "ps-1.globals.sql"
    hashed.write_text("CREATE ROLE ps_app;\nALTER ROLE ps_app WITH LOGIN PASSWORD 'SCRAM-SHA-256$4096:abc';\n")
    res = run_restore(fake, "--globals", _posix(hashed))
    assert res.returncode == 1
    assert "password hashes" in res.stderr

    res = run_restore(fake)
    assert res.returncode == 1
    assert "nothing to do" in res.stderr


@needs_bash
def test_restore_replace_drops_first(fake: dict[str, object]) -> None:
    dump = fake["tmp"] / "ps-1.dump"  # type: ignore[operator]
    dump.write_text("PGDMP fake")
    res = run_restore(fake, "--dump", _posix(dump), "--db", "ps_restored", "--replace", FAKE_DB_EXISTS="1")
    assert res.returncode == 0, res.stderr
    calls = fake["log"].read_text()  # type: ignore[attr-defined]
    assert calls.index("DROP DATABASE") < calls.index("CREATE DATABASE")


@needs_bash
def test_restore_globals_skips_existing_roles(fake: dict[str, object]) -> None:
    g = fake["tmp"] / "ps-1.globals.sql"  # type: ignore[operator]
    g.write_text("CREATE ROLE ps_app;\nALTER ROLE ps_app WITH LOGIN;\n")
    res = run_restore(fake, "--globals", _posix(g), FAKE_GLOBALS_ERR='ERROR:  role "ps_app" already exists')
    assert res.returncode == 0, res.stderr
    assert "1 statements skipped" in res.stderr
    res = run_restore(fake, "--globals", _posix(g), FAKE_GLOBALS_ERR="ERROR:  syntax error")
    assert res.returncode == 1


@needs_bash
def test_restore_assets_never_overwrite_and_verify(fake: dict[str, object]) -> None:
    tmp: Path = fake["tmp"]  # type: ignore[assignment]
    src = tmp / "src"
    (src / "ab").mkdir(parents=True)
    (src / "ab" / "one").write_bytes(b"one")
    (src / "ab" / "two").write_bytes(b"two")
    dst = tmp / "assets"
    (dst / "ab").mkdir(parents=True)
    (dst / "ab" / "two").write_bytes(b"kept")  # an existing file is never replaced
    # 0:0 so a run as root (CI containers) has an owner that exists; as a normal user there is no chown.
    res = run_restore(fake, "--assets", _posix(src), "--assets-dir", _posix(dst), "--assets-owner", "0:0")
    assert res.returncode == 0, res.stderr
    assert (dst / "ab" / "one").read_bytes() == b"one"
    assert (dst / "ab" / "two").read_bytes() == b"kept"

    def rows(files: dict[str, bytes]) -> str:
        return "".join(f"{hashlib.sha256(v).hexdigest()}  {k}\n" for k, v in sorted(files.items()))

    rows_file = tmp / "rows"
    rows_file.write_text(rows({"ab/one": b"one", "ab/two": b"kept"}), newline="\n")
    args = ("--verify-assets", "--db", "ps_restored", "--assets-dir", _posix(dst))
    res = run_restore(fake, *args, FAKE_ASSET_ROWS=_posix(rows_file))
    assert res.returncode == 0, res.stderr
    assert "2 rows, 2 match, 0 differ, 0 missing, 0 files with no row" in res.stderr

    (dst / "ab" / "stray").write_bytes(b"x")
    rows_file.write_text(rows({"ab/one": b"one", "ab/two": b"two", "ab/gone": b"g"}), newline="\n")
    res = run_restore(fake, *args, FAKE_ASSET_ROWS=_posix(rows_file))
    assert res.returncode == 1
    assert "3 rows, 1 match, 1 differ, 1 missing, 1 files with no row" in res.stderr
    assert "ORPHAN: ab/stray" in res.stderr


# ---------------------------------------------------------------- deploy/vm: launcher and ps-run


def _release(tmp: Path) -> Path:
    rel = tmp / "opt" / "sha-abc1234"
    (rel / "deploy" / "vm").mkdir(parents=True)
    (rel / "venv" / "bin").mkdir(parents=True)
    (rel / "static").mkdir()
    shutil.copy(VM / "ps-launch.sh", rel / "deploy" / "vm" / "ps-launch.sh")
    (rel / "deploy" / "vm" / "ps-launch.sh").chmod(0o755)
    _write_exec(rel / "venv" / "bin" / "packetstation", FAKE_PACKETSTATION)
    (rel / "BUILD").write_text("sha-abc1234\n", newline="\n")
    return rel


def _launch(rel: Path, env: dict[str, str], *args: str) -> dict[str, str]:
    assert BASH is not None
    base = {k: v for k, v in os.environ.items() if not k.endswith("_FILE") and k != "CREDENTIALS_DIRECTORY"}
    base.pop("MFILES_PASSWORD", None)
    base.update(env)
    res = subprocess.run(
        [BASH, _posix(rel / "deploy" / "vm" / "ps-launch.sh"), *args],
        capture_output=True,
        text=True,
        env=base,
        timeout=60,
    )
    assert res.returncode == 0, res.stderr
    return dict(line.split("=", 1) for line in res.stdout.splitlines())


@needs_bash
def test_launcher_maps_credentials_and_sets_release_paths(tmp_path: Path) -> None:
    rel = _release(tmp_path)
    creds = tmp_path / "creds"
    creds.mkdir()
    (creds / "postgres_app_password").write_text("pw-app")
    (creds / "mfileskit_password").write_text("mf-secret\r\n")
    (creds / "seal_key").write_text("k" * 32)
    out = _launch(
        rel,
        {
            "CREDENTIALS_DIRECTORY": _posix(creds),
            "VISUAL_PASSWORD_FILE": "/etc/packetstation/secrets/visualkit_password",
            "STATIC_DIR": "/elsewhere",
            "PS_BUILD": "wrong",
        },
        "worker",
    )
    c = _posix(creds)
    assert out["args"] == "worker"
    assert out["PG_APP_PASSWORD_FILE"] == f"{c}/postgres_app_password"
    assert out["SEAL_KEY_FILE"] == f"{c}/seal_key"
    assert out["MFILES_PASSWORD_FILE"] == f"{c}/mfileskit_password"
    # Not a credential of this unit: the env file's value stays and nothing is invented.
    assert out["VISUAL_PASSWORD_FILE"] == "/etc/packetstation/secrets/visualkit_password"
    assert out["PG_MIGRATE_PASSWORD_FILE"] == "<unset>"
    # Exported from the file, trailing CR LF dropped (the entrypoint drops a CR; the launcher reads with $(<)).
    assert out["MFILES_PASSWORD"] == "mf-secret"
    # The release wins over the env file for these.
    assert out["STATIC_DIR"] == f"{_posix(rel)}/static"
    assert out["PS_BUILD"] == "sha-abc1234"
    assert out["WORKER_BEAT_FILE"] == "/run/packetstation/worker.beat"
    assert out["TZ"] == "UTC"


@needs_bash
def test_launcher_without_credentials_defaults_to_app(tmp_path: Path) -> None:
    rel = _release(tmp_path)
    out = _launch(rel, {"MFILES_PASSWORD_FILE": "/nonexistent/mfileskit_password"})
    assert out["args"] == "app"
    assert out["MFILES_PASSWORD"] == "<unset>"
    assert out["PG_APP_PASSWORD_FILE"] == "<unset>"


@needs_bash
def test_ps_run_passes_env_file_credentials_and_sandbox(tmp_path: Path) -> None:
    rel = _release(tmp_path)
    secrets = tmp_path / "secrets"
    secrets.mkdir()
    (secrets / "postgres_app_password").write_text("x")
    (secrets / "ldap_bind_password").write_text("y")
    env_file = tmp_path / "packetstation.env"
    env_file.write_text("STACK=test\n", newline="\n")
    fake_run = tmp_path / "systemd-run"
    _write_exec(fake_run, FAKE_SYSTEMD_RUN)
    assert BASH is not None
    env = dict(os.environ)
    env.update(
        PS_RUN_ALLOW_NONROOT="1",
        SYSTEMD_RUN=_posix(fake_run),
        PS_ENV_FILE_PATH=_posix(env_file),
        PS_SECRETS_DIR=_posix(secrets),
        PS_CURRENT=_posix(rel),
    )
    res = subprocess.run(
        [BASH, _posix(VM / "ps-run.sh"), "doctor", "--skip", "visual"],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
    )
    assert res.returncode == 0, res.stderr
    argv = res.stdout.splitlines()
    s = _posix(secrets)
    assert argv[:5] == ["--wait", "--pipe", "--collect", "--quiet", "--service-type=exec"]
    assert "User=packetstation" in argv and "Group=packetstation" in argv
    assert f"EnvironmentFile={_posix(env_file)}" in argv
    assert "ProtectSystem=strict" in argv and "NoNewPrivileges=yes" in argv
    assert f"InaccessiblePaths=-{s}" in argv
    creds = [a for a in argv if a.startswith("LoadCredential=")]
    assert creds == [
        f"LoadCredential=postgres_app_password:{s}/postgres_app_password",
        f"LoadCredential=ldap_bind_password:{s}/ldap_bind_password",
    ]
    assert argv[-4:] == [f"{_posix(rel)}/deploy/vm/ps-launch.sh", "doctor", "--skip", "visual"]

    env.pop("PS_RUN_ALLOW_NONROOT")
    if sys.platform != "win32" and os.geteuid() != 0:
        res = subprocess.run([BASH, _posix(VM / "ps-run.sh"), "doctor"], capture_output=True, text=True, env=env)
        assert res.returncode == 2
        assert "run as root" in res.stderr


# ---------------------------------------------------------------- static checks


SCRIPTS = [BACKUP, RESTORE, *sorted(VM.glob("*.sh"))]


@needs_bash
@pytest.mark.parametrize("script", SCRIPTS, ids=lambda p: p.name)
def test_scripts_are_strict_and_parse(script: Path) -> None:
    assert BASH is not None
    text = script.read_text(encoding="utf-8")
    assert "\r" not in text
    assert text.startswith("#!/usr/bin/env bash\n")
    assert "set -euo pipefail" in text
    if script in (BACKUP, RESTORE):
        assert "umask 077" in text
    if script.name != "make-release.sh":
        assert "docker exec" not in text and "docker run" not in text
    res = subprocess.run([BASH, "-n", _posix(script)], capture_output=True, text=True)
    assert res.returncode == 0, res.stderr


Unit = dict[str, dict[str, list[str]]]


def _unit(name: str) -> Unit:
    """A unit file as {section: {key: [values]}}. systemd repeats keys (LoadCredential=, ReadWritePaths=)."""
    out: Unit = {}
    section = ""
    for raw in (UNITS / name).read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("[") and line.endswith("]"):
            section = line[1:-1]
            out.setdefault(section, {})
            continue
        key, _, value = line.partition("=")
        out[section].setdefault(key.strip(), []).append(value.strip())
    return out


def _values(u: Unit, section: str, key: str) -> list[str]:
    return u.get(section, {}).get(key, [])


LAUNCH = "/opt/packetstation/current/deploy/vm/ps-launch.sh"


@pytest.mark.parametrize(
    ("unit", "command", "creds"),
    [
        ("packetstation-migrate.service", "migrate", {"postgres_migrate_password", "postgres_app_password"}),
        (
            "packetstation-app.service",
            "app",
            {"postgres_app_password", "ldap_bind_password", "session_key", "seal_key"},
        ),
        (
            "packetstation-worker.service",
            "worker",
            {"postgres_app_password", "visualkit_password", "mfileskit_password", "ldap_bind_password", "seal_key"},
        ),
    ],
)
def test_service_units(unit: str, command: str, creds: set[str]) -> None:
    text = (UNITS / unit).read_text(encoding="utf-8")
    assert "\r" not in text
    cp = _unit(unit)
    svc = "Service"
    assert _values(cp, svc, "ExecStart") == [f"{LAUNCH} {command}"]
    assert _values(cp, svc, "User") == ["packetstation"]
    assert _values(cp, svc, "EnvironmentFile") == ["/etc/packetstation/packetstation.env"]
    for key, want in (
        ("NoNewPrivileges", "yes"),
        ("ProtectSystem", "strict"),
        ("ProtectHome", "yes"),
        ("PrivateTmp", "yes"),
        ("InaccessiblePaths", "/etc/packetstation/secrets"),
        ("CapabilityBoundingSet", ""),
    ):
        assert _values(cp, svc, key) == [want], key
    loaded = {v.split(":", 1)[0] for v in _values(cp, svc, "LoadCredential")}
    assert loaded == creds
    for v in _values(cp, svc, "LoadCredential"):
        name, path = v.split(":", 1)
        assert path == f"/etc/packetstation/secrets/{name}"
    rw = _values(cp, svc, "ReadWritePaths")
    if command == "migrate":
        assert rw == []
    else:
        # The app runs the archive check and the worker copies: both need the archive mount.
        assert rw == ["/var/lib/packetstation/assets", "-/mnt/packetstation-archive"]
        assert "packetstation-migrate.service" in _values(cp, "Unit", "Requires")[0]


def test_app_unit_needs_the_firewall_table() -> None:
    cp = _unit("packetstation-app.service")
    assert _values(cp, "Service", "ExecStartPre") == ["+/usr/sbin/nft list table inet packetstation"]
    fw = _unit("packetstation-firewall.service")
    assert _values(fw, "Service", "ExecStart") == ["/usr/sbin/nft -f /etc/packetstation/packetstation.nft"]


def test_launcher_knows_every_credential_the_units_load() -> None:
    launch = (VM / "ps-launch.sh").read_text(encoding="utf-8")
    run = (VM / "ps-run.sh").read_text(encoding="utf-8")
    names: set[str] = set()
    for unit in UNITS.glob("*.service"):
        names |= {v.split(":", 1)[0] for v in _values(_unit(unit.name), "Service", "LoadCredential")}
    for n in names:
        assert re.search(rf"(^|')({n}) [A-Z_]+_FILE'?$", launch, re.M), n
        assert n in run, n


@pytest.mark.parametrize("stack", ["prod", "test"])
def test_env_examples(stack: str) -> None:
    text = (VM / f"{stack}.env.example").read_text(encoding="utf-8")
    assert "\r" not in text
    values = dict(line.split("=", 1) for line in text.splitlines() if line and not line.startswith("#") and "=" in line)
    assert values["STACK"] == stack
    assert values["PG_HOST"] == "/var/run/postgresql"
    assert values["TRUSTED_PROXY_HOST"] == "192.168.101.12"
    assert values["ASSETS_DIR"] == "/var/lib/packetstation/assets"
    assert values["BACKUP_DIR"] == "/var/backups/packetstation"
    assert "#ARCHIVE_ROOT=/mnt/packetstation-archive" in text
    for k, v in values.items():
        if k.endswith("_FILE"):
            assert v.startswith("/etc/packetstation/secrets/"), k
        assert "PASSWORD=" not in f"{k}=", k
    for k in ("STATIC_DIR", "PS_BUILD", "WORKER_BEAT_FILE", "MFILES_CA_BUNDLE"):
        assert k not in values, k
    if stack == "prod":
        assert (values["VISUAL_SERVER"], values["VISUAL_DATABASE"]) == ("vsql.pesco.com", "PESCOINC")
        assert values["NOTIFY_DRY_RUN"] == "false"
    else:
        assert (values["VISUAL_SERVER"], values["VISUAL_DATABASE"]) == ("'vsql.pesco.com\\DEV'", "PESCOINCDEV")
        assert values["NOTIFY_DRY_RUN"] == "true"


def test_backup_units() -> None:
    svc = (REPO / "scripts" / "backup" / "packetstation-backup.service").read_text(encoding="utf-8")
    timer = (REPO / "scripts" / "backup" / "packetstation-backup.timer").read_text(encoding="utf-8")
    assert "ExecStart=/usr/local/sbin/ps-backup --config /etc/packetstation/backup.conf" in svc
    assert "docker" not in svc.lower()
    assert "OnCalendar=*-*-* 02:00:00 America/Denver" in timer
    assert "Persistent=true" in timer