restic-backup-docker/tests/sabotage.py

"""Self test of the suite: prove that it goes red when the backup is broken.

Run with `task test:sabotage`.

A throwaway copy of the tree gets a backup.sh whose `restic backup` call is a no-op, the
nastiest kind of regression (the script still reports success). The compat tests run
against an image built from that copy. The check passes only if the assertions about the
backup itself fail by name. Red for another reason (an image that cannot start, a
non-executable /bin/backup) proves nothing and is reported as INVALID.

Exit codes: 0 sabotage detected, 1 suite did not detect it, 2 check invalid or broken.
Tracked files are never modified.
"""

from __future__ import annotations

import os
import shutil
import signal
import subprocess
import sys
import tempfile
import uuid
import xml.etree.ElementTree as ET
from pathlib import Path

from helpers import raise_interrupt

ROOT = Path(__file__).resolve().parent.parent
COPIED = ("Dockerfile", "backup.sh", "check.sh", "entry.sh")
# Tests about the backup itself; each must fail on the sabotaged image.
EXPECTED_RED = (
    "test_one_snapshot_after_one_backup",
    "test_second_backup_adds_snapshot",
)


STEP_FAILED = "StepFailed"  # name of helpers.StepFailed in the JUnit report


class Invalid(Exception):
    """The check itself is broken; the suite result means nothing."""


def sabotage_tree(tmp: Path) -> None:
    for name in COPIED:
        shutil.copy2(ROOT / name, tmp / name)
    script = tmp / "backup.sh"
    lines = script.read_text().splitlines(keepends=True)
    patched = [
        ("true " + ln[len("restic backup ") :]) if ln.startswith("restic backup ") else ln
        for ln in lines
    ]
    if patched == lines:
        raise Invalid("sabotage patch did not apply, backup.sh changed shape")
    script.write_text("".join(patched))
    # Dockerfile COPY has no chmod: the image only works when the file is executable.
    script.chmod(script.stat().st_mode | 0o111)
    if not os.access(script, os.X_OK):
        raise Invalid("sabotaged backup.sh is not executable")


def red_tests(report: Path) -> tuple[set[str], set[str]]:
    """Names of tests with an assertion failure, and of tests with an error.

    Errors are setup errors and failed scenario steps (StepFailed): neither is an assertion
    about the behaviour under test.
    """
    try:
        cases = list(ET.parse(report).getroot().iter("testcase"))
    except ET.ParseError as exc:
        raise Invalid(f"pytest report is malformed or truncated: {exc}") from exc
    errors = {c.get("name", "") for c in cases if c.find("error") is not None}
    failures = set()
    for case in cases:
        failure = case.find("failure")
        if failure is None:
            continue
        if STEP_FAILED in failure.get("message") or "":
            errors.add(case.get("name", ""))
        else:
            failures.add(case.get("name", ""))
    return failures, errors


def main() -> int:
    tag = f"rbd-sabotage:{uuid.uuid4().hex[:8]}"
    tmp = Path(tempfile.mkdtemp(prefix="rbd-sabotage-"))
    try:
        sabotage_tree(tmp)
        print(f">>> building {tag} from a sabotaged copy")
        build = subprocess.run(["docker", "build", "--quiet", "-t", tag, str(tmp)])
        if build.returncode != 0:
            raise Invalid("image build failed")
        print(">>> running the compat tests against the sabotaged image (failure expected)")
        report = tmp / "report.xml"
        proc = subprocess.run(
            [
                sys.executable,
                "-m",
                "pytest",
                "tests/test_compat.py",
                "-p",
                "no:cacheprovider",
                f"--junitxml={report}",
            ],
            cwd=ROOT,
            env={**os.environ, "IMAGE": tag},
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
        )
        print(proc.stdout)
        if proc.returncode == 0:
            print(
                "SABOTAGE CHECK FAILED: the compat tests passed on a broken image", file=sys.stderr
            )
            return 1
        if "permission denied" in proc.stdout.lower():
            raise Invalid("log contains 'permission denied', suite failed for the wrong reason")
        if not report.is_file():
            raise Invalid(f"pytest produced no report (exit code {proc.returncode})")
        failures, errors = red_tests(report)
        broken = [name for name in EXPECTED_RED if name in errors]
        if broken:
            raise Invalid(f"expected tests errored or hit a failed scenario step instead of failing an assertion: {broken}")
        missing = [name for name in EXPECTED_RED if name not in failures]
        for name in missing:
            print(
                f"SABOTAGE CHECK FAILED: expected assertion did not fail: {name}", file=sys.stderr
            )
        if missing:
            return 1
        print("\nsabotage check ok: the compat tests detect a broken backup")
        return 0
    except Invalid as exc:
        print(f"SABOTAGE CHECK INVALID: {exc}", file=sys.stderr)
        return 2
    finally:
        shutil.rmtree(tmp, ignore_errors=True)
        subprocess.run(["docker", "rmi", "-f", tag], capture_output=True)


if __name__ == "__main__":
    signal.signal(signal.SIGTERM, raise_interrupt)
    sys.exit(main())