restic-backup-docker/tests/test_features.py

"""Feature-gated tests.

Each class is guarded by `@pytest.mark.feature(<gate>)`: it runs as soon as the built
image ships the feature (detected in conftest.py) and is skipped before that. A gate
listed in tests/required-features.txt turns the skip into a failure. A scenario step
that fails fails only the tests that assert on its result.

Scenario fixtures request `feature_gate_class` first, so a gated class never starts a
container on an image without the feature.
"""

from __future__ import annotations

from collections.abc import Iterator

import pytest
from conftest import LABELS
from helpers import Container, Scenario, run_once, running

STREAM_ENV = {
    "RESTIC_STREAM": "1",
    "RESTIC_STREAM_CMD": "echo hello-stream",
    "RESTIC_STREAM_FILENAME": "stream.txt",
}
BASH_HOOK = """#!/bin/bash
arr=(one two)
if [[ "${arr[1]}" == "two" ]]; then echo "BASH-HOOK-RAN"; fi
"""
MSMTP_CONFIG = "defaults\naccount default\nhost localhost\n"


# ---------------------------------------------------------------------------
# stream backup (PR #109)
# ---------------------------------------------------------------------------


@pytest.fixture(scope="class")
def stream_run(feature_gate_class, run_container) -> Iterator[Scenario]:
    with run_container("stream", env=STREAM_ENV) as c:
        running(c)
        run = Scenario()
        run.record("backup", lambda: c.exec("/bin/backup"))
        run.record("snapshots", c.snapshots)
        run.record("restore", lambda: c.restore_latest("/tmp/restore"))
        run.record("restored", lambda: c.exec("cat", "/tmp/restore/stream.txt"))
        yield run


@pytest.fixture(scope="class")
def stream_jobargs_run(feature_gate_class, run_container) -> Iterator[Scenario]:
    # --host is used because it is visible in the snapshot metadata of a stdin backup.
    with run_container(
        "stream-jobargs", env={**STREAM_ENV, "RESTIC_JOB_ARGS": "--host=jobargshost"}
    ) as c:
        running(c)
        run = Scenario()
        run.record("backup", lambda: c.run("/bin/backup"))
        # restic's own --host filter only returns snapshots whose hostname equals the
        # value; both directions are checked so the filter is known to discriminate.
        run.record("matching", lambda: c.restic_json("snapshots", "--host", "jobargshost") or [])
        run.record(
            "unrelated", lambda: c.restic_json("snapshots", "--host", "some-other-host") or []
        )
        yield run


@pytest.fixture(scope="class")
def stream_misconfigured_run(feature_gate_class, run_container) -> Iterator[Scenario]:
    with run_container("stream-misconfigured", env={"RESTIC_STREAM": "1"}) as c:
        running(c)
        run = Scenario()
        run.record("backup", lambda: c.exec("/bin/backup"))
        run.record("snapshots", c.snapshots)
        yield run


@pytest.fixture(scope="class")
def stream_disabled_run(feature_gate_class, run_container) -> Iterator[Scenario]:
    with run_container("stream-disabled") as c:
        running(c)
        c.write_file("/data/hello.txt", "backup me")
        run = Scenario()
        run.record("backup", lambda: c.run("/bin/backup"))
        run.record("snapshots", c.snapshots)
        yield run


@pytest.mark.feature("stream")
class TestStream:
    def test_stream_backup_succeeds(self, stream_run):
        """stream backup succeeds"""
        assert "Backup Successful" in stream_run.backup.out

    def test_stream_backup_creates_one_snapshot(self, stream_run):
        """stream backup creates exactly one snapshot"""
        assert len(stream_run.snapshots) == 1

    def test_stream_content_stored_under_filename(self, stream_run):
        """stream content is stored under RESTIC_STREAM_FILENAME"""
        assert stream_run.restore.rc == 0, stream_run.restore.out
        assert stream_run.restored.rc == 0, stream_run.restored.out
        assert stream_run.restored.out == "hello-stream\n"

    def test_job_args_set_host_of_stream_snapshot(self, stream_jobargs_run):
        """RESTIC_JOB_ARGS sets the host of the stream backup snapshot"""
        assert len(stream_jobargs_run.matching) == 1

    def test_host_filter_does_not_match_unrelated_host(self, stream_jobargs_run):
        """the host filter does not match an unrelated hostname"""
        assert stream_jobargs_run.unrelated == []

    def test_incomplete_stream_config_does_not_report_success(self, stream_misconfigured_run):
        """incomplete stream configuration does not report success"""
        assert "Backup Successful" not in stream_misconfigured_run.backup.out

    def test_incomplete_stream_config_creates_no_snapshot(self, stream_misconfigured_run):
        """incomplete stream configuration creates no snapshot"""
        assert stream_misconfigured_run.snapshots == []

    def test_file_backup_still_works_with_stream_feature(self, stream_disabled_run):
        """file backup still works while the stream feature is present"""
        assert len(stream_disabled_run.snapshots) == 1


