From 7875361dac198fc2680178a946c9807c9ed9c4a8 Mon Sep 17 00:00:00 2001 From: Tobias Kaupat Date: Mon, 19 Oct 2026 15:31:03 +0200 Subject: [PATCH] test: port compat tests to pytest Co-Authored-By: Claude Sonnet 5.5 --- pyproject.toml | 1 + tests/conftest.py | 25 ++- tests/helpers.py | 137 +++++++++++++ tests/test_compat.py | 470 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 632 insertions(+), 1 deletion(-) create mode 100644 tests/helpers.py create mode 100644 tests/test_compat.py diff --git a/pyproject.toml b/pyproject.toml index 5cfabe2..3b150a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ default-groups = ["dev"] testpaths = ["tests"] addopts = "-ra --strict-markers" markers = [ + "container(suffix, **env): container a test class runs against, with extra environment", "feature(name): test needs an optional image feature; skipped when absent, failed when listed in tests/required-features.txt", ] diff --git a/tests/conftest.py b/tests/conftest.py index a9e744c..8fa7caa 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,11 +7,12 @@ import os import signal import subprocess import uuid -from collections.abc import Iterator +from collections.abc import Callable, Iterator from pathlib import Path import docker import pytest +from helpers import Container, remove_quietly, start_container ROOT = Path(__file__).resolve().parent.parent IMAGE = os.environ.get("IMAGE", "restic-backup-docker:test") @@ -135,3 +136,25 @@ def _feature_gate(request: pytest.FixtureRequest, detected_features: set[str]) - pytrace=False, ) pytest.skip(f"feature '{name}' not in image yet") + + +LABELS = {"rbd-test-run": RUN_ID} + + +@pytest.fixture(scope="session") +def run_container( + docker_client: docker.DockerClient, image: str +) -> Callable[..., contextlib.AbstractContextManager[Container]]: + """Factory: `with run_container("name", env={...}) as c:` starts, waits, always removes.""" + + @contextlib.contextmanager + def _run(suffix: str, env: dict[str, str] | None = None) -> Iterator[Container]: + container = start_container( + docker_client, image, f"rbd-test-{suffix}-{RUN_ID}", LABELS, env + ) + try: + yield container + finally: + remove_quietly(container) + + return _run diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..1832d1a --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,137 @@ +"""Helpers around a running test container. + +Rule for everything in here: a failing command is never turned into an empty or +zero result. Helpers that return parsed data raise on a non-zero exit, so a broken +restic call fails the test instead of looking like "no snapshots". +""" + +from __future__ import annotations + +import contextlib +import io +import json +import tarfile +import time +from dataclasses import dataclass +from pathlib import PurePosixPath +from typing import Any + +import docker +from docker.models.containers import Container as DockerContainer + +START_TIMEOUT = 60 +START_MARKER = "Container started." + + +@dataclass(frozen=True) +class ExecResult: + rc: int + out: str # stdout and stderr, interleaved like `2>&1` + + +class CommandError(AssertionError): + pass + + +class Container: + """A started image container with exec/file helpers.""" + + def __init__(self, raw: DockerContainer, started: bool, start_log: str) -> None: + self.raw = raw + self.started = started + self.start_log = start_log + + def exec(self, *cmd: str, env: dict[str, str] | None = None) -> ExecResult: + """Run a command; the exit code is returned, not checked.""" + code, out = self.raw.exec_run(list(cmd), environment=env) + return ExecResult(code, out.decode(errors="replace").replace("\r", "")) + + def run(self, *cmd: str, env: dict[str, str] | None = None) -> str: + """Run a command that must succeed; return its output.""" + res = self.exec(*cmd, env=env) + if res.rc != 0: + raise CommandError(f"{' '.join(cmd)} exited {res.rc}:\n{res.out}") + return res.out + + def write_file(self, path: str, content: str, *, executable: bool = False) -> None: + """Create a file (and its directory) without a bind mount.""" + self.run("mkdir", "-p", str(PurePosixPath(path).parent)) + data = content.encode() + info = tarfile.TarInfo(PurePosixPath(path).name) + info.size = len(data) + info.mode = 0o755 if executable else 0o644 + buf = io.BytesIO() + with tarfile.open(fileobj=buf, mode="w") as tar: + tar.addfile(info, io.BytesIO(data)) + buf.seek(0) + if not self.raw.put_archive(str(PurePosixPath(path).parent), buf): + raise CommandError(f"could not write {path}") + + def read_file(self, path: str) -> str: + return self.run("cat", path) + + def file_exists(self, path: str) -> bool: + res = self.exec("test", "-f", path) + if res.rc not in (0, 1): + raise CommandError(f"test -f {path} exited {res.rc}:\n{res.out}") + return res.rc == 0 + + def logs(self) -> str: + return self.raw.logs().decode(errors="replace").replace("\r", "") + + # restic ------------------------------------------------------------- + + def restic_json(self, *args: str) -> Any: + """Run restic with --json and parse stdout; any failure raises.""" + code, (out, err) = self.raw.exec_run(["restic", *args, "--json"], demux=True) + if code != 0: + raise CommandError(f"restic {' '.join(args)} exited {code}:\n{(err or b'').decode()}") + return json.loads((out or b"").decode()) + + def snapshots(self) -> list[dict[str, Any]]: + return self.restic_json("snapshots") or [] + + def backed_up_paths(self, snapshot: str = "latest") -> set[str]: + """Paths contained in a snapshot (`restic ls --json` emits one JSON object per line).""" + code, (out, err) = self.raw.exec_run(["restic", "ls", snapshot, "--json"], demux=True) + if code != 0: + raise CommandError(f"restic ls {snapshot} exited {code}:\n{(err or b'').decode()}") + lines = (out or b"").decode().splitlines() + return {n["path"] for n in map(json.loads, lines) if n.get("struct_type") == "node"} + + def restore_latest(self, target: str) -> ExecResult: + return self.exec("restic", "restore", "latest", "--target", target) + + +def start_container( + client: docker.DockerClient, + image: str, + name: str, + labels: dict[str, str], + env: dict[str, str] | None = None, +) -> Container: + """Start the image with a cron far in the future, so only explicit backups run.""" + environment = { + "RESTIC_PASSWORD": "testpassword", + "RESTIC_TAG": "testtag", + "BACKUP_CRON": "0 0 31 2 *", + **(env or {}), + } + raw = client.containers.run( + image, detach=True, name=name, environment=environment, labels=labels + ) + deadline = time.monotonic() + START_TIMEOUT + while time.monotonic() < deadline: + log = raw.logs().decode(errors="replace") + if START_MARKER in log: + return Container(raw, True, log) + raw.reload() + if raw.status != "running": + return Container(raw, False, f"container exited early:\n{log}") + time.sleep(0.5) + return Container(raw, False, f"timeout waiting for start:\n{raw.logs().decode()}") + + +def remove_quietly(container: Container) -> None: + with contextlib.suppress(docker.errors.NotFound): + container.raw.remove(force=True, v=True) diff --git a/tests/test_compat.py b/tests/test_compat.py new file mode 100644 index 0000000..5b7046a --- /dev/null +++ b/tests/test_compat.py @@ -0,0 +1,470 @@ +"""Backwards-compatibility suite, ported from test/compat.sh. + +Everything here describes behaviour the image has today and that users depend on. +A change that turns one of these red is a breaking change, whether or not intended. + +Layout: every scenario fixture starts one throwaway container, performs the actions +once and records the results; the tests only assert on what was recorded, so they do +not depend on each other. Each docstring is the name of the matching Bash test. +Tests run in a container with the image defaults: repository /mnt/restic, tag +`testtag`, cron far in the future. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from dataclasses import dataclass +from types import SimpleNamespace + +import docker +import pytest +from conftest import LABELS +from helpers import Container + +HELLO = "backup me" +PRE_BACKUP_HOOK = '#!/bin/sh\necho "PRE-HOOK-RAN"\n' +POST_BACKUP_HOOK = '#!/bin/sh\necho "POST-HOOK-RAN with rc=$1"\n' +PRE_CHECK_HOOK = '#!/bin/sh\necho "PRE-CHECK-HOOK-RAN"\n' +POST_CHECK_HOOK = '#!/bin/sh\necho "POST-CHECK-HOOK-RAN with rc=$1"\n' + + +@pytest.fixture(scope="class") +def started(request: pytest.FixtureRequest, run_container) -> Iterator[Container]: + """Container for the class, configured by `@pytest.mark.container(suffix, **env)`.""" + marker = request.node.get_closest_marker("container") + with run_container(marker.args[0], env=marker.kwargs) as container: + yield container + + +def _running(container: Container) -> Container: + """Scenarios that act on the container need it started; the start tests do not.""" + if not container.started: + pytest.fail(f"container did not start:\n{container.start_log}", pytrace=False) + return container + + +def _crontab_lines(container: Container) -> list[str]: + return container.read_file("/var/spool/cron/crontabs/root").splitlines() + + +# --------------------------------------------------------------------------- +# repository bootstrap and default backup +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="class") +def default_run(started: Container) -> SimpleNamespace: + c = _running(started) + c.write_file("/data/hello.txt", HELLO) + run = SimpleNamespace(container=c) + run.backup = c.exec("/bin/backup") + run.snapshots_after_first = c.snapshots() + run.restore = c.restore_latest("/tmp/restore") + run.restored = c.exec("cat", "/tmp/restore/data/hello.txt") + run.backup_last_log = c.read_file("/var/log/backup-last.log") + c.run("/bin/backup") + run.snapshots_after_second = c.snapshots() + run.check = c.exec("/bin/check") + c.write_file("/hooks/pre-backup.sh", PRE_BACKUP_HOOK, executable=True) + c.write_file("/hooks/post-backup.sh", POST_BACKUP_HOOK, executable=True) + run.hook_backup = c.exec("/bin/backup") + return run + + +@pytest.mark.container("default") +class TestDefaultBackup: + def test_container_starts(self, started): + """container starts with default configuration""" + assert started.started, started.start_log + + def test_repository_is_initialised(self, started): + """empty repository is initialised on first start""" + assert "Running restic init" in started.logs() + + def test_repository_init_succeeds(self, started): + """repository initialisation succeeds""" + assert "Repo init status 0" in started.logs() + + def test_backup_cron_in_crontab(self, started): + """BACKUP_CRON is written to the crontab""" + lines = _crontab_lines(started) + assert any(line.startswith("0 0 31 2 *") and "/bin/backup" in line for line in lines) + + def test_backup_cron_guarded_by_flock(self, started): + """backup cron job is guarded by flock""" + line = next(line for line in _crontab_lines(started) if "/bin/backup" in line) + assert "flock -n /var/run/backup.lock" in line + + def test_no_check_cron_by_default(self, started): + """no check cron job is installed while CHECK_CRON is unset""" + lines = _crontab_lines(started) + assert any("/bin/backup" in line for line in lines) # the crontab we looked at is real + assert not any("/bin/check" in line for line in lines) + + def test_backup_succeeds(self, default_run): + """backup of /data succeeds""" + assert default_run.backup.rc == 0 + assert "Backup Successful" in default_run.backup.out + + def test_missing_pre_backup_hook_reported(self, default_run): + """missing pre-backup hook is reported, not fatal""" + assert "Pre-backup script not found" in default_run.backup.out + assert default_run.backup.rc == 0 + + def test_missing_post_backup_hook_reported(self, default_run): + """missing post-backup hook is reported, not fatal""" + assert "Post-backup script not found" in default_run.backup.out + assert default_run.backup.rc == 0 + + def test_no_notification_without_config(self, default_run): + """no mail or Teams notification is attempted without configuration""" + assert "Backup Successful" in default_run.backup.out + assert "notification" not in default_run.backup.out + + def test_one_snapshot_after_one_backup(self, default_run): + """exactly one snapshot exists after one backup""" + assert len(default_run.snapshots_after_first) == 1 + + def test_snapshot_carries_restic_tag(self, default_run): + """snapshot carries the configured RESTIC_TAG""" + assert [s["tags"] for s in default_run.snapshots_after_first] == [["testtag"]] + + def test_restore_exits_zero(self, default_run): + """restic restore of the latest snapshot exits 0""" + assert default_run.restore.rc == 0, default_run.restore.out + + def test_restored_content_matches(self, default_run): + """restored file content matches the original""" + assert default_run.restored.rc == 0, default_run.restored.out + assert default_run.restored.out == HELLO + + def test_backup_last_log_records_parameters(self, default_run): + """backup-last.log records the run parameters""" + assert "RESTIC_REPOSITORY: /mnt/restic" in default_run.backup_last_log + + def test_second_backup_adds_snapshot(self, default_run): + """a second backup adds a second snapshot""" + ids = {s["id"] for s in default_run.snapshots_after_second} + assert len(default_run.snapshots_after_second) == 2 + assert {s["id"] for s in default_run.snapshots_after_first} < ids + + def test_check_passes_on_healthy_repository(self, default_run): + """restic check passes on a healthy repository""" + assert default_run.check.rc == 0 + assert "Check Successful" in default_run.check.out + + def test_missing_pre_check_hook_reported(self, default_run): + """missing pre-check hook is reported, not fatal""" + assert "Pre-check script not found" in default_run.check.out + assert "Check Successful" in default_run.check.out + + def test_pre_backup_hook_executed(self, default_run): + """pre-backup hook is executed""" + assert "PRE-HOOK-RAN" in default_run.hook_backup.out + + def test_post_backup_hook_gets_return_code(self, default_run): + """post-backup hook receives the backup return code""" + assert "POST-HOOK-RAN with rc=0" in default_run.hook_backup.out + + +# --------------------------------------------------------------------------- +# CHECK_CRON and check hooks +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="class") +def checkcron_run(started: Container) -> SimpleNamespace: + c = _running(started) + run = SimpleNamespace(container=c, crontab=_crontab_lines(c)) + c.write_file("/data/hello.txt", HELLO) + c.run("/bin/backup") + c.write_file("/hooks/pre-check.sh", PRE_CHECK_HOOK, executable=True) + c.write_file("/hooks/post-check.sh", POST_CHECK_HOOK, executable=True) + run.check = c.exec("/bin/check") + return run + + +@pytest.mark.container("checkcron", CHECK_CRON="0 0 31 2 *") +class TestCheckCron: + def test_check_cron_guarded_by_shared_lock(self, checkcron_run): + """check cron job is guarded by the shared backup lock""" + check_lines = [line for line in checkcron_run.crontab if "/bin/check" in line] + assert len(check_lines) == 1 + assert "flock -n /var/run/backup.lock" in check_lines[0] + + def test_check_cron_adds_cron_job(self, checkcron_run): + """CHECK_CRON adds a check cron job to the crontab""" + assert any( + line.startswith("0 0 31 2 *") and "/bin/check" in line for line in checkcron_run.crontab + ) + + def test_crontab_has_backup_and_check_line(self, checkcron_run): + """crontab has a backup line and a check line""" + assert len([line for line in checkcron_run.crontab if line.strip()]) == 2 + + def test_check_cron_keeps_backup_cron(self, checkcron_run): + """CHECK_CRON does not replace the backup cron job""" + assert "/bin/backup" in checkcron_run.crontab[0] + + def test_pre_check_hook_executed(self, checkcron_run): + """pre-check hook is executed""" + assert "PRE-CHECK-HOOK-RAN" in checkcron_run.check.out + + def test_post_check_hook_gets_return_code(self, checkcron_run): + """post-check hook receives the check return code""" + assert "POST-CHECK-HOOK-RAN with rc=0" in checkcron_run.check.out + + +# --------------------------------------------------------------------------- +# a corrupted repository is reported as failed +# +# check.sh prints "Check Successful" whenever restic exits 0, so a `restic check` +# that does nothing would keep the success path green. Damage the repository and +# require the failure path. The container is throwaway; the repo stays broken. +# --------------------------------------------------------------------------- + +CORRUPT_CMD = 'f=$(find /mnt/restic/data -type f | head -n1); chmod u+w "$f"; printf garbage > "$f"' + + +@pytest.fixture(scope="class") +def corrupt_run(started: Container) -> SimpleNamespace: + c = _running(started) + c.write_file("/data/hello.txt", HELLO) + c.run("/bin/backup") + run = SimpleNamespace(container=c, snapshots_before=c.snapshots()) + c.run("sh", "-c", CORRUPT_CMD) + run.check = c.exec("/bin/check") + run.error_log_exists = c.file_exists("/var/log/check-error-last.log") + return run + + +@pytest.mark.container("corrupt") +class TestCorruptedRepository: + def test_repository_to_corrupt_has_snapshot(self, corrupt_run): + """the repository to be corrupted has a snapshot""" + assert len(corrupt_run.snapshots_before) == 1 + + def test_check_detects_corruption(self, corrupt_run): + """restic check detects a corrupted repository""" + assert "Check Failed with Status" in corrupt_run.check.out + + def test_corruption_not_reported_successful(self, corrupt_run): + """a corrupted repository is not reported as successful""" + assert "Check Failed with Status" in corrupt_run.check.out # it did run and report + assert "Check Successful" not in corrupt_run.check.out + + def test_failing_check_writes_error_log(self, corrupt_run): + """failing check writes check-error-last.log""" + assert corrupt_run.error_log_exists + + +# --------------------------------------------------------------------------- +# RESTIC_FORGET_ARGS +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="class") +def forget_run(started: Container) -> SimpleNamespace: + c = _running(started) + c.write_file("/data/hello.txt", HELLO) + c.run("/bin/backup") + first = c.snapshots() + run = SimpleNamespace(container=c, first=first) + run.second_backup = c.exec("/bin/backup") + run.snapshots = c.snapshots() + return run + + +@pytest.mark.container("forget", RESTIC_FORGET_ARGS="--keep-last 1") +class TestForgetArgs: + def test_forget_runs_after_backup(self, forget_run): + """forget runs after a successful backup when configured""" + assert "Forget Successful" in forget_run.second_backup.out + + def test_keep_last_leaves_one_snapshot(self, forget_run): + """--keep-last 1 leaves exactly one snapshot""" + # two backups ran: one snapshot left that is not the first one proves forget pruned + assert len(forget_run.first) == 1 + assert len(forget_run.snapshots) == 1 + assert forget_run.snapshots[0]["id"] != forget_run.first[0]["id"] + + +# --------------------------------------------------------------------------- +# RESTIC_JOB_ARGS +# +# backup-last.log only records the variable, it does not prove restic ever saw it. +# The arguments need a visible effect, otherwise a backup that silently drops +# RESTIC_JOB_ARGS still looks healthy. +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="class") +def jobargs_run(started: Container) -> SimpleNamespace: + c = _running(started) + c.write_file("/data/keep.txt", "keep") + c.write_file("/data/skipme.txt", "skip") + run = SimpleNamespace(container=c, backup=c.exec("/bin/backup")) + run.restore = c.restore_latest("/tmp/restore") + run.restored_keep = c.exec("cat", "/tmp/restore/data/keep.txt") + run.paths = c.backed_up_paths() + return run + + +@pytest.mark.container("jobargs", RESTIC_JOB_ARGS="--exclude=/data/skipme.txt") +class TestJobArgs: + def test_backup_with_job_args_succeeds(self, jobargs_run): + """backup with RESTIC_JOB_ARGS succeeds""" + assert "Backup Successful" in jobargs_run.backup.out + + def test_job_args_do_not_break_backup(self, jobargs_run): + """RESTIC_JOB_ARGS does not break a normal backup""" + assert jobargs_run.restore.rc == 0, jobargs_run.restore.out + assert jobargs_run.restored_keep.out == "keep" + + def test_job_args_passed_to_restic(self, jobargs_run): + """RESTIC_JOB_ARGS is passed to restic backup""" + # keep.txt shows the snapshot is the one of this scenario, skipme.txt must be missing + assert "/data/keep.txt" in jobargs_run.paths + assert "/data/skipme.txt" not in jobargs_run.paths + + +# --------------------------------------------------------------------------- +# BACKUP_SOURCES +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="class") +def sources_run(started: Container) -> SimpleNamespace: + c = _running(started) + c.write_file("/opt/custom/custom.txt", "custom source") + c.write_file("/data/ignored.txt", "should not be backed up") + c.run("/bin/backup") + run = SimpleNamespace(container=c, restore=c.restore_latest("/tmp/restore")) + run.restored = c.exec("cat", "/tmp/restore/opt/custom/custom.txt") + run.paths = c.backed_up_paths() + return run + + +@pytest.mark.container("sources", BACKUP_SOURCES="/opt/custom") +class TestBackupSources: + def test_sources_override_default(self, sources_run): + """BACKUP_SOURCES overrides the default /data source""" + assert sources_run.restore.rc == 0, sources_run.restore.out + assert sources_run.restored.out == "custom source" + + def test_sources_exclude_data(self, sources_run): + """BACKUP_SOURCES excludes /data""" + assert "/opt/custom/custom.txt" in sources_run.paths + assert not [p for p in sources_run.paths if p.startswith("/data")] + + +# --------------------------------------------------------------------------- +# failure handling +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="class") +def failure_run(started: Container) -> SimpleNamespace: + c = _running(started) + c.write_file("/data/hello.txt", HELLO) + # Back up a source that does not exist. The repository stays the initialised + # default one, so the snapshot list below is of the repo the run really used. + run = SimpleNamespace( + container=c, backup=c.exec("/bin/backup", env={"BACKUP_SOURCES": "/does-not-exist"}) + ) + run.error_log_exists = c.file_exists("/var/log/backup-error-last.log") + run.snapshots = c.snapshots() + return run + + +@pytest.mark.container("failure") +class TestFailureHandling: + def test_failing_backup_reported(self, failure_run): + """a failing backup is reported as failed""" + assert "Backup Failed with Status" in failure_run.backup.out + + def test_failing_backup_writes_error_log(self, failure_run): + """failing backup writes backup-error-last.log""" + assert failure_run.error_log_exists + + def test_failed_backup_leaves_no_snapshot(self, failure_run): + """a failed backup leaves no snapshot behind""" + # snapshots() raises on a restic error, so [] means a readable, empty repository + assert "Backup Failed with Status" in failure_run.backup.out + assert failure_run.snapshots == [] + + +# --------------------------------------------------------------------------- +# timezone +# +# The base image already ships tzdata, so TZ works today. Pinned here so that +# neither a base image change nor a slimming effort silently breaks cron times +# for users who already rely on it. +# --------------------------------------------------------------------------- + + +@dataclass +class TimezoneRunner: + client: docker.DockerClient + image: str + + def zone(self, env: dict[str, str] | None = None) -> str: + out = self.client.containers.run( + self.image, entrypoint=["date", "+%Z"], environment=env, remove=True, labels=LABELS + ) + return out.decode().strip() + + +@pytest.fixture(scope="module") +def timezone(docker_client, image) -> TimezoneRunner: + return TimezoneRunner(docker_client, image) + + +class TestTimezone: + def test_default_timezone_is_utc(self, timezone): + """the default timezone is UTC""" + assert timezone.zone() == "UTC" + + def test_tz_variable_changes_timezone(self, timezone): + """TZ=Europe/Berlin changes the container timezone""" + assert timezone.zone({"TZ": "Europe/Berlin"}) in ("CET", "CEST") + + +# --------------------------------------------------------------------------- +# RESTIC_TAG handling +# +# The Dockerfile declares RESTIC_TAG="", so in a normal container the variable is +# always set and an empty value means "no tag". The `${RESTIC_TAG?...}` guard in +# backup.sh only triggers when the variable is truly unset. +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="class") +def tag_run(started: Container) -> SimpleNamespace: + c = _running(started) + c.write_file("/data/hello.txt", HELLO) + run = SimpleNamespace(container=c, empty=c.exec("/bin/backup", env={"RESTIC_TAG": ""})) + run.snapshots_after_empty = c.snapshots() + run.unset = c.exec("env", "-u", "RESTIC_TAG", "/bin/backup") + run.snapshots_after_unset = c.snapshots() + return run + + +@pytest.mark.container("tag") +class TestResticTag: + def test_empty_tag_creates_untagged_backup(self, tag_run): + """an empty RESTIC_TAG creates an untagged backup""" + assert "Backup Successful" in tag_run.empty.out + + def test_empty_tag_still_produces_snapshot(self, tag_run): + """an empty RESTIC_TAG still produces a snapshot""" + assert len(tag_run.snapshots_after_empty) == 1 + + def test_unset_tag_refuses_backup(self, tag_run): + """backup refuses to run when RESTIC_TAG is unset entirely""" + assert "Missing environment variable RESTIC_TAG" in tag_run.unset.out + + def test_unset_tag_creates_no_snapshot(self, tag_run): + """no extra snapshot is created when RESTIC_TAG is unset""" + assert len(tag_run.snapshots_after_empty) == 1 # there was something to not add to + assert len(tag_run.snapshots_after_unset) == 1