Release version 3.6.1

The --empty pre-clean is a catalog-wide sweep: it drops every non-template database and every non-pg_ role of the instance. On a dedicated instance that is exactly right, because the dump recreates all of it. On a shared one it destroys databases the dump does not carry, with nothing to restore them from - and no test ever executed that sweep, because the e2e dropped the cluster by hand first and left the pre-clean with zero rows to generate.

Scoping the sweep to the dump's own inventory looks like the fix and is worse. A surviving database that owns or merely grants to one of the dump's roles pins that role in pg_shdepend; DROP OWNED BY only reaches the control database the pre-clean is connected to, so DROP ROLE fails - after phase 1 has already dropped the dump's databases. ON_ERROR_STOP aborts, the replay never starts, and the instance is left half emptied.

So the instance is checked instead. --empty now refuses when the instance holds a database the dump does not carry, names it, and touches nothing. The sweep stays as it was, safe behind that refusal. Reading the dump's inventory needs a real identifier parser: a quoted name may hold spaces, and psql options precede the target of a \\connect line.

The e2e no longer drops the cluster itself, so --empty has to do it and the replay has to put it back; a second pass then adds a foreign database and requires the refusal to leave both it and the restored data alone.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -1,5 +1,64 @@
 # Changelog
 
+## [3.6.1] - 2026-08-17
+
+- Restore: *--empty* on a cluster dump is a catalog-wide sweep — it drops every
+  non-template database and every non-pg_ role of the instance. On a dedicated
+  instance that is right, because the dump recreates all of it; on a shared one
+  it destroys databases the dump does not carry, with nothing to restore them
+  from. No test had ever executed that sweep: the e2e dropped the cluster by
+  hand first and left the pre-clean with zero rows to generate.
+- Restore: the instance is checked instead of the sweep being narrowed.
+  *--empty* refuses when the instance holds a database the dump does not carry,
+  names it, and touches nothing. Narrowing the sweep is the obvious fix and is
+  worse — a surviving database that owns or merely grants to one of the dump's
+  roles pins it in *pg_shdepend*, *DROP OWNED BY* reaches only the control
+  database the pre-clean is connected to, so *DROP ROLE* fails after the dump's
+  own databases are already gone and the replay never starts.
+- Restore: the dump's inventory is read with a real identifier parser. A quoted
+  name may hold spaces, and psql options precede the target of a *\connect*
+  line, so a character class that stops at whitespace read *"odd name"* as
+  *odd* and *-reuse-previous=on* as a database.
+- Tests: the cluster e2e no longer empties the instance itself, so *--empty*
+  has to do it and the replay has to put it back. A second pass adds a foreign
+  database and requires the refusal to leave both it and the restored data
+  untouched.
+
+## [3.6.0] - 2026-08-17
+
+- Restore: *--empty* drops the schema in one session and replays in the next,
+  with no rollback across the two, so a dump the engine could not parse left an
+  emptied database behind. The dump's header is now checked against the running
+  engine before anything is dropped, and a newer dump is refused.
+  Forward across a major version stays allowed; *--no-version-check* is the way out.
+- Restore: a volume with driver options — NFS, a bind device, tmpfs — keeps the
+  usual *_data* path, but docker mounts its real storage over it only while a
+  container holds it. Restoring meanwhile landed under the mount, stayed hidden
+  there, and rsync reported success. That volume is now refused until something
+  mounts it.
+- Backup: the same volume sits in a snapshot as an empty directory, so it was
+  copied empty and the generation stamped complete. Capture is decided per volume
+  now — an uncaptured one is copied live, the rest keep their snapshot. A single
+  NFS volume no longer costs the whole host its consistent backup.
+
+## [3.5.0] - 2026-08-17
+
+- Restore: a *database = '*'* row makes the backup write
+  *<instance>.cluster.backup.sql* via *pg_dumpall*, and nothing could read it
+  back — the CLI knew *files*, *postgres* and *mariadb*, so that dump was
+  stored and unrestorable. *baudolo-restore cluster* replays it against the
+  control database, deliberately without *--single-transaction* because
+  CREATE DATABASE is forbidden inside a transaction block, and filters out the
+  CREATE ROLE of the connecting role, which the pre-clean cannot drop while it
+  holds the session. *--empty* drops the cluster's databases first, then
+  releases what its roles still own, then the roles themselves.
+
+- Lint: ruff was never wired into the repository — no target, no CI step, no
+  pin — and reported 45 findings across sources and tests. *make ruff* and
+  *make lint* now run it over every file, *make test* gates on a clean run as a
+  fourth parallel spur, and the linter is pinned in a *lint* extra because a
+  minor bump changes which rules fire.
+
 ## [3.4.3] - 2026-08-16
 
 - Backup: *create_version_directory* carried *exist_ok=True*, so a run starting

Makefile

@@ -1,4 +1,4 @@
-.PHONY: install build clean \
+.PHONY: install install-lint build clean lint ruff ruff-fix \
 		test test-unit test-integration test-e2e \
 		test-unit-run test-integration-run test-e2e-run
 
@@ -34,13 +34,30 @@ build:
 clean:
 	git clean -fdX .
 
-# clean + build run once and in order, then the three suites run concurrently
-# via -j3; the *-run targets carry no clean/build prereq so the sub-make cannot
-# race a second clean against build.
+# Separate from `install` so the test image does not have to carry the linter.
+install-lint:
+	@$(PY_DEFAULT) -m pip install -q -e ".[lint]"
+
+# Runs on the host, not in the image, so it also covers what the Dockerfile
+# does not copy.
+ruff: install-lint
+	@echo ">> Running ruff over the whole repository"
+	@$(PY_DEFAULT) -m ruff check .
+	@$(PY_DEFAULT) -m ruff format --check .
+
+ruff-fix: install-lint
+	@$(PY_DEFAULT) -m ruff check --fix .
+	@$(PY_DEFAULT) -m ruff format .
+
+lint: ruff
+
+# clean + build run once and in order, then lint and the three suites run
+# concurrently via -j4; the *-run targets carry no clean/build prereq so the
+# sub-make cannot race a second clean against build.
 test:
 	@$(MAKE) clean
 	@$(MAKE) build
-	@$(MAKE) -j3 test-unit-run test-integration-run test-e2e-run
+	@$(MAKE) -j4 lint test-unit-run test-integration-run test-e2e-run
 
 test-unit: clean build test-unit-run
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "backup-docker-to-local"
-version = "3.4.3"
+version = "3.6.1"
 description = "Backup Docker volumes to local with rsync and optional DB dumps."
 readme = "README.md"
 requires-python = ">=3.9"
@@ -16,6 +16,11 @@ dependencies = [
   "dirval",
 ]
 
+[project.optional-dependencies]
+# Pinned: a ruff minor bump changes which rules fire, and `make test` gates on
+# a clean run, so an unpinned lint would fail the suite on an unrelated day.
+lint = ["ruff==0.16.1"]
+
 [project.scripts]
 baudolo = "baudolo.backup.__main__:main"
 baudolo-restore = "baudolo.restore.__main__:main"

src/baudolo/backup/__main__.py

@@ -1,9 +1,6 @@
-#!/usr/bin/env python3
-
 from __future__ import annotations
 
 from .app import main
 
-
 if __name__ == "__main__":
     raise SystemExit(main())

src/baudolo/backup/app.py

@@ -22,25 +22,21 @@ from .layout import (
     stamp_directory,
 )
 from .policy import requires_stop, volume_is_fully_ignored
-from .snapshot import volume_snapshot
-from .volume import backup_volume, get_storage_path
+from .snapshot import snapshot_source, volume_snapshot
+from .volume import backup_volume, inspect_backing
 
 
 def main() -> int:
     args = parse_args()
 
     machine_id = get_machine_id()
-    backup_time = datetime.now().strftime("%Y%m%d%H%M%S")
+    # Local wall clock on purpose: generations sort by this name, and UTC would
+    # order new ones before the existing ones wherever the offset is positive.
+    backup_time = datetime.now().strftime("%Y%m%d%H%M%S")  # noqa: DTZ005
 
     versions_dir = os.path.join(args.backups_dir, machine_id, args.repo_name)
     version_dir = create_version_directory(versions_dir, backup_time)
 
-    # IMPORTANT:
-    # - keep_default_na=False prevents empty fields from turning into NaN
-    # - dtype=str keeps all columns stable for comparisons/validation
-    #
-    # Robust behavior:
-    # - if the file is missing or empty, we continue without DB dumps.
     databases_df = load_databases_df(args.databases_csv)
 
     print("💾 Start volume backups...", flush=True)
@@ -80,36 +76,44 @@ def main() -> int:
                 database_containers=args.database_containers,
             )
 
-            if args.dump_only_sql:
-                if found_db:
-                    if not dumped_any:
-                        print(
-                            f"WARNING: dump-only-sql requested but no DB dump was produced for DB volume '{volume_name}'. "
-                            "Falling back to file backup.",
-                            flush=True,
-                        )
-                    else:
-                        continue
+            if args.dump_only_sql and found_db:
+                if not dumped_any:
+                    print(
+                        f"WARNING: dump-only-sql requested but no DB dump was produced for DB volume '{volume_name}'. "
+                        "Falling back to file backup.",
+                        flush=True,
+                    )
+                else:
+                    continue
 
-            live_source = get_storage_path(volume_name)
+            backing = inspect_backing(volume_name)
+            live_source = backing.source
 
-            def copy(*, authoritative: bool, source: str = live_source) -> None:
+            def copy(
+                *,
+                authoritative: bool,
+                source: str = live_source,
+                volume: str = volume_name,
+                target: str = vol_dir,
+            ) -> None:
                 backup_volume(
                     versions_dir,
-                    volume_name,
-                    vol_dir,
+                    volume,
+                    target,
                     authoritative=authoritative,
                     source=source,
                 )
 
             if resolve_source is not None:
-                snapshot_source = resolve_source(live_source)
-                if os.path.isdir(snapshot_source):
-                    copy(authoritative=True, source=snapshot_source)
+                source, reason = snapshot_source(
+                    resolve_source, backing, args.snapshot_subject
+                )
+                if source is not None:
+                    copy(authoritative=True, source=source)
                 else:
                     print(
                         f"WARNING: volume '{volume_name}' is not in the snapshot "
-                        "(created after it was taken); copying it live instead.",
+                        f"({reason}); copying it live instead.",
                         flush=True,
                     )
                     copy(authoritative=False)

src/baudolo/backup/compose.py

@@ -4,10 +4,9 @@ import os
 import shutil
 import subprocess
 from pathlib import Path
-from typing import List, Optional
 
 
-def _build_compose_cmd(project_dir: str, passthrough: List[str]) -> List[str]:
+def _build_compose_cmd(project_dir: str, passthrough: list[str]) -> list[str]:
     """
     Build the compose command for this project directory.
 
@@ -30,7 +29,7 @@ def _build_compose_cmd(project_dir: str, passthrough: List[str]) -> List[str]:
     raise RuntimeError("Neither 'compose' nor 'docker' found in PATH")
 
 
-def _find_compose_file(project_dir: str) -> Optional[Path]:
+def _find_compose_file(project_dir: str) -> Path | None:
     """
     Detect a compose file in `project_dir` (case-insensitive).
 

src/baudolo/backup/db.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
 
+import logging
 import os
 import pathlib
 import re
-import logging
-from typing import Optional
 
 import pandas
 
@@ -22,7 +21,7 @@ def get_instance(container: str, database_containers: list[str]) -> str:
     return re.split(r"(_|-)(database|db|postgres)", container)[0]
 
 
-def _validate_database_value(value: Optional[str], *, instance: str) -> str:
+def _validate_database_value(value: str | None, *, instance: str) -> str:
     """
     Enforce explicit database semantics:
 
@@ -70,7 +69,7 @@ def backup_database(
     container: str,
     volume_dir: str,
     db_type: str,
-    databases_df: "pandas.DataFrame",
+    databases_df: pandas.DataFrame,
     database_containers: list[str],
 ) -> bool:
     """