# ---------------------------------------------------------------------------
# docker cli (PR #99)
# ---------------------------------------------------------------------------


@pytest.mark.feature("docker-cli")
def test_docker_cli_usable(docker_client, image):
    """docker cli is usable inside the image"""
    out = run_once(
        docker_client, image, entrypoint=["docker"], command=["--version"], labels=LABELS
    )
    assert "Docker version" in out


# ---------------------------------------------------------------------------
# bash in hooks (PR #99)
# ---------------------------------------------------------------------------


@pytest.fixture(scope="class")
def bash_hook_run(feature_gate_class, run_container) -> Iterator[Scenario]:
    with run_container("bash-hooks") as c:
        running(c)
        c.write_file("/hooks/pre-backup.sh", BASH_HOOK, executable=True)
        c.write_file("/data/hello.txt", "backup me")
        run = Scenario()
        run.record("backup", lambda: c.exec("/bin/backup"))
        yield run


@pytest.mark.feature("bash")
class TestBashHooks:
    def test_bash_hook_runs(self, bash_hook_run):
        """a hook with a bash shebang and bash syntax runs"""
        assert "BASH-HOOK-RAN" in bash_hook_run.backup.out


# ---------------------------------------------------------------------------
# msmtp mail relay (PR #84)
# ---------------------------------------------------------------------------


@pytest.fixture(scope="class")
def msmtp_config_run(feature_gate_class, run_container) -> Iterator[Container]:
    # entry.sh copies the config at startup, so it must exist before the container starts.
    with run_container("msmtp-config", files={"/config/msmtprc": MSMTP_CONFIG}) as c:
        yield running(c)


@pytest.fixture(scope="class")
def msmtp_missing_config_run(feature_gate_class, run_container) -> Iterator[Scenario]:
    # Users who configured mail before msmtp existed must not lose their backups.
    env = {"MAILX_ARGS": "-r from@example.com -s subject to@example.com"}
    with run_container("msmtp-missing-config", env=env) as c:
        running(c)
        c.write_file("/data/hello.txt", "backup me")
        run = Scenario()
        run.record("backup", lambda: c.exec("/bin/backup"))
        run.record("snapshots", c.snapshots)
        yield run


@pytest.mark.feature("msmtp")
class TestMsmtp:
    def test_sendmail_is_msmtp(self, docker_client, image):
        """sendmail is provided by msmtp"""
        target = run_once(
            docker_client,
            image,
            entrypoint=["sh", "-c", "readlink -f /usr/sbin/sendmail"],
            labels=LABELS,
        )
        assert "msmtp" in target

    def test_config_copied_to_etc(self, msmtp_config_run):
        """msmtp config from /config is copied to /etc/msmtprc"""
        assert msmtp_config_run.file_exists("/etc/msmtprc")

    def test_copied_config_not_world_readable(self, msmtp_config_run):
        """copied msmtp config is not world readable"""
        assert msmtp_config_run.run("stat", "-c", "%a", "/etc/msmtprc").strip() == "600"

    def test_backup_succeeds_without_msmtp_config(self, msmtp_missing_config_run):
        """backup succeeds with MAILX_ARGS set but no msmtp config"""
        assert "Backup Successful" in msmtp_missing_config_run.backup.out

    def test_snapshot_created_without_msmtp_config(self, msmtp_missing_config_run):
        """snapshot is created even when the mail relay is unconfigured"""
        assert len(msmtp_missing_config_run.snapshots) == 1