@@ -97,7 +96,6 @@ def backup_database(
 
         db_value = _validate_database_value(raw_db, instance=instance_name)
 
-        # Explicit: dump ALL databases
         if db_value == "*":
             if db_type != "postgres":
                 raise ValueError(
@@ -110,7 +108,6 @@ def backup_database(
             produced = True
             continue
 
-        # Concrete database dump
         db_name = db_value
         dump_file = os.path.join(out_dir, f"{db_name}.backup.sql")
 
@@ -135,7 +132,6 @@ def backup_database(
                 _atomic_write_cmd(cmd, dump_file)
                 produced = True
             except BackupException as e:
-                # Explicit DB dump failed -> hard error
                 raise BackupException(
                     f"Postgres dump failed for instance '{instance_name}', "
                     f"database '{db_name}'. This database was explicitly configured "

src/baudolo/backup/docker.py

@@ -98,5 +98,5 @@ def docker_volume_exists(volume: str) -> bool:
             f"docker volume inspect {volume} >/dev/null 2>&1 && echo OK"
         )
         return True
-    except Exception:
+    except BackupException:
         return False

src/baudolo/backup/dumps.py

@@ -15,7 +15,7 @@ def backup_mariadb_or_postgres(
     *,
     container: str,
     volume_dir: str,
-    databases_df: "pandas.DataFrame",
+    databases_df: pandas.DataFrame,
     database_containers: list[str],
 ) -> tuple[bool, bool]:
     """
@@ -34,7 +34,7 @@ def backup_mariadb_or_postgres(
     return False, False
 
 
-def _empty_databases_df() -> "pandas.DataFrame":
+def _empty_databases_df() -> pandas.DataFrame:
     """
     Create an empty DataFrame with the expected schema for databases.csv.
 
@@ -44,7 +44,7 @@ def _empty_databases_df() -> "pandas.DataFrame":
     return pandas.DataFrame(columns=["instance", "database", "username", "password"])
 
 
-def load_databases_df(csv_path: str) -> "pandas.DataFrame":
+def load_databases_df(csv_path: str) -> pandas.DataFrame:
     """
     Load databases.csv robustly.
 
@@ -74,7 +74,7 @@ def backup_dumps_for_volume(
     *,
     containers: list[str],
     vol_dir: str,
-    databases_df: "pandas.DataFrame",
+    databases_df: pandas.DataFrame,
     database_containers: list[str],
 ) -> tuple[bool, bool]:
     """

src/baudolo/backup/snapshot.py

@@ -9,6 +9,13 @@ built for. That also removes the reason to stop containers at all.
 The snapshot kind is stated by the caller rather than probed, because falling
 back to a live copy when a probe is inconclusive would hand out backups that
 look consistent and are not.
+
+Which volumes a snapshot of the subject contains is a different question, and
+it is decided per volume: a volume with a backing store of its own appears
+inside the snapshot as an existing empty directory, so copying from there
+succeeds and stores nothing. Such a volume is copied live instead - correct
+data without the point in time - while every other volume of the same run
+keeps its snapshot.
 """
 
 from __future__ import annotations
@@ -18,6 +25,7 @@ from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 
 from .shell import BackupException, execute_shell_command
+from .volume import Backing
 
 KINDS = ("btrfs", "zfs")
 
@@ -62,6 +70,60 @@ def _zfs(subject: str, name: str, run: Callable[[str], list[str]]) -> tuple[str,
 _CREATE = {"btrfs": _btrfs, "zfs": _zfs}
 
 
+def unsnapshotted(backing: Backing, subject: str) -> str | None:
+    """Return why a snapshot of ``subject`` does not hold this volume's data.
+
+    Docker mounts a volume's own backing store lazily and unmounts it when the
+    last consumer stops, so the declaration is what gets checked: it is true at
+    every moment, where the mount table is only true while a container happens
+    to hold the volume.
+
+    Args:
+        backing: the volume as the daemon describes it.
+        subject: the snapshot subject, e.g. ``/var/lib/docker``.
+
+    Returns:
+        The reason, or None when the snapshot holds the volume.
+    """
+    if backing.driver != "local":
+        return f"it uses the {backing.driver} driver"
+    if backing.options:
+        return f"it declares its own backing store {backing.options}"
+    if not backing.mountpoint:
+        return "it reports no mountpoint"
+    real = os.path.realpath(backing.mountpoint)
+    if os.path.ismount(real):
+        return f"its mountpoint {backing.mountpoint} sits on its own mount"
+    try:
+        crosses = os.stat(real).st_dev != os.stat(os.path.realpath(subject)).st_dev
+    except OSError as error:
+        return f"its mountpoint {backing.mountpoint} could not be read: {error}"
+    if crosses:
+        return f"its mountpoint {backing.mountpoint} crosses a filesystem boundary"
+    return None
+
+
+def snapshot_source(
+    resolve: Callable[[str], str], backing: Backing, subject: str
+) -> tuple[str | None, str]:
+    """Resolve where to read a volume from, and why if not from the snapshot.
+
+    Returns:
+        ``(path, "")`` to copy from the snapshot, or ``(None, reason)`` to copy
+        it live.
+    """
+    reason = unsnapshotted(backing, subject)
+    if reason:
+        return None, reason
+    try:
+        source = resolve(backing.source)
+    except SnapshotError as error:
+        return None, str(error)
+    if not os.path.isdir(source):
+        return None, "it was created after the snapshot was taken"
+    return source, ""
+
+
 @contextmanager
 def volume_snapshot(
     kind: str,

src/baudolo/backup/volume.py

@@ -1,16 +1,43 @@
 from __future__ import annotations
 
+import json
 import os
 import pathlib
+from dataclasses import dataclass, field
 
 from .shell import BackupException, execute_shell_command
 
 
-def get_storage_path(volume_name: str) -> str:
-    path = execute_shell_command(
-        f"docker volume inspect --format '{{{{ .Mountpoint }}}}' {volume_name}"
+@dataclass(frozen=True)
+class Backing:
+    """Where a docker volume actually keeps its data.
+
+    Args:
+        mountpoint: the path the daemon reports.
+        driver: the volume driver, ``local`` for the built-in one.
+        options: the driver options; a non-empty map means the mountpoint is a
+            mount target rather than the storage itself.
+    """
+
+    mountpoint: str
+    driver: str = "local"
+    options: dict = field(default_factory=dict)
+
+    @property
+    def source(self) -> str:
+        return f"{self.mountpoint}/"
+
+
+def inspect_backing(volume_name: str) -> Backing:
+    reported = execute_shell_command(
+        f"docker volume inspect --format '{{{{json .}}}}' {volume_name}"
     )[0]
-    return f"{path}/"
+    data = json.loads(reported)
+    return Backing(
+        data.get("Mountpoint") or "",
+        data.get("Driver") or "",
+        data.get("Options") or {},
+    )
 
 
 def get_last_backup_dir(

src/baudolo/restore/__main__.py

@@ -3,10 +3,11 @@ from __future__ import annotations
 import argparse
 import sys
 
-from .paths import BackupPaths
-from .files import restore_volume_files
-from .db.postgres import restore_postgres_sql
+from .db.cluster import restore_cluster_sql
 from .db.mariadb import restore_mariadb_sql
+from .db.postgres import restore_postgres_sql
+from .files import restore_volume_files
+from .paths import BackupPaths
 
 
 def _add_common_backup_args(p: argparse.ArgumentParser) -> None:
@@ -26,6 +27,20 @@ def _add_common_backup_args(p: argparse.ArgumentParser) -> None:
     )
 
 
+def _add_common_engine_args(p: argparse.ArgumentParser) -> None:
+    p.add_argument("--container", required=True)
+    p.add_argument("--db-password", required=True)
+    p.add_argument("--empty", action="store_true")
+    p.add_argument(
+        "--no-version-check",
+        action="store_true",
+        help=(
+            "Replay even if the dump comes from a newer engine than the target. "
+            "With --empty this can leave an emptied database behind."
+        ),
+    )
+
+
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(
         prog="baudolo-restore",
@@ -33,9 +48,6 @@ def main(argv: list[str] | None = None) -> int:
     )
     sub = parser.add_subparsers(dest="cmd", required=True)
 
-    # ------------------------------------------------------------------
-    # files
-    # ------------------------------------------------------------------
     p_files = sub.add_parser("files", help="Restore files into a docker volume")
     _add_common_backup_args(p_files)
     p_files.add_argument(
@@ -48,36 +60,40 @@ def main(argv: list[str] | None = None) -> int:
         ),
     )
 
-    # ------------------------------------------------------------------
-    # postgres
-    # ------------------------------------------------------------------
     p_pg = sub.add_parser("postgres", help="Restore a single PostgreSQL database dump")
     _add_common_backup_args(p_pg)
-    p_pg.add_argument("--container", required=True)
+    _add_common_engine_args(p_pg)
     p_pg.add_argument("--db-name", required=True)
     p_pg.add_argument("--db-user", default=None, help="Defaults to db-name if omitted")
-    p_pg.add_argument("--db-password", required=True)
-    p_pg.add_argument("--empty", action="store_true")
 
-    # ------------------------------------------------------------------
-    # mariadb
-    # ------------------------------------------------------------------
+    p_cluster = sub.add_parser(
+        "cluster", help="Restore a full PostgreSQL cluster dump (pg_dumpall)"
+    )
+    _add_common_backup_args(p_cluster)
+    _add_common_engine_args(p_cluster)
+    p_cluster.add_argument(
+        "--instance",
+        required=True,
+        help="Instance the dump was taken from; names <instance>.cluster.backup.sql",
+    )
+    p_cluster.add_argument(
+        "--db-user",
+        required=True,
+        help="Superuser of the instance; the dump creates roles and databases",
+    )
+
     p_mdb = sub.add_parser(
         "mariadb", help="Restore a single MariaDB/MySQL-compatible dump"
     )
     _add_common_backup_args(p_mdb)
-    p_mdb.add_argument("--container", required=True)
+    _add_common_engine_args(p_mdb)
     p_mdb.add_argument("--db-name", required=True)
     p_mdb.add_argument("--db-user", default=None, help="Defaults to db-name if omitted")
-    p_mdb.add_argument("--db-password", required=True)
-    p_mdb.add_argument("--empty", action="store_true")
 
     args = parser.parse_args(argv)
 
     try:
         if args.cmd == "files":
-            # target volume = args.volume_name
-            # source volume (backup key) defaults to target volume
             source_volume = args.source_volume or args.volume_name
 
             bp_files = BackupPaths(
@@ -108,6 +124,24 @@ def main(argv: list[str] | None = None) -> int:
                     backups_dir=args.backups_dir,
                 ).sql_file(args.db_name),
                 empty=args.empty,
+                check_version=not args.no_version_check,
+            )
+            return 0
+
+        if args.cmd == "cluster":
+            restore_cluster_sql(
+                container=args.container,
+                user=args.db_user,
+                password=args.db_password,
+                sql_path=BackupPaths(
+                    args.volume_name,
+                    args.backup_hash,
+                    args.version,
+                    repo_name=args.repo_name,
+                    backups_dir=args.backups_dir,
+                ).cluster_file(args.instance),
+                empty=args.empty,
+                check_version=not args.no_version_check,
             )
             return 0
 
@@ -126,13 +160,14 @@ def main(argv: list[str] | None = None) -> int:
                     backups_dir=args.backups_dir,
                 ).sql_file(args.db_name),
                 empty=args.empty,
+                check_version=not args.no_version_check,
             )
             return 0
 
         parser.error("Unhandled command")
         return 2
 
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 - CLI boundary: any failure becomes exit 1
         print(f"ERROR: {e}", file=sys.stderr)
         return 1
 

src/baudolo/restore/db/cluster.py

@@ -0,0 +1,248 @@
+"""Replay a full PostgreSQL cluster dump produced by ``pg_dumpall``.
+
+The backup side writes one when a databases.csv row asks for every database of
+an instance (``database = '*'``, see ``backup/db.py``). Until now nothing read
+it back, so that dump was stored and unrestorable - a format whose producer has
+no consumer.
+
+A cluster stream differs from a single-database one in three ways that decide
+the implementation:
+
+* it recreates roles and databases, so it must be replayed against the control
+  database rather than into a target database;
+* ``CREATE DATABASE`` cannot run inside a transaction block, so unlike
+  :mod:`baudolo.restore.db.postgres` the replay must not be wrapped in
+  ``--single-transaction``;
+* it is replayed as a superuser, so the superuser-only statements that the
+  single-database path filters out are exactly the ones that have to survive.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import tempfile
+from collections.abc import Iterable, Iterator
+
+from ..run import docker_exec
+from .version import guard
+
+CONTROL_DB = "postgres"
+_CLUSTER_PRECLEAN_SQL = os.path.join(os.path.dirname(__file__), "cluster_preclean.sql")
+_CREATE_ROLE = re.compile(rb'^CREATE ROLE "?([^";]+)"?;\s*$')
+_CREATE_DATABASE = re.compile(rb"^CREATE DATABASE\s+(.*)$")
+_CREATE_ROLE_LINE = re.compile(rb"^CREATE ROLE\s+(.*)$")
+_CONNECT = re.compile(rb"^\\connect\s+(.*)$")
+_NO_ROWS = "SELECT ''::text WHERE false"
+
+
+def _first_identifier(rest: str) -> str | None:
+    """The first SQL identifier in *rest*, quoted or bare.
+
+    A quoted identifier may hold spaces and doubled quotes, so it cannot be
+    read with a character class that stops at whitespace - which is how a
+    database called ``odd name`` used to leave the inventory as ``odd``.
+    """
+    text = rest.strip()
+    if not text:
+        return None
+    if text.startswith('"'):
+        out = []
+        index = 1
+        while index < len(text):
+            char = text[index]
+            if char == '"':
+                if index + 1 < len(text) and text[index + 1] == '"':
+                    out.append('"')
+                    index += 2
+                    continue
+                return "".join(out)
+            out.append(char)
+            index += 1
+        return None
+    return re.split(r"[\s;(]", text, maxsplit=1)[0] or None
+
+
+def _connect_target(rest: str) -> str | None:
+    """The database a ``\\connect`` line switches to.
+
+    psql options precede the name (``\\connect -reuse-previous=on dbname=x``),
+    and the name may arrive as a ``dbname=`` assignment rather than bare.
+    """
+    for token in rest.strip().split():
+        if token.startswith("-"):
+            continue
+        if token.startswith("dbname="):
+            return _first_identifier(token[len("dbname=") :])
+        return _first_identifier(rest.strip()[rest.strip().index(token) :])
+    return None
+
+
+def dump_inventory(sql_path: str) -> tuple[list[str], list[str]]:
+    """The databases and roles a cluster dump recreates.
+
+    Args:
+        sql_path: the ``pg_dumpall`` stream.
+
+    Returns:
+        ``(databases, roles)``, each in the order the dump names them. The
+        pre-clean is scoped to these: everything else in the instance belongs
+        to no backup this restore holds, and dropping it would destroy data
+        the replay cannot bring back.
+    """
+    databases: list[str] = []
+    roles: list[str] = []
+    with open(sql_path, "rb") as handle:
+        for raw in handle:
+            line = raw.decode("utf-8", "replace")
+            for pattern, sink, read in (
+                (_CREATE_DATABASE, databases, _first_identifier),
+                (_CONNECT, databases, _connect_target),
+                (_CREATE_ROLE_LINE, roles, _first_identifier),
+            ):
+                found = pattern.match(raw)
+                if not found:
+                    continue
+                name = read(line[found.start(1) :])
+                if name and name not in sink:
+                    sink.append(name)
+    return databases, roles
+
+
+def preclean_sql() -> str:
+    """The catalog-wide pre-clean, safe only behind the instance check."""
+    with open(_CLUSTER_PRECLEAN_SQL, encoding="utf-8") as preclean:
+        return preclean.read()
+
+
+def instance_databases(container: str, user: str, docker_env: dict) -> list[str]:
+    """The instance's own databases, templates and control database aside."""
+    listed = docker_exec(
+        container,
+        [
+            "psql",
+            "-U",
+            user,
+            "-d",
+            CONTROL_DB,
+            "-tAc",
+            (
+                "SELECT datname FROM pg_database "
+                "WHERE NOT datistemplate AND datname <> current_database()"
+            ),
+        ],
+        capture=True,
+        docker_env=docker_env,
+    ).stdout
+    text = listed.decode() if isinstance(listed, bytes) else listed
+    return [name for name in text.split() if name]
+
+
+def assert_instance_matches_dump(
+    container: str, user: str, sql_path: str, docker_env: dict
+) -> None:
+    """Refuse ``--empty`` on an instance holding anything the dump lacks.
+
+    The pre-clean is a catalog-wide sweep, so a foreign database would be
+    destroyed with no way back. Scoping the sweep instead is not a fix: a
+    surviving database that owns or grants to one of the dump's roles pins
+    that role in pg_shdepend, and DROP ROLE then fails after the dump's own
+    databases are already gone.
+
+    Raises:
+        RuntimeError: the instance carries databases this dump cannot restore.
+    """
+    dumped, _roles = dump_inventory(sql_path)
+    present = instance_databases(container, user, docker_env)
+    foreign = sorted(set(present) - set(dumped))
+    if foreign:
+        raise RuntimeError(
+            f"{container} also holds {', '.join(foreign)}, which "
+            f"{os.path.basename(sql_path)} does not carry. --empty wipes the "
+            "instance, so those would be destroyed with nothing to restore "
+            "them from. Move them off this instance, or drop them yourself if "
+            "they are disposable."
+        )
+
+
+def _psql(user: str) -> list[str]:
+    """The replay client: no --single-transaction, CREATE DATABASE forbids it."""
+    return ["psql", "-v", "ON_ERROR_STOP=1", "-U", user, "-d", CONTROL_DB]
+
+
+def filter_own_role_creation(lines: Iterable[bytes], user: str) -> Iterator[bytes]:
+    """Drop the ``CREATE ROLE`` of the role holding this session.
+
+    A pg_dumpall stream recreates every role of the cluster, the bootstrap
+    superuser included, and the pre-clean cannot drop the one it is connected
+    as - so that single statement always collides. Its ``ALTER ROLE`` is kept:
+    that is what re-applies the attributes and the password the dump captured.
+
+    Args:
+        lines: dump lines including their trailing newlines.
+        user: the connecting role.
+
+    Yields:
+        Every line except that one CREATE.
+    """
+    for line in lines:
+        found = _CREATE_ROLE.match(line)
+        if found and found.group(1).decode() == user:
+            continue
+        yield line
+
+
+def restore_cluster_sql(
+    *,
+    container: str,
+    user: str,
+    password: str,
+    sql_path: str,
+    empty: bool,
+    check_version: bool = True,
+) -> None:
+    """Replay a pg_dumpall stream into a running instance.
+
+    Args:
+        container: the running engine to replay into.
+        user: a superuser of that instance; the dump creates roles and
+            databases, which an application role may not do.
+        password: its password, handed to psql through the container's env.
+        sql_path: the ``<instance>.cluster.backup.sql`` of a generation.
+        empty: drop the cluster's databases and roles first. Without it the
+            replay stops at the first object that already exists, which is the
+            honest outcome: recreating a cluster over a populated one is a
+            decision, not a default.
+        check_version: refuse a dump from a newer major version than the
+            running engine before anything is dropped.
+    """
+    if not os.path.isfile(sql_path):
+        raise FileNotFoundError(sql_path)
+
+    if check_version:
+        guard(
+            sql_path=sql_path,
+            engine="postgres",
+            container=container,
+            user=user,
+            password=password,
+        )
+
+    docker_env = {"PGPASSWORD": password}
+
+    if empty:
+        assert_instance_matches_dump(container, user, sql_path, docker_env)
+        docker_exec(
+            container,
+            _psql(user),
+            stdin=preclean_sql().encode(),
+            docker_env=docker_env,
+        )
+
+    with open(sql_path, "rb") as src, tempfile.TemporaryFile() as filtered:
+        for line in filter_own_role_creation(src, user):
+            filtered.write(line)
+        filtered.seek(0)
+        docker_exec(container, _psql(user), stdin=filtered, docker_env=docker_env)
+
+    print(f"PostgreSQL cluster restore complete from '{os.path.basename(sql_path)}'.")

src/baudolo/restore/db/cluster_preclean.sql

@@ -0,0 +1,33 @@
+-- Pre-clean for `restore cluster --empty`. A pg_dumpall stream recreates roles
+-- and databases, so replaying it into a populated cluster dies on the first
+-- CREATE ROLE. Emitted as one DROP per row and run via \gexec so each executes
+-- as its own top-level statement: DROP DATABASE cannot run inside a
+-- transaction block, which rules out a single DO block.
+-- The phase column pins the order: databases must be gone before their owners
+-- can be dropped, and DROP OWNED BY releases what a role still holds in the
+-- control database. Template databases, the control database itself, the pg_*
+-- system roles and the connecting role are kept - the dump does not recreate
+-- them and dropping them would end the session.
+-- The sweep stays catalog-wide on purpose: a scoped one leaves databases that
+-- pin a dumped role in pg_shdepend, and phase 3 then fails after phase 1 has
+-- already dropped. assert_instance_matches_dump refuses before this runs.
+SELECT statement
+  FROM (
+    SELECT 1 AS phase,
+           format('DROP DATABASE IF EXISTS %I', datname) AS statement
+      FROM pg_database
+     WHERE NOT datistemplate
+       AND datname <> current_database()
+    UNION ALL
+    SELECT 2, format('DROP OWNED BY %I', rolname)
+      FROM pg_roles
+     WHERE NOT starts_with(rolname, 'pg_')
+       AND rolname <> current_user
+    UNION ALL
+    SELECT 3, format('DROP ROLE IF EXISTS %I', rolname)
+      FROM pg_roles
+     WHERE NOT starts_with(rolname, 'pg_')
+       AND rolname <> current_user
+  ) drops
+ ORDER BY phase
+\gexec

src/baudolo/restore/db/mariadb.py

@@ -4,6 +4,7 @@ import os
 import sys
 
 from ..run import docker_exec, docker_exec_sh
+from .version import guard
 
 
 def _pick_client(container: str) -> str:
@@ -22,11 +23,11 @@ exit 42
         if not out:
             raise RuntimeError("empty client detection output")
         return out
-    except Exception as e:
+    except Exception:
         print(
             "ERROR: neither 'mariadb' nor 'mysql' found in container.", file=sys.stderr
         )
-        raise e
+        raise
 
 
 def restore_mariadb_sql(
@@ -37,16 +38,25 @@ def restore_mariadb_sql(
     password: str,
     sql_path: str,
     empty: bool,
+    check_version: bool = True,
 ) -> None:
     client = _pick_client(container)
 
     if not os.path.isfile(sql_path):
         raise FileNotFoundError(sql_path)
 
+    if check_version:
+        guard(
+            sql_path=sql_path,
+            engine="mariadb",
+            container=container,
+            user=user,
+            password=password,
+            client=client,
+        )
+
     if empty:
-        # IMPORTANT:
-        # Do NOT hardcode 'mysql' here. Use the detected client.
-        # MariaDB 11 images may not contain the mysql binary at all.
+        # Do not hardcode 'mysql': MariaDB 11 images may not ship that binary.
         result = docker_exec(
             container,
             [

src/baudolo/restore/db/postgres.py

@@ -5,6 +5,7 @@ import tempfile
 from collections.abc import Iterable, Iterator
 
 from ..run import docker_exec
+from .version import guard
 
 _SUPERUSER_ONLY_PREFIXES = (b"COMMENT ON EXTENSION", b"ALTER DEFAULT PRIVILEGES")
 _EMPTY_PRECLEAN_SQL = os.path.join(os.path.dirname(__file__), "empty_preclean.sql")
@@ -46,11 +47,20 @@ def restore_postgres_sql(
     password: str,
     sql_path: str,
     empty: bool,
+    check_version: bool = True,
 ) -> None:
     if not os.path.isfile(sql_path):
         raise FileNotFoundError(sql_path)
 
-    # Make password available INSIDE the container for psql.
+    if check_version:
+        guard(
+            sql_path=sql_path,
+            engine="postgres",
+            container=container,
+            user=user,
+            password=password,
+        )
+
     docker_env = {"PGPASSWORD": password}
 
     if empty:

src/baudolo/restore/db/version.py

@@ -0,0 +1,146 @@
+"""Refuse a dump the target engine is too old to read.
+
+A restore with ``--empty`` destroys before it replays: the pre-clean drops the
+schema in one session and the dump goes in the next, with no rollback across
+the two. A dump the engine cannot parse therefore does not fail harmlessly -
+it leaves an emptied database behind. Comparing the two versions first turns
+that into a refusal.
+
+Both engines state their origin in the dump's own header, and they do not
+state it the same way. Postgres writes ``-- Dumped from database version``
+around line seven. MariaDB opens line two with ``-- MariaDB dump 10.19-11.8.8``,
+where the first number is mariadb-dump's own version, and names the server only
+further down on the tab-separated ``-- Server version`` line. Matching the first
+number in the header would read the tool on one engine and the server on the
+other, so each engine gets its own pattern.
+
+A ``pg_dumpall`` cluster dump has no version line of its own: its header opens
+with the cluster banner and the roles section, and the first
+``-- Dumped from database version`` belongs to the first database's embedded
+``pg_dump`` output, arbitrarily far down. Hence the scan runs to
+``SCAN_LINES`` rather than to a header-sized handful.
+"""
+
+from __future__ import annotations
+
+import re
+
+from ..run import docker_exec, stdout_of
+
+SCAN_LINES = 2000
+DUMP_VERSION = {
+    "postgres": re.compile(r"^-- Dumped from database version (\S+)"),
+    "mariadb": re.compile(r"^-- Server version\s+(\S+)"),
+}
+
+
+class VersionMismatch(Exception):
+    """The dump cannot be replayed into this engine."""
+
+
+def major_of(version: str) -> int:
+    """The major number of an engine version string.
+
+    Args:
+        version: as the engine spells it, e.g. ``17.11`` or
+            ``11.8.8-MariaDB-ubu2404``.
+
+    Raises:
+        VersionMismatch: the string does not start with a number.
+    """
+    leading = re.match(r"(\d+)", version)
+    if not leading:
+        raise VersionMismatch(f"cannot read a major version from '{version}'")
+    return int(leading.group(1))
+
+
+def dump_version(sql_path: str, engine: str) -> str:
+    """Read the engine version a dump was taken from, out of its own header.
+
+    Args:
+        sql_path: the dump to read.
+        engine: ``postgres`` or ``mariadb``.
+
+    Returns:
+        The version string as the dump spells it.
+
+    Raises:
+        VersionMismatch: no version line within the first ``SCAN_LINES``.
+    """
+    pattern = DUMP_VERSION[engine]
+    with open(sql_path, encoding="utf-8", errors="replace") as handle:
+        for _ in range(SCAN_LINES):
+            line = handle.readline()
+            if not line:
+                break
+            found = pattern.search(line)
+            if found:
+                return found.group(1)
+    raise VersionMismatch(
+        f"{sql_path} carries no {engine} version header in its first {SCAN_LINES} lines"
+    )
+
+
+def server_version(
+    container: str, engine: str, user: str, password: str, client: str = ""
+) -> str:
+    """Ask the running engine which version it is."""
+    if engine == "postgres":
+        return stdout_of(
+            docker_exec(
+                container,
+                ["psql", "-U", user, "-tAc", "SHOW server_version"],
+                capture=True,
+                docker_env={"PGPASSWORD": password},
+            )
+        )
+    return stdout_of(
+        docker_exec(
+            container,
+            [
+                client or "mariadb",
+                "-u",
+                user,
+                f"--password={password}",
+                "-N",
+                "-B",
+                "-e",
+                "SELECT VERSION()",
+            ],
+            capture=True,
+        )
+    )
+
+
+def assert_replayable(sql_path: str, engine: str, dumped: str, serving: str) -> None:
+    """Refuse a dump from a newer major version than the target engine.
+
+    Restoring forward across a major version is the upgrade path and stays
+    allowed; backward is refused, because a newer dump uses syntax an older
+    server rejects and the pre-clean would already have dropped the schema.
+
+    Raises:
+        VersionMismatch: the dump is newer than the engine.
+    """
+    if major_of(dumped) > major_of(serving):
+        raise VersionMismatch(
+            f"{sql_path} came from {engine} {dumped} but {serving} is running; "
+            "a newer dump does not replay into an older engine, and --empty "
+            "would drop the schema before finding out"
+        )
+
+
+def guard(
+    *,
+    sql_path: str,
+    engine: str,
+    container: str,
+    user: str,
+    password: str,
+    client: str = "",
+) -> None:
+    """Compare the dump's origin against the running engine before replaying."""
+    dumped = dump_version(sql_path, engine)
+    serving = server_version(container, engine, user, password, client)
+    assert_replayable(sql_path, engine, dumped, serving)
+    print(f"OK: dump is from {engine} {dumped}, {serving} is serving.")

src/baudolo/restore/files.py

@@ -1,9 +1,23 @@
+"""Restore a volume's file tree by writing into its mountpoint.
+
+That shortcut only holds for a plain local volume, where the mountpoint *is*
+the storage. A volume with driver options - NFS, a bind device, tmpfs - keeps
+the same ``/var/lib/docker/volumes/<name>/_data`` path, but docker mounts the
+real backing store over it on demand and unmounts it again when the last
+consumer stops. Writing there while nothing has it mounted lands in the empty
+directory underneath, is hidden by the next mount, and rsync reports success.
+"""
+
 from __future__ import annotations
 
 import os
 import sys
 
-from .run import docker_volume_exists, run
+from .run import docker_volume_exists, run, stdout_of
+
+INSPECT_FORMAT = (
+    "{{ .Mountpoint }}|{{ .Driver }}|{{ if .Options }}opts{{ else }}plain{{ end }}"
+)
 
 
 def restore_volume_files(volume_name: str, backup_files_dir: str) -> int:
@@ -18,11 +32,11 @@ def restore_volume_files(volume_name: str, backup_files_dir: str) -> int:
         print(f"Volume {volume_name} already exists.")
 
     cp = run(
-        ["docker", "volume", "inspect", "--format", "{{ .Mountpoint }}", volume_name],
+        ["docker", "volume", "inspect", "--format", INSPECT_FORMAT, volume_name],
         capture=True,
     )
-    raw = cp.stdout or b""
-    mountpoint = (raw.decode() if isinstance(raw, bytes) else raw).strip()
+    fields = stdout_of(cp).split("|")
+    mountpoint = fields[0] if fields else ""
     if not mountpoint:
         print(
             f"ERROR: could not resolve mountpoint for volume {volume_name}",
@@ -30,6 +44,17 @@ def restore_volume_files(volume_name: str, backup_files_dir: str) -> int:
         )
         return 2
 
+    driver, options = (fields + ["local", "plain"])[1:3]
+    if (driver != "local" or options == "opts") and not os.path.ismount(mountpoint):
+        print(
+            f"ERROR: volume {volume_name} has a backing store of its own "
+            f"(driver {driver}) but nothing has it mounted; writing to "
+            f"{mountpoint} now would land under the mount and be lost. "
+            "Start a container that mounts the volume, then restore again.",
+            file=sys.stderr,
+        )
+        return 2
+
     src = os.path.join(backup_files_dir, "")
     dest = os.path.join(mountpoint, "")
     run(["rsync", "-avv", "--delete", src, dest])

src/baudolo/restore/paths.py

@@ -27,3 +27,7 @@ class BackupPaths:
 
     def sql_file(self, db_name: str) -> str:
         return os.path.join(self.root(), "sql", f"{db_name}.backup.sql")
+
+    def cluster_file(self, instance: str) -> str:
+        """The pg_dumpall stream a `database = '*'` row produces."""
+        return os.path.join(self.root(), "sql", f"{instance}.cluster.backup.sql")

src/baudolo/restore/run.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import subprocess
 import sys
-from typing import Optional
 
 
 def run(
@@ -10,7 +9,7 @@ def run(
     *,
     stdin=None,
     capture: bool = False,
-    env: Optional[dict] = None,
+    env: dict | None = None,
 ) -> subprocess.CompletedProcess:
     try:
         kwargs: dict = {
@@ -26,32 +25,35 @@ def run(
         else:
             kwargs["stdin"] = stdin
 
-        return subprocess.run(cmd, **kwargs)
+        return subprocess.run(cmd, **kwargs)  # noqa: PLW1510 - check lives in kwargs
 
     except subprocess.CalledProcessError as e:
         msg = f"ERROR: command failed ({e.returncode}): {' '.join(cmd)}"
         print(msg, file=sys.stderr)
-        if e.stdout:
+        for stream in (e.stdout, e.stderr):
+            if not stream:
+                continue
             try:
-                print(e.stdout.decode(), file=sys.stderr)
-            except Exception:
-                print(e.stdout, file=sys.stderr)
-        if e.stderr:
-            try:
-                print(e.stderr.decode(), file=sys.stderr)
-            except Exception:
-                print(e.stderr, file=sys.stderr)
+                print(stream.decode(), file=sys.stderr)
+            except (UnicodeDecodeError, AttributeError):
+                print(stream, file=sys.stderr)
         raise
 
 
+def stdout_of(completed: subprocess.CompletedProcess) -> str:
+    """The captured stdout as stripped text, whether it came back bytes or str."""
+    raw = completed.stdout or b""
+    return (raw.decode() if isinstance(raw, bytes) else raw).strip()
+
+
 def docker_exec(
     container: str,
     argv: list[str],
     *,
     stdin=None,
     capture: bool = False,
-    env: Optional[dict] = None,
-    docker_env: Optional[dict[str, str]] = None,
+    env: dict | None = None,
+    docker_env: dict[str, str] | None = None,
 ) -> subprocess.CompletedProcess:
     cmd: list[str] = ["docker", "exec", "-i"]
     if docker_env:
@@ -67,8 +69,8 @@ def docker_exec_sh(
     *,
     stdin=None,
     capture: bool = False,
-    env: Optional[dict] = None,
-    docker_env: Optional[dict[str, str]] = None,
+    env: dict | None = None,
+    docker_env: dict[str, str] | None = None,
 ) -> subprocess.CompletedProcess:
     return docker_exec(
         container,
@@ -85,5 +87,6 @@ def docker_volume_exists(volume: str) -> bool:
         ["docker", "volume", "inspect", volume],
         stdout=subprocess.DEVNULL,
         stderr=subprocess.DEVNULL,
+        check=False,
     )
     return p.returncode == 0

src/baudolo/seed/__main__.py

@@ -1,18 +1,17 @@
-#!/usr/bin/env python3
 from __future__ import annotations
 
 import argparse
 import os
 import re
 import sys
+
 import pandas as pd
-from typing import Optional
 from pandas.errors import EmptyDataError
 
 DB_NAME_RE = re.compile(r"^[a-zA-Z0-9_][a-zA-Z0-9_-]*$")
 
 
-def _validate_database_value(value: Optional[str], *, instance: str) -> str:
+def _validate_database_value(value: str | None, *, instance: str) -> str:
     v = (value or "").strip()
     if v == "":
         raise ValueError(
@@ -40,7 +39,7 @@ def _empty_df() -> pd.DataFrame:
 def check_and_add_entry(
     file_path: str,
     instance: str,
-    database: Optional[str],
+    database: str | None,
     username: str,
     password: str,
 ) -> None:
@@ -108,7 +107,7 @@ def main() -> None:
             username=args.username,
             password=args.password,
         )
-    except Exception as exc:
+    except Exception as exc:  # noqa: BLE001 - CLI boundary: any failure becomes exit 1
         print(f"ERROR: {exc}", file=sys.stderr)
         sys.exit(1)
 

tests/e2e/faithful_driver.py

@@ -0,0 +1,92 @@
+"""Show what a real snapshot holds for a volume that has its own storage.
+
+Runs inside the privileged container that built the btrfs subject. Prints one
+PASS/FAIL line per assertion and exits non-zero on the first failure.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+sys.path.insert(0, "/src")
+
+from baudolo.backup.snapshot import (
+    SnapshotError,
+    snapshot_source,
+    unsnapshotted,
+    volume_snapshot,
+)
+from baudolo.backup.volume import Backing
+
+SUBJECT = sys.argv[1]
+
+
+def shell(command: str) -> list[str]:
+    proc = subprocess.run(
+        command, shell=True, capture_output=True, text=True, check=False
+    )
+    if proc.returncode != 0:
+        raise SnapshotError(
+            f"{command} exited {proc.returncode}: {proc.stderr.strip()}"
+        )
+    return proc.stdout.splitlines()
+
+
+def check(label: str, condition: bool) -> None:
+    print(f"{'PASS' if condition else 'FAIL'} {label}", flush=True)
+    if not condition:
+        sys.exit(1)
+
+
+def volume(name: str, payload: str) -> Path:
+    path = Path(SUBJECT) / "volumes" / name / "_data"
+    path.mkdir(parents=True, exist_ok=True)
+    (path / "state").write_text(payload)
+    return path
+
+
+plain = volume("plain", "plain-payload")
+own = Path(SUBJECT) / "volumes" / "own" / "_data"
+own.mkdir(parents=True, exist_ok=True)
+shell(f"mount -t tmpfs tmpfs {own}")
+(own / "state").write_text("own-payload")
+
+check("a plain volume is captured", unsnapshotted(Backing(str(plain)), SUBJECT) is None)
+check(
+    "a volume on a mount of its own is not",
+    unsnapshotted(Backing(str(own)), SUBJECT) is not None,
+)
+check(
+    "a declared backing store is not, mounted or not",
+    unsnapshotted(Backing(str(plain), options={"type": "nfs"}), SUBJECT) is not None,
+)
+check(
+    "a foreign driver is not",
+    unsnapshotted(Backing(str(plain), driver="rexray"), SUBJECT) is not None,
+)
+
+with volume_snapshot("btrfs", SUBJECT, "e2e", run=shell) as resolve:
+    frozen_plain = Path(resolve(str(plain)))
+    frozen_own = Path(resolve(str(own)))
+
+    check(
+        "the snapshot carries the plain volume",
+        (frozen_plain / "state").read_text() == "plain-payload",
+    )
+    check(
+        "the snapshot shows the other volume as an empty directory",
+        frozen_own.is_dir() and not any(frozen_own.iterdir()),
+    )
+
+    source, reason = snapshot_source(resolve, Backing(str(plain)), SUBJECT)
+    check(
+        "the plain volume is read from the snapshot",
+        source is not None and source.rstrip("/") == str(frozen_plain),
+    )
+
+    source, reason = snapshot_source(resolve, Backing(str(own)), SUBJECT)
+    check(f"the other volume degrades to live: {reason[:60]}", source is None)
+
+print("ALL OK", flush=True)

tests/e2e/helpers/__init__.py

@@ -1,4 +1,4 @@
 """Shared e2e helpers, re-exported so tests import one name."""
 
-from .fixtures import *  # noqa: F401,F403
-from .process import *  # noqa: F401,F403
+from .fixtures import *
+from .process import *

tests/e2e/helpers/fixtures.py

@@ -97,8 +97,7 @@ def write_databases_csv(path: str, rows: list[tuple[str, str, str, str]]) -> Non
     Path(path).parent.mkdir(parents=True, exist_ok=True)
     with open(path, "w", encoding="utf-8") as f:
         f.write("instance;database;username;password\n")
-        for inst, db, user, pw in rows:
-            f.write(f"{inst};{db};{user};{pw}\n")
+        f.writelines(f"{inst};{db};{user};{pw}\n" for inst, db, user, pw in rows)
 
 
 def cleanup_docker(*, containers: list[str], volumes: list[str]) -> None:

tests/e2e/snapshot_db_driver.py

@@ -12,8 +12,8 @@ import sys
 
 sys.path.insert(0, "/src")
 
-from baudolo.backup.snapshot import SnapshotError, volume_snapshot  # noqa: E402
-from baudolo.backup.volume import backup_volume  # noqa: E402
+from baudolo.backup.snapshot import SnapshotError, volume_snapshot
+from baudolo.backup.volume import backup_volume
 
 SUBJECT = "/subject/docker"
 VOLUME = "mariadb_data"
@@ -23,7 +23,9 @@ GENERATION = f"{VERSIONS}/20260731"
 
 
 def shell(command: str) -> list[str]:
-    proc = subprocess.run(command, shell=True, capture_output=True, text=True)
+    proc = subprocess.run(
+        command, shell=True, capture_output=True, text=True, check=False
+    )
     if proc.returncode != 0:
         raise SnapshotError(
             f"{command} exited {proc.returncode}: {proc.stderr.strip()}"

tests/e2e/snapshot_driver.py

@@ -12,7 +12,7 @@ from pathlib import Path
 
 sys.path.insert(0, "/src")
 
-from baudolo.backup.snapshot import SnapshotError, volume_snapshot  # noqa: E402
+from baudolo.backup.snapshot import SnapshotError, volume_snapshot
 
 KIND = sys.argv[1]
 SUBJECT = sys.argv[2]
@@ -20,7 +20,9 @@ EXPECT = sys.argv[3]
 
 
 def shell(command: str) -> list[str]:
-    proc = subprocess.run(command, shell=True, capture_output=True, text=True)
+    proc = subprocess.run(
+        command, shell=True, capture_output=True, text=True, check=False
+    )
     if proc.returncode != 0:
         raise SnapshotError(
             f"{command} exited {proc.returncode}: {proc.stderr.strip()}"

tests/e2e/test_e2e_dump_only_fallback_to_files.py

@@ -1,9 +1,8 @@
-# tests/e2e/test_e2e_dump_only_fallback_to_files.py
 import unittest
 
 from .helpers import (
-    POSTGRES_IMAGE,
     POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
     backup_path,
     cleanup_docker,
     create_minimal_compose_dir,
@@ -12,8 +11,8 @@ from .helpers import (
     require_docker,
     run,
     unique,
-    write_databases_csv,
     wait_for_postgres,
+    write_databases_csv,
 )
 
 
@@ -37,7 +36,6 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
 
         run(["docker", "volume", "create", cls.pg_volume])
 
-        # Start Postgres (creates a real DB volume)
         run(
             [
                 "docker",

tests/e2e/test_e2e_dump_only_sql_mixed_run.py

@@ -1,8 +1,8 @@
 import unittest
 
 from .helpers import (
-    POSTGRES_IMAGE,
     POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
     backup_path,
     cleanup_docker,
     create_minimal_compose_dir,
@@ -35,7 +35,6 @@ class TestE2EDumpOnlySqlMixedRun(unittest.TestCase):
         cls.containers: list[str] = []
         cls.volumes = [cls.db_volume, cls.files_volume]
 
-        # Create volumes
         run(["docker", "volume", "create", cls.db_volume])
         run(["docker", "volume", "create", cls.files_volume])
 
@@ -114,7 +113,6 @@ class TestE2EDumpOnlySqlMixedRun(unittest.TestCase):
             [(cls.pg_container, cls.pg_db, cls.pg_user, cls.pg_password)],
         )
 
-        # Run baudolo with dump-only-sql
         cmd = [
             "baudolo",
             "--compose-dir",

tests/e2e/test_e2e_engine_detection_registry_prefix.py

@@ -1,19 +1,19 @@
 import unittest
 
 from .helpers import (
-    POSTGRES_IMAGE,
     POSTGRES_DATA_DIR,
-    backup_run,
+    POSTGRES_IMAGE,
     backup_path,
+    backup_run,
     cleanup_docker,
     create_minimal_compose_dir,
     ensure_empty_dir,
     latest_version_dir,
     require_docker,
-    unique,
-    write_databases_csv,
     run,
+    unique,
     wait_for_postgres,
+    write_databases_csv,
 )
 
 REGISTRY_HOST = "svc-db-mariadb-swarm-mgr-01:5000"

tests/e2e/test_e2e_files_full.py

@@ -1,16 +1,16 @@
 import unittest
 
 from .helpers import (
-    backup_run,
     backup_path,
+    backup_run,
     cleanup_docker,
     create_minimal_compose_dir,
     ensure_empty_dir,
     latest_version_dir,
     require_docker,
+    run,
     unique,
     write_databases_csv,
-    run,
 )
 
 
@@ -30,7 +30,6 @@ class TestE2EFilesFull(unittest.TestCase):
         cls.containers = []
         cls.volumes = [cls.volume_src, cls.volume_dst]
 
-        # create source volume with a file
         run(["docker", "volume", "create", cls.volume_src])
         run(
             [
@@ -50,7 +49,6 @@ class TestE2EFilesFull(unittest.TestCase):
         cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
         write_databases_csv(cls.databases_csv, [])
 
-        # Run backup (files should be copied)
         backup_run(
             backups_dir=cls.backups_dir,
             repo_name=cls.repo_name,
@@ -97,7 +95,6 @@ class TestE2EFilesFull(unittest.TestCase):
             ]
         )
 
-        # verify restored file exists in dst volume
         p = run(
             [
                 "docker",

tests/e2e/test_e2e_files_no_copy.py

@@ -1,16 +1,16 @@
 import unittest
 
 from .helpers import (
-    backup_run,
     backup_path,
+    backup_run,
     cleanup_docker,
     create_minimal_compose_dir,
     ensure_empty_dir,
     latest_version_dir,
     require_docker,
+    run,
     unique,
     write_databases_csv,
-    run,
 )
 
 
@@ -29,7 +29,6 @@ class TestE2EFilesNoCopy(unittest.TestCase):
         cls.containers: list[str] = []
         cls.volumes = [cls.volume_src]
 
-        # Create source volume and write a marker file
         run(["docker", "volume", "create", cls.volume_src])
         run(
             [

tests/e2e/test_e2e_images_no_backup_required_early_skip.py

@@ -1,4 +1,3 @@
-# tests/e2e/test_e2e_images_no_backup_required_early_skip.py
 import unittest
 
 from .helpers import (
@@ -34,11 +33,9 @@ class TestE2EImagesNoBackupRequiredEarlySkip(unittest.TestCase):
         cls.containers = [cls.redis_container]
         cls.volumes = [cls.ignored_volume, cls.normal_volume]
 
-        # Create volumes
         run(["docker", "volume", "create", cls.ignored_volume])
         run(["docker", "volume", "create", cls.normal_volume])
 
-        # Start redis container using the ignored volume
         run(
             [
                 "docker",
@@ -71,7 +68,6 @@ class TestE2EImagesNoBackupRequiredEarlySkip(unittest.TestCase):
         cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
         write_databases_csv(cls.databases_csv, [])
 
-        # Run baudolo with images-no-backup-required redis
         cmd = [
             "baudolo",
             "--compose-dir",

tests/e2e/test_e2e_mariadb_anonymous_preemption.py

@@ -30,8 +30,8 @@ import pandas
 from baudolo.backup import db as db_mod
 
 from .helpers import (
-    MARIADB_IMAGE,
     MARIADB_DATA_DIR,
+    MARIADB_IMAGE,
     cleanup_docker,
     require_docker,
     run,

tests/e2e/test_e2e_mariadb_full.py

@@ -1,21 +1,20 @@
-# tests/e2e/test_e2e_mariadb_full.py
 import unittest
 
 from .helpers import (
-    MARIADB_IMAGE,
     MARIADB_DATA_DIR,
-    backup_run,
+    MARIADB_IMAGE,
     backup_path,
+    backup_run,
     cleanup_docker,
     create_minimal_compose_dir,
     ensure_empty_dir,
     latest_version_dir,
     require_docker,
-    unique,
-    write_databases_csv,
     run,
+    unique,
     wait_for_mariadb,
     wait_for_mariadb_sql,
+    write_databases_csv,
 )
 
 
@@ -71,7 +70,6 @@ class TestE2EMariaDBFull(unittest.TestCase):
             cls.db_container, user=cls.db_user, password=cls.db_password, timeout_s=90
         )
 
-        # Create table + data via the dedicated user (TCP)
         run(
             [
                 "docker",
@@ -79,9 +77,11 @@ class TestE2EMariaDBFull(unittest.TestCase):
                 cls.db_container,
                 "sh",
                 "-lc",
-                f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
-                f'-e "CREATE TABLE {cls.db_name}.t (id INT PRIMARY KEY, v VARCHAR(50)); '
-                f"INSERT INTO {cls.db_name}.t VALUES (1,'ok');\"",
+                (
+                    f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
+                    f'-e "CREATE TABLE {cls.db_name}.t (id INT PRIMARY KEY, v VARCHAR(50)); '
+                    f"INSERT INTO {cls.db_name}.t VALUES (1,'ok');\""
+                ),
             ]
         )
 
@@ -112,8 +112,10 @@ class TestE2EMariaDBFull(unittest.TestCase):
                 cls.db_container,
                 "sh",
                 "-lc",
-                f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
-                f'-e "DROP TABLE {cls.db_name}.t;"',
+                (
+                    f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
+                    f'-e "DROP TABLE {cls.db_name}.t;"'
+                ),
             ]
         )
 
@@ -161,8 +163,10 @@ class TestE2EMariaDBFull(unittest.TestCase):
                 self.db_container,
                 "sh",
                 "-lc",
-                f"mariadb -h 127.0.0.1 -u{self.db_user} -p{self.db_password} "
-                f'-N -e "SELECT v FROM {self.db_name}.t WHERE id=1;"',
+                (
+                    f"mariadb -h 127.0.0.1 -u{self.db_user} -p{self.db_password} "
+                    f'-N -e "SELECT v FROM {self.db_name}.t WHERE id=1;"'
+                ),
             ]
         )
         self.assertEqual((p.stdout or "").strip(), "ok")

tests/e2e/test_e2e_mariadb_no_copy.py

@@ -1,21 +1,20 @@
-# tests/e2e/test_e2e_mariadb_no_copy.py
 import unittest
 
 from .helpers import (
-    MARIADB_IMAGE,
     MARIADB_DATA_DIR,
-    backup_run,
+    MARIADB_IMAGE,
     backup_path,
+    backup_run,
     cleanup_docker,
     create_minimal_compose_dir,
     ensure_empty_dir,
     latest_version_dir,
     require_docker,
-    unique,
-    write_databases_csv,
     run,
+    unique,
     wait_for_mariadb,
     wait_for_mariadb_sql,
+    write_databases_csv,
 )
 
 
@@ -69,7 +68,6 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
             cls.db_container, user=cls.db_user, password=cls.db_password, timeout_s=90
         )
 
-        # Create table + data (TCP)
         run(
             [
                 "docker",
@@ -77,9 +75,11 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
                 cls.db_container,
                 "sh",
                 "-lc",
-                f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
-                f'-e "CREATE TABLE {cls.db_name}.t (id INT PRIMARY KEY, v VARCHAR(50)); '
-                f"INSERT INTO {cls.db_name}.t VALUES (1,'ok');\"",
+                (
+                    f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
+                    f'-e "CREATE TABLE {cls.db_name}.t (id INT PRIMARY KEY, v VARCHAR(50)); '
+                    f"INSERT INTO {cls.db_name}.t VALUES (1,'ok');\""
+                ),
             ]
         )
 
@@ -110,8 +110,10 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
                 cls.db_container,
                 "sh",
                 "-lc",
-                f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
-                f'-e "DROP TABLE {cls.db_name}.t;"',
+                (
+                    f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
+                    f'-e "DROP TABLE {cls.db_name}.t;"'
+                ),
             ]
         )
 
@@ -158,8 +160,10 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
                 self.db_container,
                 "sh",
                 "-lc",
-                f"mariadb -h 127.0.0.1 -u{self.db_user} -p{self.db_password} "
-                f'-N -e "SELECT v FROM {self.db_name}.t WHERE id=1;"',
+                (
+                    f"mariadb -h 127.0.0.1 -u{self.db_user} -p{self.db_password} "
+                    f'-N -e "SELECT v FROM {self.db_name}.t WHERE id=1;"'
+                ),
             ]
         )
         self.assertEqual((p.stdout or "").strip(), "ok")

tests/e2e/test_e2e_postgres_cluster_restore.py

@@ -0,0 +1,215 @@
+import unittest
+
+from .helpers import (
+    POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
+    backup_path,
+    backup_run,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    wait_for_postgres,
+    write_databases_csv,
+)
+
+# One statement per entry: psql wraps a multi-statement -c in a transaction,
+# and CREATE DATABASE is forbidden inside one.
+SEED_SQL = (
+    "CREATE ROLE app LOGIN PASSWORD 'apppw'",
+    "CREATE DATABASE first OWNER app",
+    "CREATE DATABASE second OWNER app",
+)
+SIBLING_SQL = (
+    "CREATE ROLE neighbour LOGIN PASSWORD 'neighbourpw'",
+    "CREATE DATABASE sibling OWNER neighbour",
+)
+SIBLING_PAYLOAD = "CREATE TABLE t (v text); INSERT INTO t VALUES ('sibling-payload');"
+FIRST_SQL = "CREATE TABLE t (v text); INSERT INTO t VALUES ('first-payload');"
+SECOND_SQL = "CREATE TABLE t (v text); INSERT INTO t VALUES ('second-payload');"
+
+
+class TestE2EPostgresClusterRestore(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-pg-cluster")
+        cls.backups_dir = f"/tmp/{cls.prefix}/Backups"
+        ensure_empty_dir(cls.backups_dir)
+        cls.compose_dir = create_minimal_compose_dir(f"/tmp/{cls.prefix}")
+        cls.repo_name = cls.prefix
+
+        cls.pg_container = f"{cls.prefix}-pg"
+        cls.pg_volume = f"{cls.prefix}-pg-vol"
+        cls.containers = [cls.pg_container]
+        cls.volumes = [cls.pg_volume]
+
+        run(["docker", "volume", "create", cls.pg_volume])
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.pg_container,
+                "-e",
+                "POSTGRES_PASSWORD=pgpw",
+                "-v",
+                f"{cls.pg_volume}:{POSTGRES_DATA_DIR}",
+                POSTGRES_IMAGE,
+            ]
+        )
+        wait_for_postgres(cls.pg_container, user="postgres")
+
+        for statement in SEED_SQL:
+            cls._psql("postgres", statement)
+        cls._psql("first", FIRST_SQL)
+        cls._psql("second", SECOND_SQL)
+
+        cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
+        write_databases_csv(
+            cls.databases_csv, [(cls.pg_container, "*", "postgres", "pgpw")]
+        )
+
+        backup_run(
+            backups_dir=cls.backups_dir,
+            repo_name=cls.repo_name,
+            compose_dir=cls.compose_dir,
+            databases_csv=cls.databases_csv,
+            database_containers=[cls.pg_container],
+            images_no_stop_required=[POSTGRES_IMAGE],
+        )
+        cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
+        cls.dump = (
+            backup_path(cls.backups_dir, cls.repo_name, cls.version, cls.pg_volume)
+            / "sql"
+            / f"{cls.pg_container}.cluster.backup.sql"
+        )
+
+        run(
+            [
+                "baudolo-restore",
+                "cluster",
+                cls.pg_volume,
+                cls.hash,
+                cls.version,
+                "--backups-dir",
+                cls.backups_dir,
+                "--repo-name",
+                cls.repo_name,
+                "--container",
+                cls.pg_container,
+                "--instance",
+                cls.pg_container,
+                "--db-user",
+                "postgres",
+                "--db-password",
+                "pgpw",
+                "--empty",
+            ]
+        )
+
+        for statement in SIBLING_SQL:
+            cls._psql("postgres", statement)
+        cls._psql("sibling", SIBLING_PAYLOAD)
+        cls.refused = run(cls._restore_argv(), check=False)
+
+    @classmethod
+    def _restore_argv(cls) -> list:
+        return [
+            "baudolo-restore",
+            "cluster",
+            cls.pg_volume,
+            cls.hash,
+            cls.version,
+            "--backups-dir",
+            cls.backups_dir,
+            "--repo-name",
+            cls.repo_name,
+            "--container",
+            cls.pg_container,
+            "--instance",
+            cls.pg_container,
+            "--db-user",
+            "postgres",
+            "--db-password",
+            "pgpw",
+            "--empty",
+        ]
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=cls.containers, volumes=cls.volumes)
+
+    @classmethod
+    def _psql(cls, database: str, sql: str) -> str:
+        p = run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                f'psql -U postgres -d {database} -t -A -c "{sql}"',
+            ]
+        )
+        return (p.stdout or "").strip()
+
+    def test_the_backup_wrote_a_cluster_dump(self) -> None:
+        self.assertTrue(self.dump.is_file(), f"no cluster dump at {self.dump}")
+
+    def test_the_preclean_really_dropped_a_populated_cluster(self) -> None:
+        self.assertEqual(self._psql("first", "SELECT v FROM t"), "first-payload")
+
+    def test_a_second_empty_is_refused_once_a_foreign_database_exists(self) -> None:
+        self.assertNotEqual(self.refused.returncode, 0, self.refused.stdout)
+        self.assertIn("sibling", self.refused.stderr)
+
+    def test_the_refusal_left_the_foreign_database_alone(self) -> None:
+        self.assertEqual(self._psql("sibling", "SELECT v FROM t"), "sibling-payload")
+
+    def test_the_refusal_dropped_nothing_of_its_own(self) -> None:
+        self.assertEqual(self._psql("first", "SELECT v FROM t"), "first-payload")
+
+    def test_both_databases_are_back(self) -> None:
+        listed = self._psql(
+            "postgres",
+            "SELECT datname FROM pg_database WHERE datname IN ('first','second') ORDER BY 1",
+        )
+        self.assertEqual(listed.split(), ["first", "second"])
+
+    def test_each_database_carries_its_own_payload(self) -> None:
+        self.assertEqual(self._psql("first", "SELECT v FROM t"), "first-payload")
+        self.assertEqual(self._psql("second", "SELECT v FROM t"), "second-payload")
+
+    def test_the_superusers_own_create_was_filtered(self) -> None:
+        self.assertEqual(
+            self._psql(
+                "postgres", "SELECT rolsuper FROM pg_roles WHERE rolname = 'postgres'"
+            ),
+            "t",
+        )
+
+    def test_the_owning_role_is_back(self) -> None:
+        self.assertEqual(
+            self._psql(
+                "postgres", "SELECT rolname FROM pg_roles WHERE rolname = 'app'"
+            ),
+            "app",
+        )
+
+    def test_ownership_survived(self) -> None:
+        self.assertEqual(
+            self._psql(
+                "postgres",
+                "SELECT pg_get_userbyid(datdba) FROM pg_database WHERE datname = 'first'",
+            ),
+            "app",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/e2e/test_e2e_postgres_empty_drop_hard.py

@@ -1,9 +1,8 @@
-# tests/e2e/test_e2e_postgres_empty_drop_hard.py
 import unittest
 
 from .helpers import (
-    POSTGRES_IMAGE,
     POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
     backup_run,
     cleanup_docker,
     create_minimal_compose_dir,

tests/e2e/test_e2e_postgres_full.py

@@ -1,20 +1,19 @@
-# tests/e2e/test_e2e_postgres_full.py
 import unittest
 
 from .helpers import (
-    POSTGRES_IMAGE,
     POSTGRES_DATA_DIR,
-    backup_run,
+    POSTGRES_IMAGE,
     backup_path,
+    backup_run,
     cleanup_docker,
     create_minimal_compose_dir,
     ensure_empty_dir,
     latest_version_dir,
     require_docker,
-    unique,
-    write_databases_csv,
     run,
+    unique,
     wait_for_postgres,
+    write_databases_csv,
 )
 
 
@@ -55,7 +54,6 @@ class TestE2EPostgresFull(unittest.TestCase):
         )
         wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
 
-        # Create a table + data
         run(
             [
                 "docker",

tests/e2e/test_e2e_postgres_no_copy.py

@@ -1,20 +1,19 @@
-# tests/e2e/test_e2e_postgres_no_copy.py
 import unittest
 
 from .helpers import (
-    POSTGRES_IMAGE,
     POSTGRES_DATA_DIR,
-    backup_run,
+    POSTGRES_IMAGE,
     backup_path,
+    backup_run,
     cleanup_docker,
     create_minimal_compose_dir,
     ensure_empty_dir,
     latest_version_dir,
     require_docker,
-    unique,
-    write_databases_csv,
     run,
+    unique,
     wait_for_postgres,
+    write_databases_csv,
 )
 
 

tests/e2e/test_e2e_postgres_single_transaction_live_writer.py

@@ -1,9 +1,8 @@
-# tests/e2e/test_e2e_postgres_single_transaction_live_writer.py
 import unittest
 
 from .helpers import (
-    POSTGRES_IMAGE,
     POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
     backup_run,
     cleanup_docker,
     create_minimal_compose_dir,

tests/e2e/test_e2e_restore_files_backing_store.py

@@ -0,0 +1,120 @@
+"""Restoring files into a volume that has a backing store of its own.
+
+Docker keeps the same ``/var/lib/docker/volumes/<name>/_data`` path for such a
+volume and mounts the real storage over it only while a container holds it.
+Writing there unmounted lands in the empty directory underneath, is hidden by
+the next mount, and rsync reports success - so the restore has to refuse.
+"""
+
+import unittest
+from pathlib import Path
+
+from .helpers import (
+    backup_path,
+    cleanup_docker,
+    ensure_empty_dir,
+    machine_hash,
+    require_docker,
+    run,
+    unique,
+)
+
+MARKER = "restored-payload"
+VERSION = "20260817000000"
+
+
+def mountpoint_of(volume: str) -> Path:
+    return Path("/var/lib/docker/volumes") / volume / "_data"
+
+
+def contents(directory: Path) -> list[str]:
+    return sorted(p.name for p in directory.iterdir()) if directory.is_dir() else []
+
+
+class TestE2ERestoreFilesBackingStore(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-backing")
+        cls.repo_name = cls.prefix
+        cls.backups_dir = f"/tmp/{cls.prefix}/Backups"
+        cls.backing = Path(f"/tmp/{cls.prefix}/backing")
+        ensure_empty_dir(cls.backups_dir)
+        ensure_empty_dir(str(cls.backing))
+
+        cls.bound_volume = f"{cls.prefix}-bound"
+        cls.plain_volume = f"{cls.prefix}-plain"
+        cls.volumes = [cls.bound_volume, cls.plain_volume]
+
+        for volume in cls.volumes:
+            files = (
+                backup_path(cls.backups_dir, cls.repo_name, VERSION, volume) / "files"
+            )
+            files.mkdir(parents=True, exist_ok=True)
+            (files / "marker.txt").write_text(MARKER, encoding="utf-8")
+
+        run(
+            [
+                "docker",
+                "volume",
+                "create",
+                "--driver",
+                "local",
+                "--opt",
+                "type=none",
+                "--opt",
+                "o=bind",
+                "--opt",
+                f"device={cls.backing}",
+                cls.bound_volume,
+            ]
+        )
+        run(["docker", "volume", "create", cls.plain_volume])
+
+        cls.refused = cls.restore(cls.bound_volume)
+        cls.accepted = cls.restore(cls.plain_volume)
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=[], volumes=cls.volumes)
+
+    @classmethod
+    def restore(cls, volume: str):
+        return run(
+            [
+                "baudolo-restore",
+                "files",
+                volume,
+                machine_hash(),
+                VERSION,
+                "--backups-dir",
+                cls.backups_dir,
+                "--repo-name",
+                cls.repo_name,
+            ],
+            check=False,
+        )
+
+    def test_a_volume_with_its_own_backing_store_is_refused(self) -> None:
+        self.assertEqual(self.refused.returncode, 2, self.refused.stdout)
+        self.assertIn("backing store of its own", self.refused.stderr)
+
+    def test_nothing_was_written_into_the_backing_store(self) -> None:
+        self.assertEqual(contents(self.backing), [])
+
+    def test_nothing_was_written_under_the_mount_either(self) -> None:
+        self.assertEqual(
+            contents(mountpoint_of(self.bound_volume)),
+            [],
+            "the copy landed in the directory the next mount hides",
+        )
+
+    def test_a_plain_volume_is_still_restored(self) -> None:
+        self.assertEqual(self.accepted.returncode, 0, self.accepted.stderr)
+        restored = mountpoint_of(self.plain_volume) / "marker.txt"
+        self.assertTrue(restored.is_file(), f"{restored} missing")
+        self.assertEqual(restored.read_text(encoding="utf-8"), MARKER)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/e2e/test_e2e_seed_star_and_db_entries_backup_postgres.py

@@ -1,8 +1,8 @@
 import unittest
 
 from .helpers import (
-    POSTGRES_IMAGE,
     POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
     backup_path,
     cleanup_docker,
     create_minimal_compose_dir,

tests/e2e/test_e2e_snapshot.py

@@ -17,6 +17,7 @@ from .helpers import require_docker, run, unique
 
 REPO_SRC = Path(__file__).resolve().parents[2] / "src"
 DRIVER = Path(__file__).resolve().parent / "snapshot_driver.py"
+FAITHFUL_DRIVER = Path(__file__).resolve().parent / "faithful_driver.py"
 IMAGE = "alpine:3.20"
 PACKAGES = "apk add -q btrfs-progs e2fsprogs zfs python3 util-linux"
 ATTACH = (
@@ -76,20 +77,17 @@ def required(fstype: str) -> bool:
     return fstype in demanded.replace(",", " ").split()
 
 
-def stage() -> Path:
+def stage(driver: Path) -> Path:
     """Copy source and driver under /tmp, the only path the DinD daemon shares."""
     staged = Path("/tmp") / unique("baudolo-e2e-snapshot")
     shutil.copytree(REPO_SRC, staged / "src")
-    shutil.copy(DRIVER, staged / "driver.py")
+    shutil.copy(driver, staged / "driver.py")
     return staged
 
 
-def drive(fstype: str, kind: str, expect: str) -> str:
-    staged = stage()
-    script = (
-        f"set -e; {mount_script(fstype)}; "
-        f"python3 /driver.py {kind} /subject/docker {expect}"
-    )
+def drive(fstype: str, arguments: str, *, driver: Path = DRIVER) -> str:
+    staged = stage(driver)
+    script = f"set -e; {mount_script(fstype)}; python3 /driver.py {arguments}"
     try:
         proc = run(
             [
@@ -115,7 +113,7 @@ def drive(fstype: str, kind: str, expect: str) -> str:
         shutil.rmtree(staged, ignore_errors=True)
     if proc.returncode != 0:
         raise AssertionError(
-            f"{fstype}/{kind} driver failed:\n{proc.stdout}\n{proc.stderr}"
+            f"{fstype} driver failed on {arguments}:\n{proc.stdout}\n{proc.stderr}"
         )
     return proc.stdout
 
@@ -126,7 +124,7 @@ class TestE2ESnapshot(unittest.TestCase):
         require_docker()
 
     def assert_freezes(self, fstype: str) -> None:
-        output = drive(fstype, fstype, "supported")
+        output = drive(fstype, f"{fstype} /subject/docker supported")
         self.assertIn("PASS the snapshot exposes the volume", output)
         self.assertIn("PASS a later write does not reach the snapshot", output)
         self.assertIn("PASS the snapshot is removed afterwards", output)
@@ -148,13 +146,24 @@ class TestE2ESnapshot(unittest.TestCase):
         self.assert_freezes("zfs")
 
     def test_ext4_has_no_snapshot_and_says_so(self) -> None:
-        output = drive("ext4", "btrfs", "unsupported")
+        output = drive("ext4", "btrfs /subject/docker unsupported")
         self.assertIn("PASS refused loudly", output)
 
     def test_an_unknown_kind_is_refused_before_touching_the_filesystem(self) -> None:
-        output = drive("ext4", "lvm", "unsupported")
+        output = drive("ext4", "lvm /subject/docker unsupported")
         self.assertIn("PASS refused loudly", output)
 
+    def test_a_volume_with_its_own_storage_is_copied_live_not_from_the_snapshot(
+        self,
+    ) -> None:
+        output = drive("btrfs", "/subject/docker", driver=FAITHFUL_DRIVER)
+        self.assertIn(
+            "PASS the snapshot shows the other volume as an empty directory", output
+        )
+        self.assertIn("PASS the other volume degrades to live", output)
+        self.assertIn("PASS the plain volume is read from the snapshot", output)
+        self.assertIn("ALL OK", output)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/e2e/test_e2e_swarm_task_skip.py

@@ -1,5 +1,3 @@
-# tests/e2e/test_e2e_swarm_task_skip.py
-#
 # Reproduces the swarm flake fixed on this branch: baudolo used to stop a
 # swarm task container around the volume file backup because its image was
 # not whitelisted; the orchestrator immediately replaced the stopped task and

tests/e2e/test_e2e_version_gate.py

@@ -0,0 +1,283 @@
+"""A dump from a newer engine must be refused before --empty destroys anything.
+
+The pre-clean and the replay are two separate sessions with no rollback across
+them, so a dump the engine cannot parse leaves an emptied database behind. The
+decisive assertion here is not the non-zero exit - it is that the payload is
+still readable afterwards.
+"""
+
+import re
+import unittest
+from pathlib import Path
+
+from .helpers import (
+    MARIADB_DATA_DIR,
+    MARIADB_IMAGE,
+    POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
+    backup_path,
+    backup_run,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    wait_for_mariadb,
+    wait_for_mariadb_sql,
+    wait_for_postgres,
+    write_databases_csv,
+)
+
+PAYLOAD = "gate-payload"
+FUTURE = "99.0"
+
+
+def rewrite_version(dump: Path, pattern: str, version: str) -> str:
+    """Make the dump claim ``version``; return what it claimed before."""
+    text = dump.read_text(encoding="utf-8", errors="replace")
+    found = re.search(pattern, text)
+    if not found:
+        raise AssertionError(f"{dump} carries no version header matching {pattern}")
+    claimed = found.group(1)
+    dump.write_text(
+        text.replace(found.group(0), found.group(0).replace(claimed, version), 1),
+        encoding="utf-8",
+    )
+    return claimed
+
+
+class GateCase:
+    """Drive one engine through refusal, escape hatch and truthful replay."""
+
+    engine = ""
+    pattern = ""
+
+    @classmethod
+    def restore(cls, *extra: str):
+        return run(
+            [
+                "baudolo-restore",
+                cls.engine,
+                cls.volume,
+                cls.hash,
+                cls.version,
+                "--backups-dir",
+                cls.backups_dir,
+                "--repo-name",
+                cls.repo_name,
+                "--container",
+                cls.container,
+                "--db-name",
+                cls.db_name,
+                "--db-user",
+                cls.db_user,
+                "--db-password",
+                cls.db_password,
+                "--empty",
+                *extra,
+            ],
+            check=False,
+        )
+
+    @classmethod
+    def prepare(cls) -> None:
+        cls.backups_dir = f"/tmp/{cls.prefix}/Backups"
+        ensure_empty_dir(cls.backups_dir)
+        cls.compose_dir = create_minimal_compose_dir(f"/tmp/{cls.prefix}")
+        cls.repo_name = cls.prefix
+        cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
+        write_databases_csv(
+            cls.databases_csv,
+            [(cls.container, cls.db_name, cls.db_user, cls.db_password)],
+        )
+        backup_run(
+            backups_dir=cls.backups_dir,
+            repo_name=cls.repo_name,
+            compose_dir=cls.compose_dir,
+            databases_csv=cls.databases_csv,
+            database_containers=[cls.container],
+            images_no_stop_required=[cls.image],
+        )
+        cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
+        cls.dump = (
+            backup_path(cls.backups_dir, cls.repo_name, cls.version, cls.volume)
+            / "sql"
+            / f"{cls.db_name}.backup.sql"
+        )
+
+        cls.truthful_version = rewrite_version(cls.dump, cls.pattern, FUTURE)
+        cls.refused = cls.restore()
+        cls.payload_after_refusal = cls.read_payload()
+
+        cls.forced = cls.restore("--no-version-check")
+        cls.payload_after_force = cls.read_payload()
+
+        rewrite_version(cls.dump, cls.pattern, cls.truthful_version)
+        cls.replayed = cls.restore()
+        cls.payload_after_replay = cls.read_payload()
+
+    def test_the_dump_states_the_engine_it_came_from(self) -> None:
+        self.assertRegex(self.truthful_version, r"^\d+")
+
+    def test_a_newer_dump_is_refused(self) -> None:
+        self.assertNotEqual(self.refused.returncode, 0, self.refused.stdout)
+
+    def test_the_refusal_names_the_version_it_refused(self) -> None:
+        self.assertIn(FUTURE, self.refused.stderr)
+        self.assertIn("older engine", self.refused.stderr)
+
+    def test_the_refusal_left_the_data_untouched(self) -> None:
+        self.assertEqual(
+            self.payload_after_refusal,
+            PAYLOAD,
+            "--empty pre-cleaned before the version was checked",
+        )
+
+    def test_the_escape_hatch_replays_anyway(self) -> None:
+        self.assertEqual(self.forced.returncode, 0, self.forced.stderr)
+        self.assertEqual(self.payload_after_force, PAYLOAD)
+
+    def test_a_truthful_dump_replays(self) -> None:
+        self.assertEqual(self.replayed.returncode, 0, self.replayed.stderr)
+        self.assertEqual(self.payload_after_replay, PAYLOAD)
+
+
+class TestE2EPostgresVersionGate(GateCase, unittest.TestCase):
+    engine = "postgres"
+    pattern = r"-- Dumped from database version (\S+)"
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-pg-gate")
+        cls.container = f"{cls.prefix}-pg"
+        cls.volume = f"{cls.prefix}-pg-vol"
+        cls.image = POSTGRES_IMAGE
+        cls.db_name = "appdb"
+        cls.db_user = "postgres"
+        cls.db_password = "pgpw"
+
+        run(["docker", "volume", "create", cls.volume])
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.container,
+                "-e",
+                f"POSTGRES_PASSWORD={cls.db_password}",
+                "-v",
+                f"{cls.volume}:{POSTGRES_DATA_DIR}",
+                POSTGRES_IMAGE,
+            ]
+        )
+        wait_for_postgres(cls.container, user=cls.db_user)
+        cls.sql("postgres", f"CREATE DATABASE {cls.db_name}")
+        cls.sql(
+            cls.db_name,
+            f"CREATE TABLE t (v text); INSERT INTO t VALUES ('{PAYLOAD}');",
+        )
+        cls.prepare()
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=[cls.container], volumes=[cls.volume])
+
+    @classmethod
+    def sql(cls, database: str, statement: str) -> str:
+        p = run(
+            [
+                "docker",
+                "exec",
+                cls.container,
+                "sh",
+                "-lc",
+                f'psql -U {cls.db_user} -d {database} -t -A -c "{statement}"',
+            ],
+            check=False,
+        )
+        return (p.stdout or "").strip()
+
+    @classmethod
+    def read_payload(cls) -> str:
+        return cls.sql(cls.db_name, "SELECT v FROM t")
+
+
+class TestE2EMariadbVersionGate(GateCase, unittest.TestCase):
+    engine = "mariadb"
+    pattern = r"-- Server version\s+(\S+)"
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-mdb-gate")
+        cls.container = f"{cls.prefix}-mdb"
+        cls.volume = f"{cls.prefix}-mdb-vol"
+        cls.image = MARIADB_IMAGE
+        cls.db_name = "appdb"
+        cls.db_user = "test"
+        cls.db_password = "testpw"
+
+        run(["docker", "volume", "create", cls.volume])
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.container,
+                "-e",
+                "MARIADB_ROOT_PASSWORD=rootpw",
+                "-e",
+                f"MARIADB_DATABASE={cls.db_name}",
+                "-e",
+                f"MARIADB_USER={cls.db_user}",
+                "-e",
+                f"MARIADB_PASSWORD={cls.db_password}",
+                "-v",
+                f"{cls.volume}:{MARIADB_DATA_DIR}",
+                MARIADB_IMAGE,
+            ]
+        )
+        wait_for_mariadb(cls.container, root_password="rootpw", timeout_s=90)
+        wait_for_mariadb_sql(
+            cls.container, user=cls.db_user, password=cls.db_password, timeout_s=90
+        )
+        cls.sql(
+            f"CREATE TABLE {cls.db_name}.t (v VARCHAR(50)); "
+            f"INSERT INTO {cls.db_name}.t VALUES ('{PAYLOAD}');"
+        )
+        cls.prepare()
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=[cls.container], volumes=[cls.volume])
+
+    @classmethod
+    def sql(cls, statement: str) -> str:
+        p = run(
+            [
+                "docker",
+                "exec",
+                cls.container,
+                "sh",
+                "-lc",
+                (
+                    f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
+                    f'-N -B -e "{statement}"'
+                ),
+            ],
+            check=False,
+        )
+        return (p.stdout or "").strip()
+
+    @classmethod
+    def read_payload(cls) -> str:
+        return cls.sql(f"SELECT v FROM {cls.db_name}.t")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_app_databases_csv.py

@@ -9,7 +9,6 @@ import pandas as pd
 # Adjust if your package name/import path differs.
 from baudolo.backup.dumps import load_databases_df
 
-
 EXPECTED_COLUMNS = ["instance", "database", "username", "password"]
 
 
@@ -33,7 +32,6 @@ class TestLoadDatabasesDf(unittest.TestCase):
     def test_empty_csv_is_handled_with_warning_and_empty_df(self) -> None:
         with tempfile.TemporaryDirectory() as td:
             empty_path = os.path.join(td, "databases.csv")
-            # Create an empty file (0 bytes)
             with open(empty_path, "w", encoding="utf-8") as f:
                 f.write("")
 

tests/unit/backup/test_app_snapshot.py

@@ -6,7 +6,9 @@ import unittest
 from unittest import mock
 
 from baudolo.backup import app
+from baudolo.backup import snapshot as snapshot_mod
 from baudolo.backup.snapshot import volume_snapshot
+from baudolo.backup.volume import Backing
 
 
 def stubbed_snapshot(kind: str, subject: str, tag: str):
@@ -26,7 +28,7 @@ ARGV = [
 ]
 
 
-def drive(*, present: bool) -> list[dict]:
+def drive(*, present: bool = True, reason: str | None = None) -> list[dict]:
     calls: list[dict] = []
 
     def record(versions_dir, volume_name, volume_dir, *, authoritative, source):
@@ -45,8 +47,11 @@ def drive(*, present: bool) -> list[dict]:
         mock.patch.object(app, "volume_is_fully_ignored", return_value=False),
         mock.patch.object(app, "backup_dumps_for_volume", return_value=(False, False)),
         mock.patch.object(
-            app, "get_storage_path", return_value="/var/lib/docker/volumes/vol/_data/"
+            app,
+            "inspect_backing",
+            return_value=Backing("/var/lib/docker/volumes/vol/_data"),
         ),
+        mock.patch.object(snapshot_mod, "unsnapshotted", return_value=reason),
         mock.patch.object(app, "stamp_directory"),
         mock.patch.object(app, "handle_docker_compose_services"),
         mock.patch.object(app.os.path, "isdir", return_value=present),
@@ -75,6 +80,14 @@ class TestSnapshotBranch(unittest.TestCase):
         self.assertEqual(call["source"], "/var/lib/docker/volumes/vol/_data/")
         self.assertFalse(call["authoritative"])
 
+    def test_a_volume_with_its_own_backing_store_is_copied_live(self) -> None:
+        call = drive(reason="it declares its own backing store")[0]
+        self.assertEqual(call["source"], "/var/lib/docker/volumes/vol/_data/")
+        self.assertFalse(
+            call["authoritative"],
+            "the snapshot holds an empty directory for it, not its data",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/backup/test_app_volumes_no_backup_required.py

@@ -7,6 +7,7 @@ import unittest
 from unittest import mock
 
 from baudolo.backup import app
+from baudolo.backup.volume import Backing
 
 ARGV = [
     "baudolo",
@@ -47,7 +48,7 @@ def drive() -> tuple[list[str], list[str], list[str]]:
         ),
         mock.patch.object(app, "volume_is_fully_ignored", return_value=False),
         mock.patch.object(app, "backup_dumps_for_volume", return_value=(False, False)),
-        mock.patch.object(app, "get_storage_path", return_value="/data/"),
+        mock.patch.object(app, "inspect_backing", return_value=Backing("/data")),
         mock.patch.object(app, "stamp_directory"),
         mock.patch.object(app, "handle_docker_compose_services"),
         mock.patch.object(app.os.path, "isdir", return_value=True),

tests/unit/backup/test_cli.py

@@ -69,14 +69,18 @@ class TestSnapshotFlags(unittest.TestCase):
 
 class TestRequiredFlags(unittest.TestCase):
     def test_backups_dir_is_required(self) -> None:
-        with mock.patch("sys.argv", ["baudolo", "--compose-dir", "/compose"]):
-            with self.assertRaises(SystemExit):
-                parse_args()
+        with (
+            mock.patch("sys.argv", ["baudolo", "--compose-dir", "/compose"]),
+            self.assertRaises(SystemExit),
+        ):
+            parse_args()
 
     def test_compose_dir_is_required(self) -> None:
-        with mock.patch("sys.argv", ["baudolo", "--backups-dir", "/backups"]):
-            with self.assertRaises(SystemExit):
-                parse_args()
+        with (
+            mock.patch("sys.argv", ["baudolo", "--backups-dir", "/backups"]),
+            self.assertRaises(SystemExit),
+        ):
+            parse_args()
 
 
 if __name__ == "__main__":

tests/unit/backup/test_compose.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import tempfile
 import unittest
 from pathlib import Path
-from typing import List
 from unittest.mock import patch
 
 from .compose_fixture import setup_compose_dir as _setup_compose_dir
@@ -99,7 +98,7 @@ class TestCompose(unittest.TestCase):
                     str(d), ["up", "-d", "--force-recreate"]
                 )
 
-            expected: List[str] = [
+            expected: list[str] = [
                 "/usr/bin/docker",
                 "compose",
                 "--chdir",

tests/unit/backup/test_compose_hard_restart.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import unittest
-from typing import List
 from unittest.mock import patch
 
 
@@ -11,7 +10,7 @@ class HardRestartArgTests(unittest.TestCase):
     the dir is a stack whose overlay network collides with compose up, pass
     nothing."""
 
-    def _parse(self, extra: List[str]):
+    def _parse(self, extra: list[str]):
         import sys
 
         from baudolo.backup import cli

tests/unit/backup/test_db_mariadb_dump.py

@@ -20,15 +20,17 @@ def _capture_commands(*, db_type, rows, container):
         captured.append(cmd)
         return []
 
-    with tempfile.TemporaryDirectory() as td:
-        with patch.object(db_mod, "execute_shell_command", side_effect=_capture):
-            db_mod.backup_database(
-                container=container,
-                volume_dir=td,
-                db_type=db_type,
-                databases_df=_df(rows),
-                database_containers=[container],
-            )
+    with (
+        tempfile.TemporaryDirectory() as td,
+        patch.object(db_mod, "execute_shell_command", side_effect=_capture),
+    ):
+        db_mod.backup_database(
+            container=container,
+            volume_dir=td,
+            db_type=db_type,
+            databases_df=_df(rows),
+            database_containers=[container],
+        )
     return captured
 
 

tests/unit/backup/test_snapshot.py

@@ -61,9 +61,11 @@ class TestBtrfs(unittest.TestCase):
 
     def test_it_removes_the_snapshot_even_when_the_body_raises(self) -> None:
         run = Runner()
-        with self.assertRaises(ZeroDivisionError):
-            with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run):
-                raise ZeroDivisionError
+        with (
+            self.assertRaises(ZeroDivisionError),
+            volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run),
+        ):
+            raise ZeroDivisionError
         self.assertTrue(run.calls[-1].startswith("btrfs subvolume delete"))
 
 
@@ -93,26 +95,30 @@ class TestZfs(unittest.TestCase):
 
     def test_an_unmounted_dataset_is_an_error(self) -> None:
         run = Runner({"zfs list": [""]})
-        with self.assertRaises(SnapshotError):
-            with volume_snapshot("zfs", "/var/lib/docker", "20260731", run=run):
-                pass
+        with (
+            self.assertRaises(SnapshotError),
+            volume_snapshot("zfs", "/var/lib/docker", "20260731", run=run),
+        ):
+            pass
 
 
 class TestRejections(unittest.TestCase):
     def test_an_unknown_kind_is_rejected(self) -> None:
         run = Runner()
-        with self.assertRaises(SnapshotError):
-            with volume_snapshot("ext4", "/var/lib/docker", "20260731", run=run):
-                pass
+        with (
+            self.assertRaises(SnapshotError),
+            volume_snapshot("ext4", "/var/lib/docker", "20260731", run=run),
+        ):
+            pass
         self.assertEqual(run.calls, [])
 
     def test_a_path_outside_the_subject_is_rejected(self) -> None:
         run = Runner()
-        with volume_snapshot(
-            "btrfs", "/var/lib/docker", "20260731", run=run
-        ) as resolve:
-            with self.assertRaises(SnapshotError):
-                resolve("/etc/passwd")
+        with (
+            volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run) as resolve,
+            self.assertRaises(SnapshotError),
+        ):
+            resolve("/etc/passwd")
 
     def test_the_subject_itself_resolves_to_the_snapshot_root(self) -> None:
         run = Runner()
@@ -137,9 +143,11 @@ class TestRemovalFailure(unittest.TestCase):
             pass
 
     def test_a_failed_removal_does_not_mask_the_body(self) -> None:
-        with self.assertRaises(ZeroDivisionError):
-            with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=Busy()):
-                raise ZeroDivisionError
+        with (
+            self.assertRaises(ZeroDivisionError),
+            volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=Busy()),
+        ):
+            raise ZeroDivisionError
 
 
 if __name__ == "__main__":

tests/unit/backup/test_snapshot_faithfulness.py

@@ -0,0 +1,126 @@
+"""Which volumes a snapshot of the subject actually contains.
+
+The failure this guards against is silent: a volume with a backing store of
+its own is present inside the snapshot as an empty directory, so rsync
+succeeds, the generation is stamped complete, and the volume is empty in it.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+from baudolo.backup.snapshot import SnapshotError, snapshot_source, unsnapshotted
+from baudolo.backup.volume import Backing
+
+
+class TestUnsnapshotted(unittest.TestCase):
+    def setUp(self) -> None:
+        self.subject = tempfile.mkdtemp()
+        self.mountpoint = os.path.join(self.subject, "volumes", "app", "_data")
+        os.makedirs(self.mountpoint)
+
+    def backing(self, **kwargs) -> Backing:
+        return Backing(kwargs.pop("mountpoint", self.mountpoint), **kwargs)
+
+    def test_a_plain_local_volume_is_captured(self) -> None:
+        self.assertIsNone(unsnapshotted(self.backing(), self.subject))
+
+    def test_a_foreign_driver_is_not(self) -> None:
+        reason = unsnapshotted(self.backing(driver="rexray"), self.subject)
+        self.assertIn("rexray", reason)
+
+    def test_declared_driver_options_are_not(self) -> None:
+        reason = unsnapshotted(
+            self.backing(options={"type": "nfs", "device": ":/exports/app"}),
+            self.subject,
+        )
+        self.assertIn("backing store", reason)
+
+    def test_the_declaration_decides_not_the_mount_table(self) -> None:
+        """Docker unmounts an NFS volume when its last container stops."""
+        with mock.patch.object(os.path, "ismount", return_value=False):
+            reason = unsnapshotted(self.backing(options={"type": "nfs"}), self.subject)
+        self.assertIsNotNone(reason)
+
+    def test_a_volume_without_a_mountpoint_is_not(self) -> None:
+        reason = unsnapshotted(Backing(""), self.subject)
+        self.assertIn("no mountpoint", reason)
+
+    def test_an_own_mount_is_not(self) -> None:
+        with mock.patch.object(os.path, "ismount", return_value=True):
+            reason = unsnapshotted(self.backing(), self.subject)
+        self.assertIn("own mount", reason)
+
+    def test_a_filesystem_boundary_is_not(self) -> None:
+        real = os.stat
+
+        def crossing(path, *args, **kwargs):
+            info = real(path, *args, **kwargs)
+            if os.path.realpath(path) == os.path.realpath(self.mountpoint):
+                return os.stat_result(
+                    (info.st_mode, info.st_ino, info.st_dev + 1, *tuple(info)[3:])
+                )
+            return info
+
+        with mock.patch.object(os, "stat", side_effect=crossing):
+            reason = unsnapshotted(self.backing(), self.subject)
+        self.assertIn("filesystem boundary", reason)
+
+    def test_an_unreadable_mountpoint_is_not(self) -> None:
+        reason = unsnapshotted(
+            self.backing(mountpoint=os.path.join(self.subject, "gone")), self.subject
+        )
+        self.assertIn("could not be read", reason)
+
+
+class TestSnapshotSource(unittest.TestCase):
+    def setUp(self) -> None:
+        self.subject = tempfile.mkdtemp()
+        self.mountpoint = os.path.join(self.subject, "volumes", "app", "_data")
+        os.makedirs(self.mountpoint)
+        self.snapshot = os.path.join(
+            self.subject, ".baudolo-tag", "volumes", "app", "_data"
+        )
+        os.makedirs(self.snapshot)
+        self.backing = Backing(self.mountpoint)
+
+    def test_a_captured_volume_reads_from_the_snapshot(self) -> None:
+        source, reason = snapshot_source(
+            lambda path: self.snapshot + "/", self.backing, self.subject
+        )
+        self.assertEqual(source, self.snapshot + "/")
+        self.assertEqual(reason, "")
+
+    def test_an_uncaptured_volume_is_refused_before_the_resolver_runs(self) -> None:
+        def resolve(path):
+            raise AssertionError("must not resolve a volume the snapshot misses")
+
+        source, reason = snapshot_source(
+            resolve, Backing(self.mountpoint, options={"type": "nfs"}), self.subject
+        )
+        self.assertIsNone(source)
+        self.assertIn("backing store", reason)
+
+    def test_a_volume_outside_the_subject_degrades_instead_of_raising(self) -> None:
+        def resolve(path):
+            raise SnapshotError(f"{path} lies outside the snapshot subject")
+
+        source, reason = snapshot_source(resolve, self.backing, self.subject)
+        self.assertIsNone(source)
+        self.assertIn("lies outside", reason)
+
+    def test_a_volume_created_after_the_snapshot_degrades(self) -> None:
+        source, reason = snapshot_source(
+            lambda path: os.path.join(self.subject, "absent") + "/",
+            self.backing,
+            self.subject,
+        )
+        self.assertIsNone(source)
+        self.assertIn("created after", reason)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_cli_version_flag.py

@@ -0,0 +1,52 @@
+import unittest
+from unittest.mock import patch
+
+from baudolo.restore import __main__ as cli
+
+ENGINES = {
+    "postgres": ("restore_postgres_sql", ["--db-name", "app"]),
+    "mariadb": ("restore_mariadb_sql", ["--db-name", "app"]),
+    "cluster": ("restore_cluster_sql", ["--instance", "central", "--db-user", "root"]),
+}
+
+
+class TestVersionFlagReachesEveryEngine(unittest.TestCase):
+    def call(self, engine: str, extra: list) -> dict:
+        target, required = ENGINES[engine]
+        argv = [
+            engine,
+            "app_vol",
+            "hash",
+            "20260817000000",
+            "--container",
+            "db",
+            "--db-password",
+            "pw",
+            *required,
+            *extra,
+        ]
+        with patch.object(cli, target) as restore:
+            self.assertEqual(cli.main(argv), 0)
+        return restore.call_args.kwargs
+
+    def test_the_gate_is_on_by_default(self) -> None:
+        for engine in ENGINES:
+            with self.subTest(engine=engine):
+                self.assertTrue(self.call(engine, [])["check_version"])
+
+    def test_the_flag_turns_it_off(self) -> None:
+        for engine in ENGINES:
+            with self.subTest(engine=engine):
+                kwargs = self.call(engine, ["--no-version-check"])
+                self.assertFalse(kwargs["check_version"])
+
+    def test_empty_stays_independent_of_the_gate(self) -> None:
+        for engine in ENGINES:
+            with self.subTest(engine=engine):
+                kwargs = self.call(engine, ["--empty"])
+                self.assertTrue(kwargs["empty"])
+                self.assertTrue(kwargs["check_version"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_cluster_preclean_scope.py

@@ -0,0 +1,126 @@
+"""The pre-clean may only drop what the dump can bring back.
+
+A shared instance carries databases and roles from other applications, and a
+database created after the backup is in no dump at all. Dropping those would
+destroy data this restore cannot restore.
+"""
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from baudolo.restore.db import cluster as cluster_mod
+
+DUMP = """--
+-- PostgreSQL database cluster dump
+--
+
+CREATE ROLE app;
+ALTER ROLE app WITH LOGIN;
+CREATE ROLE reporting;
+
+CREATE DATABASE appdb OWNER app;
+
+\\connect appdb
+
+CREATE TABLE t (v text);
+
+\\connect template1
+"""
+
+
+def dump_file(text: str) -> str:
+    path = Path(tempfile.mkdtemp()) / "central.cluster.backup.sql"
+    path.write_text(text, encoding="utf-8")
+    return str(path)
+
+
+class TestDumpInventory(unittest.TestCase):
+    def test_it_reads_databases_and_roles_the_dump_recreates(self) -> None:
+        databases, roles = cluster_mod.dump_inventory(dump_file(DUMP))
+        self.assertEqual(databases, ["appdb", "template1"])
+        self.assertEqual(roles, ["app", "reporting"])
+
+    def test_a_quoted_name_keeps_its_spaces(self) -> None:
+        databases, roles = cluster_mod.dump_inventory(
+            dump_file('\\connect "odd name"\nCREATE ROLE "odd role";\n')
+        )
+        self.assertEqual(databases, ["odd name"])
+        self.assertEqual(roles, ["odd role"])
+
+    def test_psql_options_are_not_mistaken_for_the_database(self) -> None:
+        databases, _roles = cluster_mod.dump_inventory(
+            dump_file("\\connect -reuse-previous=on dbname=appdb\n")
+        )
+        self.assertEqual(databases, ["appdb"])
+
+    def test_create_database_options_are_not_part_of_the_name(self) -> None:
+        databases, _roles = cluster_mod.dump_inventory(
+            dump_file("CREATE DATABASE appdb WITH TEMPLATE = template0 OWNER = app;\n")
+        )
+        self.assertEqual(databases, ["appdb"])
+
+    def test_a_name_is_listed_once(self) -> None:
+        databases, _roles = cluster_mod.dump_inventory(
+            dump_file("\\connect a\n\\connect a\n")
+        )
+        self.assertEqual(databases, ["a"])
+
+
+class TestInstanceRefusal(unittest.TestCase):
+    """--empty wipes the whole instance, so it may only run on one this dump
+    can rebuild. Scoping the sweep instead wedges the restore: a surviving
+    database that grants to a dumped role pins it, DROP ROLE fails, and the
+    pre-clean aborts after the dump's own databases are already gone."""
+
+    def check(self, present: str, dump: str = DUMP):
+        from unittest import mock
+
+        with mock.patch.object(
+            cluster_mod, "instance_databases", return_value=present.split()
+        ):
+            cluster_mod.assert_instance_matches_dump(
+                "db", "postgres", dump_file(dump), {}
+            )
+
+    def test_an_instance_the_dump_covers_passes(self) -> None:
+        self.check("appdb")
+
+    def test_an_empty_instance_passes(self) -> None:
+        self.check("")
+
+    def test_a_database_the_dump_lacks_is_refused(self) -> None:
+        with self.assertRaises(RuntimeError) as raised:
+            self.check("appdb sibling")
+        self.assertIn("sibling", str(raised.exception))
+
+    def test_the_refusal_names_every_foreign_database(self) -> None:
+        with self.assertRaises(RuntimeError) as raised:
+            self.check("one two")
+        message = str(raised.exception)
+        self.assertIn("one", message)
+        self.assertIn("two", message)
+
+    def test_the_refusal_happens_before_anything_is_dropped(self) -> None:
+        from unittest import mock
+
+        with (
+            mock.patch.object(
+                cluster_mod, "instance_databases", return_value=["foreign"]
+            ),
+            mock.patch.object(cluster_mod, "docker_exec") as touched,
+            self.assertRaises(RuntimeError),
+        ):
+            cluster_mod.restore_cluster_sql(
+                container="db",
+                user="postgres",
+                password="pw",
+                sql_path=dump_file(DUMP),
+                empty=True,
+                check_version=False,
+            )
+        touched.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_cluster_replay.py

@@ -0,0 +1,125 @@
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from baudolo.restore.db import cluster as cluster_mod
+from baudolo.restore.paths import BackupPaths
+
+
+class TestClusterReplay(unittest.TestCase):
+    def _replay(self, *, empty: bool):
+        calls = []
+
+        def _capture(container, argv, **kwargs):
+            if "-tAc" in argv:
+                return MagicMock(stdout=b"")
+            calls.append((argv, kwargs.get("stdin")))
+            return MagicMock()
+
+        with tempfile.NamedTemporaryFile(suffix=".sql") as sql:
+            sql.write(b"CREATE ROLE app;\nCREATE DATABASE app OWNER app;\n")
+            sql.flush()
+            with patch.object(cluster_mod, "docker_exec", side_effect=_capture):
+                cluster_mod.restore_cluster_sql(
+                    container="db",
+                    user="postgres",
+                    password="pw",
+                    sql_path=sql.name,
+                    empty=empty,
+                    check_version=False,
+                )
+        return calls
+
+    def test_the_replay_is_not_wrapped_in_a_transaction(self) -> None:
+        argv, _ = self._replay(empty=False)[0]
+        self.assertNotIn(
+            "--single-transaction",
+            argv,
+            "CREATE DATABASE cannot run inside a transaction block, so unlike the "
+            "single-database replay this stream must not be wrapped in one",
+        )
+        self.assertIn("ON_ERROR_STOP=1", argv)
+
+    def test_the_replay_targets_the_control_database(self) -> None:
+        argv, _ = self._replay(empty=False)[0]
+        self.assertEqual(argv[argv.index("-d") + 1], cluster_mod.CONTROL_DB)
+        self.assertEqual(argv[argv.index("-U") + 1], "postgres")
+
+    def test_without_empty_nothing_is_dropped_first(self) -> None:
+        self.assertEqual(len(self._replay(empty=False)), 1)
+
+    def test_empty_drops_databases_before_their_owners(self) -> None:
+        calls = self._replay(empty=True)
+        self.assertEqual(len(calls), 2, f"expected pre-clean + replay: {calls}")
+        preclean = calls[0][1].decode()
+        self.assertLess(
+            preclean.index("DROP DATABASE"),
+            preclean.index("DROP ROLE"),
+            "a role cannot be dropped while it still owns a database",
+        )
+        self.assertIn("DROP OWNED BY", preclean)
+        self.assertIn("ORDER BY phase", preclean)
+
+    def test_the_preclean_spares_what_no_dump_recreates(self) -> None:
+        preclean = self._replay(empty=True)[0][1].decode()
+        self.assertIn("NOT datistemplate", preclean)
+        self.assertIn("datname <> current_database()", preclean)
+        self.assertIn("starts_with(rolname, 'pg_')", preclean)
+        self.assertIn("rolname <> current_user", preclean)
+
+    def test_only_the_connecting_role_loses_its_create(self) -> None:
+        # Captured from pg_dumpall 17.
+        dump = [
+            b"CREATE ROLE app;\n",
+            b"ALTER ROLE app WITH NOSUPERUSER INHERIT LOGIN PASSWORD 'SCRAM-SHA-256$...';\n",
+            b"CREATE ROLE postgres;\n",
+            b"ALTER ROLE postgres WITH SUPERUSER INHERIT LOGIN PASSWORD 'SCRAM-SHA-256$...';\n",
+            b'CREATE ROLE "odd-name";\n',
+        ]
+        kept = list(cluster_mod.filter_own_role_creation(dump, "postgres"))
+        self.assertNotIn(b"CREATE ROLE postgres;\n", kept)
+        self.assertIn(b"CREATE ROLE app;\n", kept)
+        self.assertIn(b'CREATE ROLE "odd-name";\n', kept)
+        self.assertEqual(
+            sum(1 for line in kept if line.startswith(b"ALTER ROLE postgres")),
+            1,
+            "the ALTER re-applies the superuser's attributes and password",
+        )
+
+    def test_a_quoted_connecting_role_is_matched_too(self) -> None:
+        kept = list(
+            cluster_mod.filter_own_role_creation(
+                [b'CREATE ROLE "odd-name";\n'], "odd-name"
+            )
+        )
+        self.assertEqual(kept, [])
+
+    def test_a_role_whose_name_merely_starts_the_same_is_kept(self) -> None:
+        kept = list(
+            cluster_mod.filter_own_role_creation(
+                [b"CREATE ROLE postgresql;\n"], "postgres"
+            )
+        )
+        self.assertEqual(kept, [b"CREATE ROLE postgresql;\n"])
+
+    def test_a_missing_dump_is_reported_as_such(self) -> None:
+        with self.assertRaises(FileNotFoundError):
+            cluster_mod.restore_cluster_sql(
+                container="db",
+                user="postgres",
+                password="pw",
+                sql_path="/nonexistent/x.cluster.backup.sql",
+                empty=False,
+                check_version=False,
+            )
+
+    def test_the_path_helper_names_the_dumpall_file(self) -> None:
+        paths = BackupPaths("vol", "hash", "v1", repo_name="repo", backups_dir="/B")
+        self.assertEqual(
+            paths.cluster_file("bigbluebutton"),
+            "/B/hash/repo/v1/vol/sql/bigbluebutton.cluster.backup.sql",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_files_backing_store.py

@@ -0,0 +1,62 @@
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from baudolo.restore import files as files_mod
+
+
+class TestBackingStoreGuard(unittest.TestCase):
+    def restore(self, inspect: str, mounted: bool) -> tuple[int, list]:
+        calls = []
+
+        def _run(cmd, **kwargs):
+            calls.append(cmd)
+            return MagicMock(stdout=inspect.encode())
+
+        with (
+            patch.object(files_mod, "docker_volume_exists", return_value=True),
+            patch.object(files_mod.os.path, "ismount", return_value=mounted),
+            patch.object(files_mod, "run", side_effect=_run),
+        ):
+            code = files_mod.restore_volume_files("app_data", tempfile.mkdtemp())
+        return code, calls
+
+    def rsynced(self, calls: list) -> bool:
+        return any(cmd[0] == "rsync" for cmd in calls)
+
+    def test_plain_local_volume_is_restored_unmounted(self) -> None:
+        code, calls = self.restore("/var/lib/docker/volumes/a/_data|local|plain", False)
+        self.assertEqual(code, 0)
+        self.assertTrue(self.rsynced(calls))
+
+    def test_volume_with_driver_options_is_refused_while_unmounted(self) -> None:
+        code, calls = self.restore("/var/lib/docker/volumes/a/_data|local|opts", False)
+        self.assertEqual(code, 2)
+        self.assertFalse(
+            self.rsynced(calls),
+            "an NFS or bind volume writes under the mount and reports success",
+        )
+
+    def test_volume_with_driver_options_is_restored_once_mounted(self) -> None:
+        code, calls = self.restore("/var/lib/docker/volumes/a/_data|local|opts", True)
+        self.assertEqual(code, 0)
+        self.assertTrue(self.rsynced(calls))
+
+    def test_foreign_driver_is_refused_while_unmounted(self) -> None:
+        code, calls = self.restore("/mnt/gluster/a|glusterfs|plain", False)
+        self.assertEqual(code, 2)
+        self.assertFalse(self.rsynced(calls))
+
+    def test_an_unresolvable_mountpoint_still_fails_first(self) -> None:
+        code, calls = self.restore("|local|plain", False)
+        self.assertEqual(code, 2)
+        self.assertFalse(self.rsynced(calls))
+
+    def test_a_format_without_the_new_fields_is_treated_as_plain(self) -> None:
+        code, calls = self.restore("/var/lib/docker/volumes/a/_data", False)
+        self.assertEqual(code, 0)
+        self.assertTrue(self.rsynced(calls))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_mariadb_empty_drop.py

@@ -29,6 +29,7 @@ class TestMariadbEmptyDrop(unittest.TestCase):
                     password="pw",
                     sql_path=sql.name,
                     empty=True,
+                    check_version=False,
                 )
 
         drop_calls = [argv for argv in calls if any("DROP TABLE" in a for a in argv)]

tests/unit/restore/test_postgres_single_transaction.py

@@ -24,6 +24,7 @@ class TestPostgresSingleTransaction(unittest.TestCase):
                     password="pw",
                     sql_path=sql.name,
                     empty=True,
+                    check_version=False,
                 )
 
         self.assertEqual(len(calls), 2, f"expected pre-clean + replay: {calls}")

tests/unit/restore/test_version_gate.py

@@ -0,0 +1,244 @@
+import os
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from baudolo.restore.db import cluster as cluster_mod
+from baudolo.restore.db import mariadb as mdb_mod
+from baudolo.restore.db import postgres as pg_mod
+from baudolo.restore.db import version as ver
+
+POSTGRES_HEADER = """--
+-- PostgreSQL database dump
+--
+
+\\restrict BbyzwODc1rWKL3rDyLhEjgCF0Kf2TU5ma7gcTs8eQI7copLtydXkc61zdULsPav
+
+-- Dumped from database version 17.11
+-- Dumped by pg_dump version 17.11
+
+SET statement_timeout = 0;
+"""
+
+MARIADB_HEADER = """/*M!999999\\- enable the sandbox mode */
+-- MariaDB dump 10.19-11.8.8-MariaDB, for debian-linux-gnu (x86_64)
+--
+-- Host: 127.0.0.1    Database: mysql
+-- ------------------------------------------------------
+-- Server version\t11.8.8-MariaDB-ubu2404
+
+/*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
+"""
+
+
+def cluster_header(roles: int) -> str:
+    """A pg_dumpall stream: banner, N roles, then the first database's dump."""
+    head = [
+        "--",
+        "-- PostgreSQL database cluster dump",
+        "--",
+        "",
+        "SET default_transaction_read_only = off;",
+        "",
+        "--",
+        "-- Roles",
+        "--",
+    ]
+    for i in range(roles):
+        head.append(f'CREATE ROLE "app{i}";')
+        head.append(f'ALTER ROLE "app{i}" WITH NOSUPERUSER INHERIT LOGIN;')
+    head.append("\\connect app")
+    head.append("")
+    return "\n".join(head) + "\n" + POSTGRES_HEADER
+
+
+def dump_file(text: str) -> str:
+    path = os.path.join(tempfile.mkdtemp(), "app.backup.sql")
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(text)
+    return path
+
+
+class TestDumpVersion(unittest.TestCase):
+    """Headers captured from postgres:17-alpine and mariadb:11 themselves."""
+
+    def test_postgres_reads_the_source_server(self) -> None:
+        self.assertEqual(
+            ver.dump_version(dump_file(POSTGRES_HEADER), "postgres"), "17.11"
+        )
+
+    def test_mariadb_reads_the_server_not_the_dump_tool(self) -> None:
+        found = ver.dump_version(dump_file(MARIADB_HEADER), "mariadb")
+        self.assertEqual(found, "11.8.8-MariaDB-ubu2404")
+        self.assertNotEqual(
+            ver.major_of(found),
+            10,
+            "10.19 is mariadb-dump's own version, not the server's",
+        )
+
+    def test_cluster_dump_states_its_version_far_below_the_header(self) -> None:
+        path = dump_file(cluster_header(roles=200))
+        with open(path, encoding="utf-8") as handle:
+            offset = next(i for i, line in enumerate(handle) if "Dumped from" in line)
+        self.assertGreater(offset, 100, "fixture must exercise the deep scan")
+        self.assertEqual(ver.dump_version(path, "postgres"), "17.11")
+
+    def test_a_version_beyond_the_scan_limit_is_refused_not_ignored(self) -> None:
+        path = dump_file(cluster_header(roles=ver.SCAN_LINES))
+        with self.assertRaises(ver.VersionMismatch):
+            ver.dump_version(path, "postgres")
+
+    def test_a_dump_without_a_version_header_is_refused(self) -> None:
+        path = dump_file("CREATE TABLE t (id int);\n")
+        with self.assertRaises(ver.VersionMismatch):
+            ver.dump_version(path, "postgres")
+
+
+class TestMajorOf(unittest.TestCase):
+    def test_reads_the_leading_number(self) -> None:
+        self.assertEqual(ver.major_of("17.11"), 17)
+        self.assertEqual(ver.major_of("11.8.8-MariaDB-ubu2404"), 11)
+        self.assertEqual(ver.major_of("9.6.24"), 9)
+        self.assertEqual(ver.major_of("18beta1"), 18)
+
+    def test_refuses_an_unreadable_version(self) -> None:
+        with self.assertRaises(ver.VersionMismatch):
+            ver.major_of("unknown")
+
+
+class TestAssertReplayable(unittest.TestCase):
+    def test_newer_dump_into_older_engine_is_refused(self) -> None:
+        with self.assertRaises(ver.VersionMismatch) as caught:
+            ver.assert_replayable("/b/app.sql", "postgres", "17.11", "15.6")
+        self.assertIn("17.11", str(caught.exception))
+        self.assertIn("15.6", str(caught.exception))
+
+    def test_same_major_passes(self) -> None:
+        ver.assert_replayable("/b/app.sql", "postgres", "17.4", "17.11")
+
+    def test_older_dump_into_newer_engine_passes(self) -> None:
+        ver.assert_replayable("/b/app.sql", "postgres", "15.6", "17.11")
+
+
+class TestServerVersion(unittest.TestCase):
+    def test_postgres_asks_over_pgpassword(self) -> None:
+        with patch.object(ver, "docker_exec") as exec_:
+            exec_.return_value = MagicMock(stdout=b" 17.11 \n")
+            found = ver.server_version("db", "postgres", "app", "pw")
+        self.assertEqual(found, "17.11")
+        argv = exec_.call_args.args[1]
+        self.assertIn("SHOW server_version", argv)
+        self.assertEqual(exec_.call_args.kwargs["docker_env"], {"PGPASSWORD": "pw"})
+
+    def test_mariadb_asks_through_the_detected_client(self) -> None:
+        with patch.object(ver, "docker_exec") as exec_:
+            exec_.return_value = MagicMock(stdout=b"11.8.8-MariaDB-ubu2404\n")
+            found = ver.server_version("db", "mariadb", "app", "pw", client="mysql")
+        self.assertEqual(found, "11.8.8-MariaDB-ubu2404")
+        self.assertEqual(exec_.call_args.args[1][0], "mysql")
+
+
+class TestGateStopsBeforeDestroying(unittest.TestCase):
+    """--empty drops in one session and replays in the next, with no rollback
+    between them, so the refusal has to land before the first session."""
+
+    def setUp(self) -> None:
+        self.serving = patch.object(ver, "docker_exec").start()
+        self.addCleanup(patch.stopall)
+
+    def serve(self, version: str) -> None:
+        self.serving.return_value = MagicMock(stdout=version.encode())
+
+    def test_postgres_refuses_without_running_the_preclean(self) -> None:
+        self.serve("15.6")
+        path = dump_file(POSTGRES_HEADER)
+        with (
+            patch.object(pg_mod, "docker_exec") as replay,
+            self.assertRaises(ver.VersionMismatch),
+        ):
+            pg_mod.restore_postgres_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=path,
+                empty=True,
+            )
+        replay.assert_not_called()
+
+    def test_cluster_refuses_without_running_the_preclean(self) -> None:
+        self.serve("15.6")
+        path = dump_file(cluster_header(roles=3))
+        with (
+            patch.object(cluster_mod, "docker_exec") as replay,
+            self.assertRaises(ver.VersionMismatch),
+        ):
+            cluster_mod.restore_cluster_sql(
+                container="db",
+                user="postgres",
+                password="pw",
+                sql_path=path,
+                empty=True,
+            )
+        replay.assert_not_called()
+
+    def test_mariadb_refuses_without_dropping_tables(self) -> None:
+        self.serve("10.11.6-MariaDB")
+        path = dump_file(MARIADB_HEADER)
+        with (
+            patch.object(mdb_mod, "_pick_client", return_value="mariadb"),
+            patch.object(mdb_mod, "docker_exec") as replay,
+            self.assertRaises(ver.VersionMismatch),
+        ):
+            mdb_mod.restore_mariadb_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=path,
+                empty=True,
+            )
+        replay.assert_not_called()
+
+    def test_matching_versions_let_the_replay_through(self) -> None:
+        self.serve("17.11")
+        path = dump_file(POSTGRES_HEADER)
+        with patch.object(pg_mod, "docker_exec") as replay:
+            pg_mod.restore_postgres_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=path,
+                empty=False,
+            )
+        replay.assert_called_once()
+
+    def test_the_escape_hatch_asks_the_engine_nothing(self) -> None:
+        path = dump_file("CREATE TABLE t (id int);\n")
+        with patch.object(pg_mod, "docker_exec"):
+            pg_mod.restore_postgres_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=path,
+                empty=False,
+                check_version=False,
+            )
+        self.serving.assert_not_called()
+
+    def test_a_missing_dump_is_reported_as_missing_not_as_a_mismatch(self) -> None:
+        with self.assertRaises(FileNotFoundError):
+            pg_mod.restore_postgres_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=os.path.join(tempfile.mkdtemp(), "absent.sql"),
+                empty=True,
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/seed/test_main.py

@@ -131,7 +131,6 @@ class TestSeedMain(unittest.TestCase):
             warning_calls,
             "Expected a WARNING print when databases.csv is empty, but none was found.",
         )
-        # Ensure the warning goes to stderr
         _, warn_kwargs = warning_calls[0]
         self.assertEqual(warn_kwargs.get("file"), seed_main.sys.stderr)