Release version 4.0.0

The backup built command strings and handed them to shell=True, so four interpolated values per dump - user, password, container, database - were each a way out of the command. validate_database covered one of them since the previous commit; now there is nothing to cover: every command is an argv list, and a value can only ever be an argument.

execute_to_file absorbs the atomic dump write. The shell redirect into <file>.tmp and the separate mv process become a Python file handle and os.replace, and a failing dump deletes its partial file instead of leaving it. PGPASSWORD moves out of the command string into the child's environment, where a process listing does not show it.

docker exec is built in one place, docker_exec_argv; db.py's three hand-built copies and the probe use it. The dead docker_volume_exists goes - never called, and the restore side owns the living twin. The rsync quoting in --link-dest falls away: inside an argv it would have become part of the path.

The snapshot module's injected runner changes type with it, which the three e2e drivers implement - the first conversion missed them, btrfs ran with no arguments, and the e2e caught it. Marked breaking for that contract: any external runner injected into volume_snapshot must now accept a list.

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

CHANGELOG.md

@@ -1,5 +1,37 @@
 # Changelog
 
+## [4.0.0] - 2026-08-17
+
+Breaking:
+- CLI: *--repo-name* and *--databases-csv* are required. One default was the
+  literal *backup-docker-to-local* while its help promised the git folder name;
+  the other pointed into the installed package, so a forgotten flag ran the
+  whole backup silently without a single dump.
+- CLI: *--dump-only-sql* is now *--only-sql*; the old spelling exits 2.
+- CLI: *--everything* is withdrawn. Its only real effect was to ignore
+  *--images-no-stop-required*, which leaving that list empty already does.
+- Library: the runner injected into *volume_snapshot* receives an argv list
+  instead of a command string.
+
+New:
+- Backup: *--only-files* — no dumps at all, every volume as files, for hosts
+  that hold no database credentials. Needs no *--databases-csv*; mutually
+  exclusive with *--only-sql*.
+- Backup: the engine is detected by executing the dump tool in the container
+  (*pg_dumpall*, *mariadb-dump*, *mysqldump*), not by reading the image name.
+  A dedicated Postgres tagged *<app>-database* is finally dumped; an image
+  merely named like an engine no longer kills the run with exit 127. Probed
+  once per image ID, and an image shipping only *mysqldump* is dumped with it.
+- Library: *baudolo.databases* states the databases.csv contract once —
+  columns, delimiter, cluster marker, validator, *read_rows()* — for the seed,
+  the backup, and external consumers.
+
+Changed:
+- Backup: every command is an argv list; *shell=True* is gone. A database name
+  is validated on read as strictly as the seed writes it, *PGPASSWORD* travels
+  in the child's environment instead of the command string, and a failing dump
+  deletes its partial file instead of leaving it behind.
+
 ## [3.6.1] - 2026-08-17
 
 - Restore: *--empty* on a cluster dump is a catalog-wide sweep — it drops every

README.md

@@ -123,6 +123,8 @@ This information is used by `baudolo` to execute
 ```bash
 baudolo \
   --compose-dir /srv/docker \
+  --backups-dir /Backups \
+  --repo-name my-repo \
   --databases-csv /etc/baudolo/databases.csv \
   --database-containers central-postgres central-mariadb \
   --images-no-stop-required alpine postgres mariadb mysql \
@@ -133,11 +135,12 @@ baudolo \
 
 | Flag            | Description                                 |
 | --------------- | ------------------------------------------- |
-| `--everything`  | Always stop containers and re-run rsync     |
-| `--dump-only-sql`| Skip file backups only for DB volumes when dumps succeed; non-DB volumes are still backed up; fallback to files if no dump.    |
+| `--only-sql`    | Skip file backups only for DB volumes when dumps succeed; non-DB volumes are still backed up; fallback to files if no dump.    |
+| `--only-files`  | Take no dumps at all; every volume is backed up as files. Needs no `--databases-csv`. Mutually exclusive with `--only-sql`. |
 | `--shutdown`    | Do not restart containers after backup      |
-| `--backups-dir` | Backup root directory (default: `/Backups`) |
-| `--repo-name`   | Backup namespace under machine hash         |
+| `--backups-dir` | Backup root directory (required)            |
+| `--repo-name`   | Backup namespace under machine hash (required) |
+| `--databases-csv`| Path to `databases.csv` (required)         |
 
 ## ♻️ Restore Operations
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "backup-docker-to-local"
-version = "3.6.1"
+version = "4.0.0"
 description = "Backup Docker volumes to local with rsync and optional DB dumps."
 readme = "README.md"
 requires-python = ">=3.9"

src/baudolo/backup/app.py

@@ -37,7 +37,7 @@ def main() -> int:
     versions_dir = os.path.join(args.backups_dir, machine_id, args.repo_name)
     version_dir = create_version_directory(versions_dir, backup_time)
 
-    databases_df = load_databases_df(args.databases_csv)
+    databases_df = None if args.only_files else load_databases_df(args.databases_csv)
 
     print("💾 Start volume backups...", flush=True)
 
@@ -69,17 +69,19 @@ def main() -> int:
 
             vol_dir = create_volume_directory(version_dir, volume_name)
 
-            found_db, dumped_any = backup_dumps_for_volume(
-                containers=containers,
-                vol_dir=vol_dir,
-                databases_df=databases_df,
-                database_containers=args.database_containers,
-            )
+            found_db = dumped_any = False
+            if not args.only_files:
+                found_db, dumped_any = backup_dumps_for_volume(
+                    containers=containers,
+                    vol_dir=vol_dir,
+                    databases_df=databases_df,
+                    database_containers=args.database_containers,
+                )
 
-            if args.dump_only_sql and found_db:
+            if args.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}'. "
+                        f"WARNING: only-sql requested but no DB dump was produced for DB volume '{volume_name}'. "
                         "Falling back to file backup.",
                         flush=True,
                     )
@@ -119,15 +121,6 @@ def main() -> int:
                     copy(authoritative=False)
                 continue
 
-            if args.everything:
-                stoppable = filter_stoppable(containers)
-                copy(authoritative=False)
-                change_containers_status(stoppable, "stop")
-                copy(authoritative=True)
-                if not args.shutdown:
-                    change_containers_status(stoppable, "start")
-                continue
-
             copy(authoritative=False)
             if requires_stop(containers, args.images_no_stop_required):
                 stoppable = filter_stoppable(containers)

src/baudolo/backup/cli.py

@@ -1,13 +1,9 @@
 from __future__ import annotations
 
 import argparse
-import os
 
 
 def parse_args() -> argparse.Namespace:
-    dirname = os.path.dirname(__file__)
-    default_databases_csv = os.path.join(dirname, "databases.csv")
-
     p = argparse.ArgumentParser(description="Backup Docker volumes.")
 
     p.add_argument(
@@ -25,13 +21,12 @@ def parse_args() -> argparse.Namespace:
 
     p.add_argument(
         "--repo-name",
-        default="backup-docker-to-local",
-        help="Backup repo folder name under <backups-dir>/<machine-id>/ (default: git repo folder name)",
+        required=True,
+        help="Backup repo folder name under <backups-dir>/<machine-id>/",
     )
     p.add_argument(
         "--databases-csv",
-        default=default_databases_csv,
-        help=f"Path to databases.csv (default: {default_databases_csv})",
+        help="Path to databases.csv; required unless --only-files is given",
     )
     p.add_argument(
         "--backups-dir",
@@ -75,19 +70,15 @@ def parse_args() -> argparse.Namespace:
         help="Exact volume names that are never backed up, whatever containers use them. For derived trees a restore cannot reproduce, above all a nested docker data root",
     )
 
-    p.add_argument(
-        "--everything",
-        action="store_true",
-        help="Force file backup for all volumes and also execute database dumps (like old script)",
-    )
     p.add_argument(
         "--shutdown",
         action="store_true",
         help="Do not restart containers after backup",
     )
 
-    p.add_argument(
-        "--dump-only-sql",
+    scope = p.add_mutually_exclusive_group()
+    scope.add_argument(
+        "--only-sql",
         action="store_true",
         help=(
             "Create database dumps only for DB volumes. "
@@ -96,7 +87,19 @@ def parse_args() -> argparse.Namespace:
             "If a DB dump cannot be produced, baudolo falls back to a file backup."
         ),
     )
+    scope.add_argument(
+        "--only-files",
+        action="store_true",
+        help=(
+            "Take no database dumps at all and back up every volume as files. "
+            "For hosts that hold no database credentials. A database's files "
+            "are only consistent if its containers are stopped for the second "
+            "pass, so keep its image off --images-no-stop-required."
+        ),
+    )
     args = p.parse_args()
+    if not args.only_files and not args.databases_csv:
+        p.error("--databases-csv is required unless --only-files is given")
     if bool(args.snapshot) != bool(args.snapshot_subject):
         p.error("--snapshot and --snapshot-subject must be given together")
     if args.snapshot and args.shutdown:

src/baudolo/backup/db.py

@@ -7,7 +7,10 @@ import re
 
 import pandas
 
-from .shell import BackupException, execute_shell_command
+from baudolo.databases import CLUSTER_ROW, validate_database
+
+from .docker import docker_exec_argv
+from .shell import BackupException, execute_to_file
 
 log = logging.getLogger(__name__)
 
@@ -21,47 +24,21 @@ def get_instance(container: str, database_containers: list[str]) -> str:
     return re.split(r"(_|-)(database|db|postgres)", container)[0]
 
 
-def _validate_database_value(value: str | None, *, instance: str) -> str:
-    """
-    Enforce explicit database semantics:
-
-    - "*"       => dump ALL databases (cluster dump for Postgres)
-    - "<name>"  => dump exactly this database
-    - ""        => invalid configuration (would previously result in NaN / nan.backup.sql)
-    """
-    v = (value or "").strip()
-    if v == "":
-        raise ValueError(
-            f"Invalid databases.csv entry for instance '{instance}': "
-            "column 'database' must be '*' or a concrete database name (not empty)."
-        )
-    return v
-
-
-def _atomic_write_cmd(cmd: str, out_file: str) -> None:
-    """
-    Write dump output atomically:
-    - write to <file>.tmp
-    - rename to <file> only on success
-
-    This prevents empty or partial dump files from being treated as valid backups.
-    """
-    tmp = f"{out_file}.tmp"
-    execute_shell_command(f"{cmd} > {tmp}")
-    execute_shell_command(f"mv {tmp} {out_file}")
-
-
 def fallback_pg_dumpall(
     container: str, username: str, password: str, out_file: str
 ) -> None:
     """
     Perform a full Postgres cluster dump using pg_dumpall.
     """
-    cmd = (
-        f"PGPASSWORD={password} docker exec -i {container} "
-        f"pg_dumpall -U {username} -h localhost"
+    execute_to_file(
+        docker_exec_argv(
+            container,
+            ["pg_dumpall", "-U", username, "-h", "localhost"],
+            interactive=True,
+        ),
+        out_file,
+        env={"PGPASSWORD": password},
     )
-    _atomic_write_cmd(cmd, out_file)
 
 
 def backup_database(
@@ -69,12 +46,17 @@ def backup_database(
     container: str,
     volume_dir: str,
     db_type: str,
+    dump_tool: str,
     databases_df: pandas.DataFrame,
     database_containers: list[str],
 ) -> bool:
     """
     Backup databases for a given DB container.
 
+    Args:
+        dump_tool: the MariaDB client found in the container, so an image
+            that ships only mysqldump is dumped with the tool it has.
+
     Returns True if at least one dump was produced.
     """
     instance_name = get_instance(container, database_containers)
@@ -94,13 +76,13 @@ def backup_database(
         user = (getattr(row, "username", "") or "").strip()
         password = (getattr(row, "password", "") or "").strip()
 
-        db_value = _validate_database_value(raw_db, instance=instance_name)
+        db_value = validate_database(raw_db, instance=instance_name)
 
-        if db_value == "*":
+        if db_value == CLUSTER_ROW:
             if db_type != "postgres":
                 raise ValueError(
                     f"databases.csv entry for instance '{instance_name}': "
-                    "'*' is currently only supported for Postgres."
+                    f"'{CLUSTER_ROW}' is currently only supported for Postgres."
                 )
 
             cluster_file = os.path.join(out_dir, f"{instance_name}.cluster.backup.sql")
@@ -113,23 +95,46 @@ def backup_database(
 
         if db_type == "mariadb":
             # Force TCP so auth matches '<user>'@'%' instead of socket -> 'localhost'.
-            cmd = (
-                f"docker exec {container} /usr/bin/mariadb-dump "
-                f"-h 127.0.0.1 --protocol=tcp "
-                f"-u {user} -p{password} {db_name}"
+            execute_to_file(
+                docker_exec_argv(
+                    container,
+                    [
+                        dump_tool,
+                        "-h",
+                        "127.0.0.1",
+                        "--protocol=tcp",
+                        "-u",
+                        user,
+                        f"-p{password}",
+                        db_name,
+                    ],
+                ),
+                dump_file,
             )
-            _atomic_write_cmd(cmd, dump_file)
             produced = True
             continue
 
         if db_type == "postgres":
             try:
-                cmd = (
-                    f"PGPASSWORD={password} docker exec -i {container} "
-                    f"pg_dump -U {user} -d {db_name} -h localhost "
-                    f"--no-owner --no-privileges"
+                execute_to_file(
+                    docker_exec_argv(
+                        container,
+                        [
+                            "pg_dump",
+                            "-U",
+                            user,
+                            "-d",
+                            db_name,
+                            "-h",
+                            "localhost",
+                            "--no-owner",
+                            "--no-privileges",
+                        ],
+                        interactive=True,
+                    ),
+                    dump_file,
+                    env={"PGPASSWORD": password},
                 )
-                _atomic_write_cmd(cmd, dump_file)
                 produced = True
             except BackupException as e:
                 raise BackupException(

src/baudolo/backup/docker.py

@@ -1,46 +1,58 @@
 from __future__ import annotations
 
+from collections.abc import Sequence
+
 from .shell import BackupException, execute_shell_command
 
 
+def docker_exec_argv(
+    container: str, argv: Sequence[str], *, interactive: bool = False
+) -> list[str]:
+    """The argv that runs *argv* inside *container*."""
+    return ["docker", "exec", *(["-i"] if interactive else []), container, *argv]
+
+
 def get_image_info(container: str) -> str:
     return execute_shell_command(
-        f"docker inspect --format '{{{{.Config.Image}}}}' {container}"
+        ["docker", "inspect", "--format", "{{.Config.Image}}", container]
     )[0]
 
 
-def image_name(container: str) -> str:
-    """The image's repository path, without registry host, tag or digest.
+def image_id(container: str) -> str:
+    """The container's image ID, identical for every replica of one image."""
+    return execute_shell_command(
+        ["docker", "inspect", "--format", "{{.Image}}", container]
+    )[0].strip()
 
-    A swarm node that hosts the local registry puts its own hostname in front
-    of every pull, so the raw reference of a Postgres container can read
-    `svc-db-mariadb-swarm-mgr-01:5000/postgres_custom:17-3.5`. Matching the
-    whole reference finds "mariadb" there and dumps the database with
-    mariadb-dump, which the Postgres image does not ship (exit 127). Tags bite
-    the same way: `xwiki_custom:lts-postgres-tomcat`.
+
+def has_tool(container: str, tool: str) -> bool:
+    """Whether *tool* runs inside the container.
+
+    Executes the binary rather than asking a shell for it: a distroless image
+    has no shell, and `sh -c 'command -v'` would answer "absent" for every
+    tool it ships.
     """
-    reference = get_image_info(container).strip().split("@", 1)[0]
-    head, _, tail = reference.rpartition("/")
-    tail = tail.split(":", 1)[0]
-    if head:
-        registry = head.split("/", 1)[0]
-        if "." in registry or ":" in registry or registry == "localhost":
-            head = head.partition("/")[2]
-    return f"{head}/{tail}" if head else tail
-
-
-def has_image(container: str, pattern: str) -> bool:
-    """Return True if the container's image name contains the pattern."""
-    return pattern in image_name(container)
+    try:
+        execute_shell_command(docker_exec_argv(container, [tool, "--version"]))
+    except BackupException:
+        return False
+    return True
 
 
 def docker_volume_names() -> list[str]:
-    return execute_shell_command("docker volume ls --format '{{.Name}}'")
+    return execute_shell_command(["docker", "volume", "ls", "--format", "{{.Name}}"])
 
 
 def containers_using_volume(volume_name: str) -> list[str]:
     return execute_shell_command(
-        f"docker ps --filter volume=\"{volume_name}\" --format '{{{{.Names}}}}'"
+        [
+            "docker",
+            "ps",
+            "--filter",
+            f"volume={volume_name}",
+            "--format",
+            "{{.Names}}",
+        ]
     )
 
 
@@ -54,12 +66,25 @@ def is_swarm_task(container: str) -> bool:
     keeps failing the run loudly instead of silently skipping the stop."""
     try:
         out = execute_shell_command(
-            "docker inspect --format "
-            f"'{{{{index .Config.Labels \"com.docker.swarm.task.id\"}}}}' {container}"
+            [
+                "docker",
+                "inspect",
+                "--format",
+                '{{index .Config.Labels "com.docker.swarm.task.id"}}',
+                container,
+            ]
         )
     except BackupException:
         still_listed = execute_shell_command(
-            f"docker ps -a --filter name=^{container}$ --format '{{{{.Names}}}}'"
+            [
+                "docker",
+                "ps",
+                "-a",
+                "--filter",
+                f"name=^{container}$",
+                "--format",
+                "{{.Names}}",
+            ]
         )
         if still_listed and still_listed[0].strip():
             raise
@@ -86,17 +111,5 @@ def change_containers_status(containers: list[str], status: str) -> None:
     if not containers:
         print(f"No containers to {status}.", flush=True)
         return
-    names = " ".join(containers)
-    print(f"{status.capitalize()} containers: {names}...", flush=True)
-    execute_shell_command(f"docker {status} {names}")
-
-
-def docker_volume_exists(volume: str) -> bool:
-    # Avoid throwing exceptions for exists checks.
-    try:
-        execute_shell_command(
-            f"docker volume inspect {volume} >/dev/null 2>&1 && echo OK"
-        )
-        return True
-    except BackupException:
-        return False
+    print(f"{status.capitalize()} containers: {' '.join(containers)}...", flush=True)
+    execute_shell_command(["docker", status, *containers])

src/baudolo/backup/dumps.py

@@ -7,8 +7,48 @@ import sys
 import pandas
 from pandas.errors import EmptyDataError
 
+from baudolo.databases import COLUMNS, DELIMITER
+
 from .db import backup_database
-from .docker import has_image
+from .docker import has_tool, image_id
+
+DUMP_TOOLS: tuple[tuple[str, str], ...] = (
+    ("postgres", "pg_dumpall"),
+    ("mariadb", "mariadb-dump"),
+    ("mariadb", "mysqldump"),
+)
+
+_ENGINE_BY_IMAGE: dict[str, tuple[str, str] | None] = {}
+
+
+def container_engine(container: str) -> tuple[str, str] | None:
+    """The (engine, dump tool) a container can serve, or None for neither.
+
+    Asks the container what it can run instead of reading its image name. A
+    dedicated Postgres is tagged `<app>-database` or `postgis/postgis` and
+    carries no engine token at all, while a swarm registry host such as
+    `svc-db-mariadb-swarm-mgr-01:5000` carries the wrong one.
+
+    Args:
+        container: must be running - `docker exec` is the probe, and a
+            stopped container would be cached as "no engine" for its whole
+            image. The only caller feeds it `docker ps` output.
+
+    Returns:
+        The engine and the tool that dumps it, cached per image ID so that
+        replicas of one image are probed once.
+    """
+    image = image_id(container)
+    if image not in _ENGINE_BY_IMAGE:
+        _ENGINE_BY_IMAGE[image] = next(
+            (
+                (engine, tool)
+                for engine, tool in DUMP_TOOLS
+                if has_tool(container, tool)
+            ),
+            None,
+        )
+    return _ENGINE_BY_IMAGE[image]
 
 
 def backup_mariadb_or_postgres(
@@ -21,17 +61,19 @@ def backup_mariadb_or_postgres(
     """
     Returns (is_db_container, dumped_any)
     """
-    for img in ["mariadb", "postgres"]:
-        if has_image(container, img):
-            dumped = backup_database(
-                container=container,
-                volume_dir=volume_dir,
-                db_type=img,
-                databases_df=databases_df,
-                database_containers=database_containers,
-            )
-            return True, dumped
-    return False, False
+    engine = container_engine(container)
+    if engine is None:
+        return False, False
+    db_type, dump_tool = engine
+    dumped = backup_database(
+        container=container,
+        volume_dir=volume_dir,
+        db_type=db_type,
+        dump_tool=dump_tool,
+        databases_df=databases_df,
+        database_containers=database_containers,
+    )
+    return True, dumped
 
 
 def _empty_databases_df() -> pandas.DataFrame:
@@ -41,7 +83,7 @@ def _empty_databases_df() -> pandas.DataFrame:
     This allows the backup to continue without DB dumps when the CSV is missing
     or empty (pandas EmptyDataError).
     """
-    return pandas.DataFrame(columns=["instance", "database", "username", "password"])
+    return pandas.DataFrame(columns=list(COLUMNS))
 
 
 def load_databases_df(csv_path: str) -> pandas.DataFrame:
@@ -53,7 +95,9 @@ def load_databases_df(csv_path: str) -> pandas.DataFrame:
     - Valid CSV        -> return dataframe
     """
     try:
-        return pandas.read_csv(csv_path, sep=";", keep_default_na=False, dtype=str)
+        return pandas.read_csv(
+            csv_path, sep=DELIMITER, keep_default_na=False, dtype=str
+        )
     except FileNotFoundError:
         print(
             f"WARNING: databases.csv not found: {csv_path}. Continuing without database dumps.",

src/baudolo/backup/layout.py

@@ -11,7 +11,7 @@ from .shell import BackupException, execute_shell_command
 
 
 def get_machine_id() -> str:
-    return execute_shell_command("sha256sum /etc/machine-id")[0][0:64]
+    return execute_shell_command(["sha256sum", "/etc/machine-id"])[0][0:64]
 
 
 def stamp_directory(version_dir: str) -> None:

src/baudolo/backup/shell.py

@@ -1,26 +1,71 @@
+"""Running external commands without a shell.
+
+Every command is an argv list. A database name, a password or a container name
+therefore cannot close a quote and start a second command, which a formatted
+string handed to ``shell=True`` allowed.
+"""
+
 from __future__ import annotations
 
+import os
 import subprocess
+from collections.abc import Mapping, Sequence
 
 
 class BackupException(Exception):
     """Generic exception for backup errors."""
 
 
-def execute_shell_command(command: str) -> list[str]:
-    """Execute a shell command and return its output lines."""
-    print(command, flush=True)
+def _child_env(env: Mapping[str, str] | None) -> dict[str, str] | None:
+    return None if env is None else {**os.environ, **env}
+
+
+def _fail(command: Sequence[str], returncode: int, out: bytes, err: bytes) -> None:
+    raise BackupException(
+        f"Error in command: {' '.join(command)}\n"
+        f"Output: {out}\nError: {err}\n"
+        f"Exit code: {returncode}"
+    )
+
+
+def execute_shell_command(
+    command: Sequence[str], *, env: Mapping[str, str] | None = None
+) -> list[str]:
+    """Run *command* and return its stdout lines.
+
+    Args:
+        command: argv, the program first.
+        env: variables added to the child's environment, for values that must
+            not appear in the argv of a process listing.
+    """
+    command = list(command)
+    print(" ".join(command), flush=True)
     process = subprocess.Popen(
-        [command],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        shell=True,
+        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=_child_env(env)
     )
     out, err = process.communicate()
     if process.returncode != 0:
-        raise BackupException(
-            f"Error in command: {command}\n"
-            f"Output: {out}\nError: {err}\n"
-            f"Exit code: {process.returncode}"
-        )
+        _fail(command, process.returncode, out, err)
     return [line.decode("utf-8") for line in out.splitlines()]
+
+
+def execute_to_file(
+    command: Sequence[str], out_file: str, *, env: Mapping[str, str] | None = None
+) -> None:
+    """Run *command*, writing its stdout to *out_file* only once it succeeded.
+
+    The output goes to a sibling temporary file first, so a partial or empty
+    stream from a failing dump never takes the place of a valid backup.
+    """
+    command = list(command)
+    print(" ".join(command), flush=True)
+    tmp = f"{out_file}.tmp"
+    with open(tmp, "wb") as handle:
+        process = subprocess.Popen(
+            command, stdout=handle, stderr=subprocess.PIPE, env=_child_env(env)
+        )
+        _, err = process.communicate()
+    if process.returncode != 0:
+        os.unlink(tmp)
+        _fail(command, process.returncode, b"", err)
+    os.replace(tmp, out_file)

src/baudolo/backup/snapshot.py

@@ -48,23 +48,27 @@ def _resolver(subject: str, root: str) -> Callable[[str], str]:
     return resolve
 
 
-def _btrfs(subject: str, name: str, run: Callable[[str], list[str]]) -> tuple[str, str]:
+def _btrfs(
+    subject: str, name: str, run: Callable[[list[str]], list[str]]
+) -> tuple[str, list[str]]:
     # The snapshot goes inside the subject, never beside it: the kernel rejects
     # a snapshot whose destination is on another filesystem, which is exactly
     # what the parent directory is when the subject is a mountpoint of its own.
     target = os.path.join(os.path.abspath(subject), f".{name}")
-    run(f"btrfs subvolume snapshot -r {subject} {target}")
-    return target, f"btrfs subvolume delete {target}"
+    run(["btrfs", "subvolume", "snapshot", "-r", subject, target])
+    return target, ["btrfs", "subvolume", "delete", target]
 
 
-def _zfs(subject: str, name: str, run: Callable[[str], list[str]]) -> tuple[str, str]:
-    output = run(f"zfs list -H -o name {subject}")
+def _zfs(
+    subject: str, name: str, run: Callable[[list[str]], list[str]]
+) -> tuple[str, list[str]]:
+    output = run(["zfs", "list", "-H", "-o", "name", subject])
     dataset = (output[0] if output else "").strip()
     if not dataset:
         raise SnapshotError(f"no zfs dataset is mounted at {subject}")
-    run(f"zfs snapshot {dataset}@{name}")
+    run(["zfs", "snapshot", f"{dataset}@{name}"])
     root = os.path.join(subject, ".zfs", "snapshot", name)
-    return root, f"zfs destroy {dataset}@{name}"
+    return root, ["zfs", "destroy", f"{dataset}@{name}"]
 
 
 _CREATE = {"btrfs": _btrfs, "zfs": _zfs}
@@ -129,7 +133,7 @@ def volume_snapshot(
     kind: str,
     subject: str,
     tag: str,
-    run: Callable[[str], list[str]] = execute_shell_command,
+    run: Callable[[list[str]], list[str]] = execute_shell_command,
 ) -> Iterator[Callable[[str], str]]:
     """Yield a resolver mapping a path under ``subject`` into a snapshot of it.
 

src/baudolo/backup/volume.py

@@ -30,7 +30,7 @@ class Backing:
 
 def inspect_backing(volume_name: str) -> Backing:
     reported = execute_shell_command(
-        f"docker volume inspect --format '{{{{json .}}}}' {volume_name}"
+        ["docker", "volume", "inspect", "--format", "{{json .}}", volume_name]
     )[0]
     data = json.loads(reported)
     return Backing(
@@ -73,13 +73,12 @@ def backup_volume(
     pathlib.Path(dest).mkdir(parents=True, exist_ok=True)
 
     last = get_last_backup_dir(versions_dir, volume_name, dest)
-    link_dest = f"--link-dest='{last}'" if last else ""
-    verify = "--checksum " if authoritative else ""
-
-    cmd = (
-        f"rsync -aP --no-D --delete --delete-excluded "
-        f"{verify}{link_dest} {source} {dest}"
-    )
+    cmd = ["rsync", "-aP", "--no-D", "--delete", "--delete-excluded"]
+    if authoritative:
+        cmd.append("--checksum")
+    if last:
+        cmd.append(f"--link-dest={last}")
+    cmd += [source, dest]
 
     try:
         execute_shell_command(cmd)

src/baudolo/databases.py

@@ -0,0 +1,102 @@
+"""The databases.csv contract: its columns, its delimiter, and what a row means.
+
+``baudolo-seed`` writes the file, the backup reads it to learn which dumps to
+take, and a restore consumer reads it again to replay them. Stating the schema
+once keeps a column or a convention added here from being invisible to the
+other two.
+
+Field values are handed back exactly as they stand in the file. A password may
+legitimately begin or end with a space, so stripping belongs to the caller that
+compares, never to the reader.
+"""
+
+from __future__ import annotations
+
+import csv
+import re
+from typing import NamedTuple
+
+COLUMNS = ("instance", "database", "username", "password")
+DELIMITER = ";"
+CLUSTER_ROW = "*"
+
+_NAME_RE = re.compile(r"^[a-zA-Z0-9_][a-zA-Z0-9_-]*$")
+
+
+class DatabasesCsvError(ValueError):
+    """A row does not match the contract."""
+
+
+class Row(NamedTuple):
+    """One databases.csv row, verbatim.
+
+    ``database`` holds :data:`CLUSTER_ROW` when the whole instance is dumped.
+    """
+
+    instance: str
+    database: str
+    username: str
+    password: str
+
+    @property
+    def is_cluster(self) -> bool:
+        return self.database.strip() == CLUSTER_ROW
+
+
+def validate_database(value: str | None, *, instance: str) -> str:
+    """The database column of one row, or raise.
+
+    The name reaches a shell as part of the dump command, so it is checked
+    where it is read as well as where it is written: a file edited by hand
+    never passed the seed.
+
+    Args:
+        value: the raw column.
+        instance: named in the error, so a bad row can be found.
+
+    Raises:
+        DatabasesCsvError: the column is empty, literally ``nan``, or holds
+            anything but letters, numbers, ``_`` and ``-``.
+    """
+    text = (value or "").strip()
+    if not text:
+        raise DatabasesCsvError(
+            f"Invalid databases.csv entry for instance '{instance}': column "
+            f"'database' must be '{CLUSTER_ROW}' or a concrete database name "
+            "(not empty)."
+        )
+    if text == CLUSTER_ROW:
+        return CLUSTER_ROW
+    if text.lower() == "nan":
+        raise DatabasesCsvError(
+            f"Invalid databases.csv entry for instance '{instance}': "
+            "database must not be 'nan'."
+        )
+    if not _NAME_RE.match(text):
+        raise DatabasesCsvError(
+            f"Invalid databases.csv entry for instance '{instance}': invalid "
+            f"database name '{text}'. Allowed: letters, numbers, '_' and '-'."
+        )
+    return text
+
+
+def read_rows(csv_path: str) -> list[Row]:
+    """Every row of the file in file order, header skipped, blank rows dropped.
+
+    Raises:
+        DatabasesCsvError: a row holds fewer columns than :data:`COLUMNS`.
+    """
+    rows: list[Row] = []
+    with open(csv_path, newline="", encoding="utf-8") as handle:
+        reader = csv.reader(handle, delimiter=DELIMITER)
+        next(reader, None)
+        for raw in reader:
+            if not any(field.strip() for field in raw):
+                continue
+            if len(raw) < len(COLUMNS):
+                raise DatabasesCsvError(
+                    f"{csv_path} has a row with {len(raw)} column(s), "
+                    f"expected {len(COLUMNS)}"
+                )
+            rows.append(Row(*raw[: len(COLUMNS)]))
+    return rows

src/baudolo/restore/__main__.py

@@ -22,8 +22,8 @@ def _add_common_backup_args(p: argparse.ArgumentParser) -> None:
     )
     p.add_argument(
         "--repo-name",
-        default="backup-docker-to-local",
-        help="Backup repo folder name under <backups-dir>/<hash>/ (default: backup-docker-to-local)",
+        required=True,
+        help="Backup repo folder name under <backups-dir>/<hash>/",
     )
 
 

src/baudolo/seed/__main__.py

@@ -2,38 +2,16 @@ from __future__ import annotations
 
 import argparse
 import os
-import re
 import sys
 
 import pandas as pd
 from pandas.errors import EmptyDataError
 
-DB_NAME_RE = re.compile(r"^[a-zA-Z0-9_][a-zA-Z0-9_-]*$")
-
-
-def _validate_database_value(value: str | None, *, instance: str) -> str:
-    v = (value or "").strip()
-    if v == "":
-        raise ValueError(
-            f"Invalid databases.csv entry for instance '{instance}': "
-            "column 'database' must be '*' or a concrete database name (not empty)."
-        )
-    if v == "*":
-        return "*"
-    if v.lower() == "nan":
-        raise ValueError(
-            f"Invalid databases.csv entry for instance '{instance}': database must not be 'nan'."
-        )
-    if not DB_NAME_RE.match(v):
-        raise ValueError(
-            f"Invalid databases.csv entry for instance '{instance}': "
-            f"invalid database name '{v}'. Allowed: letters, numbers, '_' and '-'."
-        )
-    return v
+from baudolo.databases import COLUMNS, DELIMITER, validate_database
 
 
 def _empty_df() -> pd.DataFrame:
-    return pd.DataFrame(columns=["instance", "database", "username", "password"])
+    return pd.DataFrame(columns=list(COLUMNS))
 
 
 def check_and_add_entry(
@@ -50,13 +28,13 @@ def check_and_add_entry(
     - database MUST be set
     - database MUST be '*' or a valid database name
     """
-    database = _validate_database_value(database, instance=instance)
+    database = validate_database(database, instance=instance)
 
     if os.path.exists(file_path):
         try:
             df = pd.read_csv(
                 file_path,
-                sep=";",
+                sep=DELIMITER,
                 dtype=str,
                 keep_default_na=False,
             )
@@ -77,11 +55,11 @@ def check_and_add_entry(
         print("Adding new entry.")
         new_entry = pd.DataFrame(
             [[instance, database, username, password]],
-            columns=["instance", "database", "username", "password"],
+            columns=list(COLUMNS),
         )
         df = pd.concat([df, new_entry], ignore_index=True)
 
-    df.to_csv(file_path, sep=";", index=False)
+    df.to_csv(file_path, sep=DELIMITER, index=False)
 
 
 def main() -> None:

tests/e2e/faithful_driver.py

@@ -23,13 +23,11 @@ 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
-    )
+def shell(command: list[str]) -> list[str]:
+    proc = subprocess.run(command, capture_output=True, text=True, check=False)
     if proc.returncode != 0:
         raise SnapshotError(
-            f"{command} exited {proc.returncode}: {proc.stderr.strip()}"
+            f"{' '.join(command)} exited {proc.returncode}: {proc.stderr.strip()}"
         )
     return proc.stdout.splitlines()
 
@@ -50,7 +48,7 @@ def volume(name: str, payload: str) -> 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}")
+shell(["mount", "-t", "tmpfs", "tmpfs", own])
 (own / "state").write_text("own-payload")
 
 check("a plain volume is captured", unsnapshotted(Backing(str(plain)), SUBJECT) is None)

tests/e2e/helpers/fixtures.py

@@ -24,7 +24,7 @@ def backup_run(
     database_containers: list[str],
     images_no_stop_required: list[str],
     images_no_backup_required: list[str] | None = None,
-    dump_only_sql: bool = False,
+    only_sql: bool = False,
 ) -> None:
     cmd = [
         "baudolo",
@@ -45,8 +45,8 @@ def backup_run(
     ]
     if images_no_backup_required:
         cmd += ["--images-no-backup-required", *images_no_backup_required]
-    if dump_only_sql:
-        cmd += ["--dump-only-sql"]
+    if only_sql:
+        cmd += ["--only-sql"]
 
     try:
         run(cmd, capture=True, check=True)

tests/e2e/snapshot_db_driver.py

@@ -22,13 +22,11 @@ VERSIONS = "/backups"
 GENERATION = f"{VERSIONS}/20260731"
 
 
-def shell(command: str) -> list[str]:
-    proc = subprocess.run(
-        command, shell=True, capture_output=True, text=True, check=False
-    )
+def shell(command: list[str]) -> list[str]:
+    proc = subprocess.run(command, capture_output=True, text=True, check=False)
     if proc.returncode != 0:
         raise SnapshotError(
-            f"{command} exited {proc.returncode}: {proc.stderr.strip()}"
+            f"{' '.join(command)} exited {proc.returncode}: {proc.stderr.strip()}"
         )
     return proc.stdout.splitlines()
 

tests/e2e/snapshot_driver.py

@@ -19,13 +19,11 @@ SUBJECT = sys.argv[2]
 EXPECT = sys.argv[3]
 
 
-def shell(command: str) -> list[str]:
-    proc = subprocess.run(
-        command, shell=True, capture_output=True, text=True, check=False
-    )
+def shell(command: list[str]) -> list[str]:
+    proc = subprocess.run(command, capture_output=True, text=True, check=False)
     if proc.returncode != 0:
         raise SnapshotError(
-            f"{command} exited {proc.returncode}: {proc.stderr.strip()}"
+            f"{' '.join(command)} exited {proc.returncode}: {proc.stderr.strip()}"
         )
     return proc.stdout.splitlines()
 

tests/e2e/test_e2e_cli_contract_dump_only_sql.py

@@ -1,29 +0,0 @@
-import unittest
-
-from .helpers import run
-
-
-class TestE2ECLIContractDumpOnlySql(unittest.TestCase):
-    def test_help_mentions_new_flag(self) -> None:
-        cp = run(["baudolo", "--help"], capture=True, check=True)
-        out = (cp.stdout or "") + "\n" + (cp.stderr or "")
-        self.assertIn(
-            "--dump-only-sql",
-            out,
-            f"Expected '--dump-only-sql' to appear in --help output. Output:\n{out}",
-        )
-
-    def test_old_flag_is_rejected(self) -> None:
-        cp = run(["baudolo", "--dump-only"], capture=True, check=False)
-        self.assertEqual(
-            cp.returncode,
-            2,
-            f"Expected exitcode 2 for unknown args, got {cp.returncode}\n"
-            f"STDOUT={cp.stdout}\nSTDERR={cp.stderr}",
-        )
-        err = (cp.stderr or "") + "\n" + (cp.stdout or "")
-        # Argparse typically prints "unrecognized arguments"
-        self.assertTrue(
-            ("unrecognized arguments" in err) or ("usage:" in err.lower()),
-            f"Expected argparse-style error output. Output:\n{err}",
-        )

tests/e2e/test_e2e_cli_contract_only_sql.py

@@ -0,0 +1,33 @@
+import unittest
+
+from .helpers import run
+
+WITHDRAWN_FLAGS = ["--dump-only", "--dump-only-sql", "--everything"]
+
+
+class TestE2ECLIContractOnlySql(unittest.TestCase):
+    def test_help_mentions_the_flag(self) -> None:
+        cp = run(["baudolo", "--help"], capture=True, check=True)
+        out = (cp.stdout or "") + "\n" + (cp.stderr or "")
+        self.assertIn(
+            "--only-sql",
+            out,
+            f"Expected '--only-sql' to appear in --help output. Output:\n{out}",
+        )
+
+    def test_a_withdrawn_flag_is_rejected(self) -> None:
+        for flag in WITHDRAWN_FLAGS:
+            with self.subTest(flag=flag):
+                cp = run(["baudolo", flag], capture=True, check=False)
+                self.assertEqual(
+                    cp.returncode,
+                    2,
+                    f"Expected exitcode 2 for unknown args, got {cp.returncode}\n"
+                    f"STDOUT={cp.stdout}\nSTDERR={cp.stderr}",
+                )
+                err = (cp.stderr or "") + "\n" + (cp.stdout or "")
+                # Argparse typically prints "unrecognized arguments"
+                self.assertTrue(
+                    ("unrecognized arguments" in err) or ("usage:" in err.lower()),
+                    f"Expected argparse-style error output. Output:\n{err}",
+                )

tests/e2e/test_e2e_engine_detection_by_tool.py

@@ -0,0 +1,177 @@
+"""The engine comes from the tools a container ships, not from its image name.
+
+Two containers in one backup run, each lying in one direction:
+
+* a real Postgres tagged `<prefix>-database`, the way a dedicated database is
+  built inside an app's own stack - no engine token anywhere in the name;
+* an Alpine tagged `postgres:<prefix>`, carrying the token without shipping a
+  single Postgres binary.
+
+Reading the name gets both wrong, and the second one fatally: pg_dump exits 127
+inside Alpine and takes the whole run with it.
+"""
+
+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,
+)
+
+IMPOSTOR_BASE_IMAGE = "alpine:3.20"
+MARKER = "engine-detection-by-tool"
+
+
+class TestE2EEngineDetectionByTool(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-engine-by-tool")
+        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.engine_image = f"{cls.prefix}-database:17"
+        cls.impostor_image = f"postgres:{cls.prefix}"
+
+        cls.engine_container = f"{cls.prefix}-engine"
+        cls.impostor_container = f"{cls.prefix}-impostor"
+        cls.engine_volume = f"{cls.prefix}-engine-vol"
+        cls.impostor_volume = f"{cls.prefix}-impostor-vol"
+
+        cls.containers = [cls.engine_container, cls.impostor_container]
+        cls.volumes = [cls.engine_volume, cls.impostor_volume]
+
+        run(["docker", "pull", POSTGRES_IMAGE])
+        run(["docker", "pull", IMPOSTOR_BASE_IMAGE])
+        run(["docker", "tag", POSTGRES_IMAGE, cls.engine_image])
+        run(["docker", "tag", IMPOSTOR_BASE_IMAGE, cls.impostor_image])
+        run(["docker", "volume", "create", cls.engine_volume])
+        run(["docker", "volume", "create", cls.impostor_volume])
+
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.engine_container,
+                "-e",
+                "POSTGRES_PASSWORD=pgpw",
+                "-e",
+                "POSTGRES_DB=appdb",
+                "-e",
+                "POSTGRES_USER=postgres",
+                "-v",
+                f"{cls.engine_volume}:{POSTGRES_DATA_DIR}",
+                cls.engine_image,
+            ]
+        )
+        wait_for_postgres(cls.engine_container, user="postgres", timeout_s=90)
+        run(
+            [
+                "docker",
+                "exec",
+                cls.engine_container,
+                "sh",
+                "-lc",
+                (
+                    "psql -U postgres -d appdb -c "
+                    '"CREATE TABLE t (id int primary key, v text); '
+                    "INSERT INTO t VALUES (1,'ok');\""
+                ),
+            ]
+        )
+
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.impostor_container,
+                "-v",
+                f"{cls.impostor_volume}:/data",
+                cls.impostor_image,
+                "sh",
+                "-lc",
+                f"echo '{MARKER}' > /data/marker.txt && sleep 3600",
+            ]
+        )
+
+        cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
+        write_databases_csv(
+            cls.databases_csv,
+            [
+                (cls.engine_container, "appdb", "postgres", "pgpw"),
+                (cls.impostor_container, "appdb", "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.engine_container, cls.impostor_container],
+            images_no_stop_required=[cls.engine_image, cls.impostor_image],
+            only_sql=True,
+        )
+
+        cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=cls.containers, volumes=cls.volumes)
+        run(["docker", "rmi", cls.engine_image], check=False)
+        run(["docker", "rmi", cls.impostor_image], check=False)
+
+    def _volume_dir(self, volume: str):
+        return backup_path(self.backups_dir, self.repo_name, self.version, volume)
+
+    def test_an_engine_without_an_engine_name_is_still_dumped(self) -> None:
+        dump = self._volume_dir(self.engine_volume) / "sql" / "appdb.backup.sql"
+        self.assertTrue(
+            dump.is_file(),
+            f"a Postgres tagged '{self.engine_image}' produced no dump at {dump}",
+        )
+        self.assertIn("Dumped by pg_dump", dump.read_text(encoding="utf-8"))
+
+    def test_an_engine_name_without_an_engine_is_not_dumped(self) -> None:
+        sql_dir = self._volume_dir(self.impostor_volume) / "sql"
+        dumps = list(sql_dir.glob("*.sql")) if sql_dir.exists() else []
+        self.assertEqual(
+            dumps,
+            [],
+            f"'{self.impostor_image}' ships no Postgres yet was dumped: {dumps}",
+        )
+
+    def test_the_recognised_engine_is_dumped_instead_of_copied(self) -> None:
+        files = self._volume_dir(self.engine_volume) / "files"
+        self.assertFalse(
+            files.exists(),
+            f"--only-sql still copied the engine's files to {files}",
+        )
+
+    def test_the_impostor_falls_through_to_a_file_backup(self) -> None:
+        files = self._volume_dir(self.impostor_volume) / "files"
+        self.assertTrue(files.is_dir(), f"expected a file backup at {files}")
+        self.assertEqual(
+            (files / "marker.txt").read_text(encoding="utf-8").strip(), MARKER
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/e2e/test_e2e_files_no_copy.py

@@ -47,7 +47,7 @@ class TestE2EFilesNoCopy(unittest.TestCase):
         cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
         write_databases_csv(cls.databases_csv, [])
 
-        # dump-only-sql => non-DB volumes are STILL backed up as files
+        # only-sql => non-DB volumes are STILL backed up as files
         backup_run(
             backups_dir=cls.backups_dir,
             repo_name=cls.repo_name,
@@ -55,7 +55,7 @@ class TestE2EFilesNoCopy(unittest.TestCase):
             databases_csv=cls.databases_csv,
             database_containers=["dummy-db"],
             images_no_stop_required=["alpine:3.20"],
-            dump_only_sql=True,
+            only_sql=True,
         )
 
         cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)

tests/e2e/test_e2e_mariadb_anonymous_preemption.py

@@ -148,6 +148,7 @@ class TestE2EMariaDBAnonymousPreemption(unittest.TestCase):
                 container=self.db_container,
                 volume_dir=volume_dir,
                 db_type="mariadb",
+                dump_tool="mariadb-dump",
                 databases_df=df,
                 database_containers=[self.db_container],
             )

tests/e2e/test_e2e_mariadb_no_copy.py

@@ -89,7 +89,7 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
             [(cls.db_container, cls.db_name, cls.db_user, cls.db_password)],
         )
 
-        # dump-only-sql => no files
+        # only-sql => no files
         backup_run(
             backups_dir=cls.backups_dir,
             repo_name=cls.repo_name,
@@ -97,7 +97,7 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
             databases_csv=cls.databases_csv,
             database_containers=[cls.db_container],
             images_no_stop_required=[MARIADB_IMAGE],
-            dump_only_sql=True,
+            only_sql=True,
         )
 
         cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)

tests/e2e/test_e2e_only_files.py

@@ -0,0 +1,116 @@
+"""--only-files backs a database up as a file tree and asks for no credentials.
+
+The run deliberately passes no --databases-csv at all: a host that only copies
+files has no reason to hold database passwords, and requiring the file would
+make the flag useless there.
+"""
+
+import unittest
+
+from .helpers import (
+    POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
+    backup_path,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    wait_for_postgres,
+)
+
+MARKER = "only-files-marker"
+
+
+class TestE2EOnlyFiles(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-only-files")
+        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",
+                "-e",
+                "POSTGRES_DB=appdb",
+                "-e",
+                "POSTGRES_USER=postgres",
+                "-v",
+                f"{cls.pg_volume}:{POSTGRES_DATA_DIR}",
+                POSTGRES_IMAGE,
+            ]
+        )
+        wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
+        run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                f"echo '{MARKER}' > {POSTGRES_DATA_DIR}/marker.txt",
+            ]
+        )
+
+        cp = run(
+            [
+                "baudolo",
+                "--compose-dir",
+                cls.compose_dir,
+                "--repo-name",
+                cls.repo_name,
+                "--backups-dir",
+                cls.backups_dir,
+                "--images-no-stop-required",
+                POSTGRES_IMAGE,
+                "--only-files",
+            ],
+            capture=True,
+            check=True,
+        )
+        cls.stdout = cp.stdout or ""
+        cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=cls.containers, volumes=cls.volumes)
+
+    def _volume_dir(self):
+        return backup_path(
+            self.backups_dir, self.repo_name, self.version, self.pg_volume
+        )
+
+    def test_the_database_volume_is_backed_up_as_files(self) -> None:
+        marker = self._volume_dir() / "files" / "marker.txt"
+        self.assertTrue(marker.is_file(), f"expected a file backup at {marker}")
+        self.assertEqual(marker.read_text(encoding="utf-8").strip(), MARKER)
+
+    def test_no_dump_is_written(self) -> None:
+        sql_dir = self._volume_dir() / "sql"
+        dumps = list(sql_dir.glob("*.sql")) if sql_dir.exists() else []
+        self.assertEqual(dumps, [], f"did not expect any dump, found: {dumps}")
+
+    def test_the_missing_databases_csv_is_not_reported(self) -> None:
+        self.assertNotIn("databases.csv", self.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/e2e/test_e2e_only_sql_fallback_to_files.py

@@ -16,11 +16,11 @@ from .helpers import (
 )
 
 
-class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
+class TestE2EOnlySqlFallbackToFiles(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None:
         require_docker()
-        cls.prefix = unique("baudolo-e2e-dump-only-sql-fallback")
+        cls.prefix = unique("baudolo-e2e-only-sql-fallback")
         cls.backups_dir = f"/tmp/{cls.prefix}/Backups"
         ensure_empty_dir(cls.backups_dir)
 
@@ -57,7 +57,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
         wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
 
         # Add a deterministic marker file into the volume
-        cls.marker = "dump-only-sql-fallback-marker"
+        cls.marker = "only-sql-fallback-marker"
         run(
             [
                 "docker",
@@ -73,7 +73,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
         cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
         write_databases_csv(cls.databases_csv, [])  # empty except header
 
-        # Run baudolo with --dump-only-sql and a DB container present:
+        # Run baudolo with --only-sql and a DB container present:
         # Expected: WARNING + FALLBACK to file backup (files/ must exist)
         cmd = [
             "baudolo",
@@ -91,7 +91,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
             cls.pg_container,
             "--images-no-stop-required",
             POSTGRES_IMAGE,
-            "--dump-only-sql",
+            "--only-sql",
         ]
         cp = run(cmd, capture=True, check=True)
 
@@ -122,7 +122,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
 
     def test_warns_about_missing_dump_in_dump_only_mode(self) -> None:
         self.assertIn(
-            "WARNING: dump-only-sql requested but no DB dump was produced",
+            "WARNING: only-sql requested but no DB dump was produced",
             self.stdout,
             f"Expected warning in baudolo output. STDOUT:\n{self.stdout}",
         )

tests/e2e/test_e2e_only_sql_mixed_run.py

@@ -16,11 +16,11 @@ from .helpers import (
 )
 
 
-class TestE2EDumpOnlySqlMixedRun(unittest.TestCase):
+class TestE2EOnlySqlMixedRun(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None:
         require_docker()
-        cls.prefix = unique("baudolo-e2e-dump-only-sql-mixed-run")
+        cls.prefix = unique("baudolo-e2e-only-sql-mixed-run")
         cls.backups_dir = f"/tmp/{cls.prefix}/Backups"
         ensure_empty_dir(cls.backups_dir)
 
@@ -123,7 +123,7 @@ class TestE2EDumpOnlySqlMixedRun(unittest.TestCase):
             cls.pg_container,
             "--images-no-stop-required",
             POSTGRES_IMAGE,
-            "--dump-only-sql",
+            "--only-sql",
             "--backups-dir",
             cls.backups_dir,
             "--repo-name",
@@ -170,8 +170,8 @@ class TestE2EDumpOnlySqlMixedRun(unittest.TestCase):
             f"Expected files dir for non-DB volume at: {files}",
         )
 
-    def test_dump_only_sql_does_not_disable_non_db_files_backup(self) -> None:
-        # Regression guard: even with --dump-only-sql, non-DB volumes must still be backed up as files
+    def test_only_sql_does_not_disable_non_db_files_backup(self) -> None:
+        # Regression guard: even with --only-sql, non-DB volumes must still be backed up as files
         base = backup_path(
             self.backups_dir, self.repo_name, self.version, self.files_volume
         )

tests/e2e/test_e2e_postgres_no_copy.py

@@ -76,7 +76,7 @@ class TestE2EPostgresNoCopy(unittest.TestCase):
             databases_csv=cls.databases_csv,
             database_containers=[cls.pg_container],
             images_no_stop_required=[POSTGRES_IMAGE],
-            dump_only_sql=True,
+            only_sql=True,
         )
 
         cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)

tests/e2e/test_e2e_seed_star_and_db_entries_backup_postgres.py

@@ -157,7 +157,6 @@ class TestE2ESeedStarAndDbEntriesBackupPostgres(unittest.TestCase):
             ]
         )
 
-        # --- Run baudolo with dump-only-sql ---
         cmd = [
             "baudolo",
             "--compose-dir",
@@ -168,7 +167,7 @@ class TestE2ESeedStarAndDbEntriesBackupPostgres(unittest.TestCase):
             cls.pg_container,
             "--images-no-stop-required",
             POSTGRES_IMAGE,
-            "--dump-only-sql",
+            "--only-sql",
             "--backups-dir",
             cls.backups_dir,
             "--repo-name",
@@ -194,7 +193,7 @@ class TestE2ESeedStarAndDbEntriesBackupPostgres(unittest.TestCase):
         self.assertTrue(sql_dir.exists(), f"Expected sql dir at: {sql_dir}")
         self.assertFalse(
             files_dir.exists(),
-            f"Did not expect files dir for DB volume when dump-only-sql succeeded: {files_dir}",
+            f"Did not expect files dir for DB volume when only-sql succeeded: {files_dir}",
         )
 
         # Cluster dump file produced by '*' entry

tests/unit/backup/__init__.py

@@ -0,0 +1,10 @@
+"""The smallest argv the backup CLI accepts, shared by every test that drives it."""
+
+REQUIRED_PAIRS = [
+    ("--compose-dir", "/compose"),
+    ("--backups-dir", "/backups"),
+    ("--repo-name", "stack"),
+    ("--databases-csv", "/etc/baudolo/databases.csv"),
+]
+REQUIRED = [arg for pair in REQUIRED_PAIRS for arg in pair]
+BASE_ARGV = ["baudolo", *REQUIRED]

tests/unit/backup/test_app_only_files.py

@@ -0,0 +1,65 @@
+"""Contract of --only-files: no dump is attempted, every volume is copied."""
+
+from __future__ import annotations
+
+import unittest
+from unittest import mock
+
+from baudolo.backup import app
+from baudolo.backup.volume import Backing
+
+from . import REQUIRED_PAIRS
+
+ARGV_WITHOUT_CSV = [
+    "baudolo",
+    *[arg for pair in REQUIRED_PAIRS if pair[0] != "--databases-csv" for arg in pair],
+    "--only-files",
+]
+
+
+def drive(argv: list[str]) -> tuple[list[str], list, list]:
+    backed_up: list[str] = []
+
+    def record_backup(versions_dir, volume_name, volume_dir, *, authoritative, source):
+        backed_up.append(volume_name)
+
+    with (
+        mock.patch("sys.argv", argv),
+        mock.patch.object(app, "get_machine_id", return_value="machine"),
+        mock.patch.object(app, "create_version_directory", return_value="/gen"),
+        mock.patch.object(app, "create_volume_directory", return_value="/gen/vol"),
+        mock.patch.object(app, "load_databases_df") as load_csv,
+        mock.patch.object(app, "docker_volume_names", return_value=["pgdata"]),
+        mock.patch.object(app, "containers_using_volume", return_value=["db"]),
+        mock.patch.object(app, "volume_is_fully_ignored", return_value=False),
+        mock.patch.object(app, "backup_dumps_for_volume") as dumps,
+        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),
+        mock.patch.object(app, "backup_volume", side_effect=record_backup),
+        mock.patch.object(app, "filter_stoppable", return_value=[]),
+        mock.patch.object(app, "requires_stop", return_value=False),
+        mock.patch.object(app, "change_containers_status"),
+    ):
+        app.main()
+    return backed_up, dumps.mock_calls, load_csv.mock_calls
+
+
+class TestOnlyFiles(unittest.TestCase):
+    def test_no_dump_is_attempted(self) -> None:
+        _backed_up, dumps, _load_csv = drive(ARGV_WITHOUT_CSV)
+        self.assertEqual(dumps, [])
+
+    def test_the_databases_csv_is_never_read(self) -> None:
+        """It may legitimately be absent, so reading it would abort the run."""
+        _backed_up, _dumps, load_csv = drive(ARGV_WITHOUT_CSV)
+        self.assertEqual(load_csv, [])
+
+    def test_the_volume_is_still_copied(self) -> None:
+        backed_up, _dumps, _load_csv = drive(ARGV_WITHOUT_CSV)
+        self.assertEqual(backed_up, ["pgdata"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_app_snapshot.py

@@ -10,17 +10,15 @@ from baudolo.backup import snapshot as snapshot_mod
 from baudolo.backup.snapshot import volume_snapshot
 from baudolo.backup.volume import Backing
 
+from . import BASE_ARGV
+
 
 def stubbed_snapshot(kind: str, subject: str, tag: str):
     return volume_snapshot(kind, subject, tag, run=lambda command: [])
 
 
 ARGV = [
-    "baudolo",
-    "--compose-dir",
-    "/compose",
-    "--backups-dir",
-    "/backups",
+    *BASE_ARGV,
     "--snapshot",
     "btrfs",
     "--snapshot-subject",

tests/unit/backup/test_app_volumes_no_backup_required.py

@@ -9,12 +9,10 @@ from unittest import mock
 from baudolo.backup import app
 from baudolo.backup.volume import Backing
 
+from . import BASE_ARGV
+
 ARGV = [
-    "baudolo",
-    "--compose-dir",
-    "/compose",
-    "--backups-dir",
-    "/backups",
+    *BASE_ARGV,
     "--volumes-no-backup-required",
     "derived",
 ]

tests/unit/backup/test_cli.py

@@ -7,7 +7,7 @@ from unittest import mock
 
 from baudolo.backup.cli import parse_args
 
-REQUIRED = ["--compose-dir", "/compose", "--backups-dir", "/backups"]
+from . import REQUIRED, REQUIRED_PAIRS
 
 
 def parse(*extra: str):
@@ -68,19 +68,42 @@ 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"]),
-            self.assertRaises(SystemExit),
-        ):
-            parse_args()
+    def test_no_flag_falls_back_to_a_default(self) -> None:
+        for omitted, _ in REQUIRED_PAIRS:
+            argv = [a for pair in REQUIRED_PAIRS if pair[0] != omitted for a in pair]
+            with (
+                self.subTest(omitted=omitted),
+                mock.patch("sys.argv", ["baudolo", *argv]),
+                self.assertRaises(SystemExit),
+            ):
+                parse_args()
 
-    def test_compose_dir_is_required(self) -> None:
-        with (
-            mock.patch("sys.argv", ["baudolo", "--backups-dir", "/backups"]),
-            self.assertRaises(SystemExit),
-        ):
-            parse_args()
+
+class TestBackupScope(unittest.TestCase):
+    """--only-sql and --only-files name the two halves a generation can hold."""
+
+    def test_both_halves_by_default(self) -> None:
+        args = parse()
+        self.assertFalse(args.only_sql)
+        self.assertFalse(args.only_files)
+
+    def test_either_half_alone_is_accepted(self) -> None:
+        self.assertTrue(parse("--only-sql").only_sql)
+        self.assertTrue(parse("--only-files").only_files)
+
+    def test_asking_for_both_halves_alone_is_rejected(self) -> None:
+        with self.assertRaises(SystemExit):
+            parse("--only-sql", "--only-files")
+
+    def test_only_files_needs_no_databases_csv(self) -> None:
+        argv = [
+            arg
+            for pair in REQUIRED_PAIRS
+            if pair[0] != "--databases-csv"
+            for arg in pair
+        ]
+        with mock.patch("sys.argv", ["baudolo", *argv, "--only-files"]):
+            self.assertIsNone(parse_args().databases_csv)
 
 
 if __name__ == "__main__":

tests/unit/backup/test_compose_hard_restart.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 import unittest
 from unittest.mock import patch
 
+from . import BASE_ARGV
+
 
 class HardRestartArgTests(unittest.TestCase):
     """The hard-restart list defaults to empty (no compose down/up); callers
@@ -16,11 +18,7 @@ class HardRestartArgTests(unittest.TestCase):
         from baudolo.backup import cli
 
         argv = [
-            "baudolo",
-            "--compose-dir",
-            "/tmp",
-            "--backups-dir",
-            "/tmp/backup",
+            *BASE_ARGV,
             "--database-containers",
             "postgres",
             "--images-no-stop-required",
@@ -42,23 +40,6 @@ class HardRestartArgTests(unittest.TestCase):
         args = self._parse(["--hard-restart-projects", "mailu", "foo"])
         self.assertEqual(args.hard_restart_projects, ["mailu", "foo"])
 
-    def test_backups_dir_is_required(self) -> None:
-        import sys
-
-        from baudolo.backup import cli
-
-        argv = [
-            "baudolo",
-            "--compose-dir",
-            "/tmp",
-            "--database-containers",
-            "postgres",
-            "--images-no-stop-required",
-            "redis",
-        ]
-        with patch.object(sys, "argv", argv), self.assertRaises(SystemExit):
-            cli.parse_args()
-
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)

tests/unit/backup/test_db_mariadb_dump.py

@@ -13,21 +13,22 @@ def _df(rows):
     )
 
 
-def _capture_commands(*, db_type, rows, container):
+def _capture_dumps(*, db_type, rows, container, dump_tool="mariadb-dump"):
+    """Every (argv, env) the dump path would have run."""
     captured = []
 
-    def _capture(cmd):
-        captured.append(cmd)
-        return []
+    def _capture(command, out_file, *, env=None):
+        captured.append((list(command), env))
 
     with (
         tempfile.TemporaryDirectory() as td,
-        patch.object(db_mod, "execute_shell_command", side_effect=_capture),
+        patch.object(db_mod, "execute_to_file", side_effect=_capture),
     ):
         db_mod.backup_database(
             container=container,
             volume_dir=td,
             db_type=db_type,
+            dump_tool=dump_tool,
             databases_df=_df(rows),
             database_containers=[container],
         )
@@ -40,32 +41,64 @@ class TestMariaDBDumpUsesTCP(unittest.TestCase):
     # the connection is auth-matched against '%' instead of socket->localhost.
 
     def test_mariadb_dump_forces_tcp_loopback(self):
-        captured = _capture_commands(
+        captured = _capture_dumps(
             db_type="mariadb",
             rows=[("mariadb", "appdb", "appuser", "s3cret")],
             container="mariadb",
         )
-        dump_cmds = [c for c in captured if "mariadb-dump" in c]
-        self.assertEqual(
-            len(dump_cmds), 1, f"expected one dump command, got: {captured}"
-        )
+        self.assertEqual(len(captured), 1, f"expected one dump, got: {captured}")
 
-        cmd = dump_cmds[0]
-        self.assertIn("-h 127.0.0.1", cmd)
-        self.assertIn("--protocol=tcp", cmd)
-        self.assertIn("-u appuser", cmd)
-        self.assertIn("-ps3cret", cmd)
-        self.assertIn(" appdb", cmd)
+        argv, env = captured[0]
+        self.assertEqual(argv[:3], ["docker", "exec", "mariadb"])
+        self.assertIn("--protocol=tcp", argv)
+        self.assertEqual(argv[argv.index("-h") + 1], "127.0.0.1")
+        self.assertEqual(argv[argv.index("-u") + 1], "appuser")
+        self.assertIn("-ps3cret", argv)
+        self.assertEqual(argv[-1], "appdb")
+        self.assertIsNone(env)
+
+    def test_the_probed_client_is_the_one_invoked(self):
+        captured = _capture_dumps(
+            db_type="mariadb",
+            rows=[("mariadb", "appdb", "appuser", "s3cret")],
+            container="mariadb",
+            dump_tool="mysqldump",
+        )
+        argv, _env = captured[0]
+        self.assertIn("mysqldump", argv)
+        self.assertNotIn("mariadb-dump", argv)
 
     def test_postgres_dump_unaffected(self):
-        captured = _capture_commands(
+        captured = _capture_dumps(
             db_type="postgres",
             rows=[("pg", "appdb", "appuser", "s3cret")],
             container="pg",
         )
-        dump_cmds = [c for c in captured if "pg_dump" in c and "pg_dumpall" not in c]
-        self.assertEqual(len(dump_cmds), 1)
-        self.assertNotIn("--protocol=tcp", dump_cmds[0])
+        argv, _env = captured[0]
+        self.assertIn("pg_dump", argv)
+        self.assertNotIn("--protocol=tcp", argv)
+
+    def test_the_password_travels_in_the_environment_not_the_argv(self):
+        """A process listing shows argv; PGPASSWORD must not be in it."""
+        captured = _capture_dumps(
+            db_type="postgres",
+            rows=[("pg", "appdb", "appuser", "s3cret")],
+            container="pg",
+        )
+        argv, env = captured[0]
+        self.assertEqual(env, {"PGPASSWORD": "s3cret"})
+        self.assertNotIn("s3cret", argv)
+
+
+class TestNoShellReachesTheDump(unittest.TestCase):
+    def test_a_hostile_database_name_never_reaches_a_command(self):
+        """validate_database refuses it, so no argv is built at all."""
+        with self.assertRaises(ValueError):
+            _capture_dumps(
+                db_type="postgres",
+                rows=[("pg", "app;rm -rf /", "appuser", "s3cret")],
+                container="pg",
+            )
 
 
 if __name__ == "__main__":

tests/unit/backup/test_docker_image_name.py

@@ -1,50 +0,0 @@
-import unittest
-from unittest.mock import patch
-
-from baudolo.backup import docker as docker_mod
-
-
-def _with_image(reference: str):
-    return patch.object(docker_mod, "execute_shell_command", return_value=[reference])
-
-
-class TestImageName(unittest.TestCase):
-    def test_plain_reference(self) -> None:
-        with _with_image("postgres:16"):
-            self.assertEqual(docker_mod.image_name("c1"), "postgres")
-
-    def test_registry_host_is_dropped(self) -> None:
-        with _with_image("svc-db-mariadb-swarm-mgr-01:5000/postgres_custom:17-3.5"):
-            self.assertEqual(docker_mod.image_name("c1"), "postgres_custom")
-
-    def test_pull_through_path_is_kept(self) -> None:
-        with _with_image(
-            "svc-db-mariadb-swarm-mgr-01:5000/ghcr.io/x/mirror/docker.io/postgres:16"
-        ):
-            self.assertEqual(
-                docker_mod.image_name("c1"), "ghcr.io/x/mirror/docker.io/postgres"
-            )
-
-    def test_digest_is_dropped(self) -> None:
-        with _with_image("registry:5000/postgres@sha256:" + "0" * 64):
-            self.assertEqual(docker_mod.image_name("c1"), "postgres")
-
-
-class TestHasImage(unittest.TestCase):
-    def test_registry_hostname_does_not_decide_the_engine(self) -> None:
-        with _with_image("svc-db-mariadb-swarm-mgr-01:5000/postgres_custom:17-3.5"):
-            self.assertFalse(docker_mod.has_image("c1", "mariadb"))
-        with _with_image("svc-db-mariadb-swarm-mgr-01:5000/postgres_custom:17-3.5"):
-            self.assertTrue(docker_mod.has_image("c1", "postgres"))
-
-    def test_tag_does_not_decide_the_engine(self) -> None:
-        with _with_image("registry:5000/xwiki_custom:lts-postgres-tomcat"):
-            self.assertFalse(docker_mod.has_image("c1", "postgres"))
-
-    def test_mirrored_mariadb_still_matches(self) -> None:
-        with _with_image("registry:5000/ghcr.io/x/mirror/docker.io/mariadb:11"):
-            self.assertTrue(docker_mod.has_image("c1", "mariadb"))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/backup/test_docker_tool_probe.py

@@ -0,0 +1,44 @@
+import unittest
+from unittest.mock import patch
+
+from baudolo.backup import docker as docker_mod
+from baudolo.backup.shell import BackupException
+
+
+class TestImageId(unittest.TestCase):
+    def test_the_id_is_returned_without_surrounding_whitespace(self) -> None:
+        with patch.object(
+            docker_mod, "execute_shell_command", return_value=["sha256:abc \n"]
+        ):
+            self.assertEqual(docker_mod.image_id("c1"), "sha256:abc")
+
+
+class TestHasTool(unittest.TestCase):
+    def test_a_tool_that_runs_is_present(self) -> None:
+        with patch.object(docker_mod, "execute_shell_command", return_value=[]):
+            self.assertTrue(docker_mod.has_tool("c1", "pg_dumpall"))
+
+    def test_a_tool_that_exits_non_zero_is_absent(self) -> None:
+        with patch.object(
+            docker_mod, "execute_shell_command", side_effect=BackupException("127")
+        ):
+            self.assertFalse(docker_mod.has_tool("c1", "mariadb-dump"))
+
+    def test_the_probe_needs_no_shell_in_the_image(self) -> None:
+        """A distroless database ships no shell; `sh -c` would deny every tool."""
+        captured = []
+
+        def _capture(cmd):
+            captured.append(cmd)
+            return []
+
+        with patch.object(docker_mod, "execute_shell_command", side_effect=_capture):
+            docker_mod.has_tool("c1", "pg_dumpall")
+
+        self.assertEqual(
+            captured, [["docker", "exec", "c1", "pg_dumpall", "--version"]]
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_dumps_engine_detection.py

@@ -0,0 +1,138 @@
+import unittest
+from unittest.mock import patch
+
+import pandas
+
+from baudolo.backup import dumps as dumps_mod
+
+
+def _df(rows):
+    return pandas.DataFrame(
+        rows, columns=["instance", "database", "username", "password"]
+    )
+
+
+class _Probe:
+    def __init__(self, available, image="sha256:aaa"):
+        self.available = set(available)
+        self.image = image
+        self.calls = []
+
+    def has_tool(self, container, tool):
+        self.calls.append((container, tool))
+        return tool in self.available
+
+    def image_id(self, container):
+        return self.image if isinstance(self.image, str) else self.image[container]
+
+
+def _detect(probe, container="c1"):
+    dumps_mod._ENGINE_BY_IMAGE.clear()
+    with (
+        patch.object(dumps_mod, "has_tool", probe.has_tool),
+        patch.object(dumps_mod, "image_id", probe.image_id),
+    ):
+        return dumps_mod.container_engine(container)
+
+
+class TestContainerEngine(unittest.TestCase):
+    def test_a_postgres_is_found_by_its_dump_tool(self):
+        self.assertEqual(_detect(_Probe(["pg_dumpall"])), ("postgres", "pg_dumpall"))
+
+    def test_a_mariadb_is_found_by_its_dump_tool(self):
+        self.assertEqual(_detect(_Probe(["mariadb-dump"])), ("mariadb", "mariadb-dump"))
+
+    def test_an_image_with_only_mysqldump_is_dumped_with_mysqldump(self):
+        self.assertEqual(_detect(_Probe(["mysqldump"])), ("mariadb", "mysqldump"))
+
+    def test_a_container_without_either_tool_is_no_database(self):
+        self.assertIsNone(_detect(_Probe([])))
+
+    def test_the_probe_stops_at_the_first_tool_it_finds(self):
+        probe = _Probe(["pg_dumpall", "mariadb-dump"])
+        _detect(probe)
+        self.assertEqual(probe.calls, [("c1", "pg_dumpall")])
+
+    def test_the_image_name_does_not_decide_the_engine(self):
+        """The trap the old substring test fell into, from both directions."""
+        probe = _Probe(["pg_dumpall"], image="svc-db-mariadb-mgr-01:5000/pg_custom")
+        self.assertEqual(_detect(probe), ("postgres", "pg_dumpall"))
+
+        probe = _Probe(["mariadb-dump"], image="discourse-database:17")
+        self.assertEqual(_detect(probe), ("mariadb", "mariadb-dump"))
+
+
+class TestProbeCache(unittest.TestCase):
+    def test_replicas_of_one_image_are_probed_once(self):
+        probe = _Probe(["pg_dumpall"])
+        dumps_mod._ENGINE_BY_IMAGE.clear()
+        with (
+            patch.object(dumps_mod, "has_tool", probe.has_tool),
+            patch.object(dumps_mod, "image_id", probe.image_id),
+        ):
+            first = dumps_mod.container_engine("replica-1")
+            second = dumps_mod.container_engine("replica-2")
+        self.assertEqual(first, second)
+        self.assertEqual(len(probe.calls), 1)
+
+    def test_a_second_image_is_probed_separately(self):
+        probe = _Probe(["pg_dumpall"], image={"pg": "sha256:aaa", "app": "sha256:bbb"})
+        dumps_mod._ENGINE_BY_IMAGE.clear()
+        with (
+            patch.object(dumps_mod, "has_tool", probe.has_tool),
+            patch.object(dumps_mod, "image_id", probe.image_id),
+        ):
+            self.assertEqual(
+                dumps_mod.container_engine("pg"), ("postgres", "pg_dumpall")
+            )
+            probe.available = set()
+            self.assertIsNone(dumps_mod.container_engine("app"))
+
+
+class TestBackupDispatch(unittest.TestCase):
+    def test_the_probed_tool_reaches_the_dump(self):
+        probe = _Probe(["mysqldump"])
+        seen = {}
+
+        def _fake_backup_database(**kwargs):
+            seen.update(kwargs)
+            return True
+
+        dumps_mod._ENGINE_BY_IMAGE.clear()
+        with (
+            patch.object(dumps_mod, "has_tool", probe.has_tool),
+            patch.object(dumps_mod, "image_id", probe.image_id),
+            patch.object(dumps_mod, "backup_database", _fake_backup_database),
+        ):
+            is_db, dumped = dumps_mod.backup_mariadb_or_postgres(
+                container="c1",
+                volume_dir="/tmp",
+                databases_df=_df([("c1", "appdb", "u", "p")]),
+                database_containers=["c1"],
+            )
+
+        self.assertTrue(is_db)
+        self.assertTrue(dumped)
+        self.assertEqual(seen["db_type"], "mariadb")
+        self.assertEqual(seen["dump_tool"], "mysqldump")
+
+    def test_a_non_database_container_is_left_to_the_file_backup(self):
+        probe = _Probe([])
+        dumps_mod._ENGINE_BY_IMAGE.clear()
+        with (
+            patch.object(dumps_mod, "has_tool", probe.has_tool),
+            patch.object(dumps_mod, "image_id", probe.image_id),
+        ):
+            self.assertEqual(
+                dumps_mod.backup_mariadb_or_postgres(
+                    container="c1",
+                    volume_dir="/tmp",
+                    databases_df=_df([]),
+                    database_containers=[],
+                ),
+                (False, False),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_snapshot.py

@@ -10,13 +10,13 @@ from baudolo.backup.snapshot import SnapshotError, volume_snapshot
 
 class Runner:
     def __init__(self, replies: dict[str, list[str]] | None = None) -> None:
-        self.calls: list[str] = []
+        self.calls: list[list[str]] = []
         self.replies = replies or {}
 
-    def __call__(self, command: str) -> list[str]:
-        self.calls.append(command)
+    def __call__(self, command: list[str]) -> list[str]:
+        self.calls.append(list(command))
         for prefix, reply in self.replies.items():
-            if command.startswith(prefix):
+            if " ".join(command).startswith(prefix):
                 return reply
         return []
 
@@ -28,7 +28,14 @@ class TestBtrfs(unittest.TestCase):
             pass
         self.assertEqual(
             run.calls[0],
-            "btrfs subvolume snapshot -r /var/lib/docker /var/lib/docker/.baudolo-20260731",
+            [
+                "btrfs",
+                "subvolume",
+                "snapshot",
+                "-r",
+                "/var/lib/docker",
+                "/var/lib/docker/.baudolo-20260731",
+            ],
         )
 
     def test_it_removes_the_snapshot_afterwards(self) -> None:
@@ -36,7 +43,8 @@ class TestBtrfs(unittest.TestCase):
         with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run):
             pass
         self.assertEqual(
-            run.calls[-1], "btrfs subvolume delete /var/lib/docker/.baudolo-20260731"
+            run.calls[-1],
+            ["btrfs", "subvolume", "delete", "/var/lib/docker/.baudolo-20260731"],
         )
 
     def test_it_maps_a_volume_path_into_the_snapshot(self) -> None:
@@ -66,7 +74,7 @@ class TestBtrfs(unittest.TestCase):
             volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run),
         ):
             raise ZeroDivisionError
-        self.assertTrue(run.calls[-1].startswith("btrfs subvolume delete"))
+        self.assertEqual(run.calls[-1][:3], ["btrfs", "subvolume", "delete"])
 
 
 class TestZfs(unittest.TestCase):
@@ -77,13 +85,15 @@ class TestZfs(unittest.TestCase):
         run = self._run()
         with volume_snapshot("zfs", "/var/lib/docker", "20260731", run=run):
             pass
-        self.assertIn("zfs snapshot tank/docker@baudolo-20260731", run.calls)
+        self.assertIn(["zfs", "snapshot", "tank/docker@baudolo-20260731"], run.calls)
 
     def test_it_destroys_the_snapshot_afterwards(self) -> None:
         run = self._run()
         with volume_snapshot("zfs", "/var/lib/docker", "20260731", run=run):
             pass
-        self.assertEqual(run.calls[-1], "zfs destroy tank/docker@baudolo-20260731")
+        self.assertEqual(
+            run.calls[-1], ["zfs", "destroy", "tank/docker@baudolo-20260731"]
+        )
 
     def test_it_maps_a_volume_path_through_the_dot_zfs_directory(self) -> None:
         run = self._run()
@@ -131,8 +141,8 @@ class TestRejections(unittest.TestCase):
 
 
 class Busy(Runner):
-    def __call__(self, command: str) -> list[str]:
-        if command.startswith("btrfs subvolume delete"):
+    def __call__(self, command: list[str]) -> list[str]:
+        if command[:3] == ["btrfs", "subvolume", "delete"]:
             raise BackupException("target is busy")
         return super().__call__(command)
 

tests/unit/backup/test_volume.py

@@ -48,7 +48,7 @@ class TestBackupVolume(unittest.TestCase):
 
     def test_it_keeps_no_twin_of_what_the_second_pass_replaces(self) -> None:
         command = self.copy(authoritative=True)
-        self.assertIn("rsync -aP ", command)
+        self.assertEqual(command[:2], ["rsync", "-aP"])
         self.assertNotIn("--backup", command)
 
     def test_it_creates_the_destination(self) -> None:

tests/unit/restore/test_cli_version_flag.py

@@ -18,6 +18,8 @@ class TestVersionFlagReachesEveryEngine(unittest.TestCase):
             "app_vol",
             "hash",
             "20260817000000",
+            "--repo-name",
+            "repo",
             "--container",
             "db",
             "--db-password",

tests/unit/seed/test_main.py

@@ -17,20 +17,15 @@ class TestSeedMain(unittest.TestCase):
             columns=["instance", "database", "username", "password"]
         )
 
-    def test_validate_database_value_rejects_empty(self) -> None:
+    def test_a_rejected_database_never_reaches_the_file(self) -> None:
         with self.assertRaises(ValueError):
-            seed_main._validate_database_value("", instance="x")
-
-    def test_validate_database_value_accepts_star(self) -> None:
-        self.assertEqual(seed_main._validate_database_value("*", instance="x"), "*")
-
-    def test_validate_database_value_rejects_nan(self) -> None:
-        with self.assertRaises(ValueError):
-            seed_main._validate_database_value("nan", instance="x")
-
-    def test_validate_database_value_rejects_invalid_name(self) -> None:
-        with self.assertRaises(ValueError):
-            seed_main._validate_database_value("bad name", instance="x")
+            seed_main.check_and_add_entry(
+                file_path="/nonexistent/databases.csv",
+                instance="x",
+                database="bad name",
+                username="u",
+                password="p",
+            )
 
     def _mock_df_mask_any(self, *, any_value: bool) -> MagicMock:
         """

tests/unit/test_databases.py

@@ -0,0 +1,84 @@
+"""Contract of databases.csv: the seed writes it, the backup and a restore read it."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from baudolo.databases import (
+    CLUSTER_ROW,
+    COLUMNS,
+    DELIMITER,
+    DatabasesCsvError,
+    Row,
+    read_rows,
+    validate_database,
+)
+
+HEADER = DELIMITER.join(COLUMNS)
+
+
+def _csv(*lines: str) -> str:
+    path = Path(tempfile.mkdtemp()) / "databases.csv"
+    path.write_text("\n".join((HEADER, *lines)) + "\n", encoding="utf-8")
+    return str(path)
+
+
+class TestValidateDatabase(unittest.TestCase):
+    def test_a_concrete_name_passes(self) -> None:
+        self.assertEqual(validate_database("app_db-1", instance="x"), "app_db-1")
+
+    def test_the_cluster_marker_passes(self) -> None:
+        self.assertEqual(validate_database(CLUSTER_ROW, instance="x"), CLUSTER_ROW)
+
+    def test_an_empty_column_is_rejected(self) -> None:
+        with self.assertRaises(DatabasesCsvError):
+            validate_database("", instance="x")
+
+    def test_the_string_nan_is_rejected(self) -> None:
+        """pandas used to hand back NaN, which wrote a nan.backup.sql."""
+        with self.assertRaises(DatabasesCsvError):
+            validate_database("nan", instance="x")
+
+    def test_a_name_that_could_reach_a_shell_is_rejected(self) -> None:
+        for hostile in ("bad name", "a;rm -rf /", "$(id)", "a`id`", "a/b"):
+            with self.subTest(name=hostile), self.assertRaises(DatabasesCsvError):
+                validate_database(hostile, instance="x")
+
+    def test_the_error_is_a_value_error(self) -> None:
+        """Callers predating the shared module catch ValueError."""
+        with self.assertRaises(ValueError):
+            validate_database("", instance="x")
+
+
+class TestReadRows(unittest.TestCase):
+    def test_the_header_is_skipped(self) -> None:
+        rows = read_rows(_csv(f"pg{DELIMITER}app{DELIMITER}u{DELIMITER}p"))
+        self.assertEqual(rows, [Row("pg", "app", "u", "p")])
+
+    def test_a_blank_row_is_dropped(self) -> None:
+        rows = read_rows(_csv("", f"pg{DELIMITER}app{DELIMITER}u{DELIMITER}p", ""))
+        self.assertEqual(len(rows), 1)
+
+    def test_a_short_row_is_refused(self) -> None:
+        with self.assertRaises(DatabasesCsvError):
+            read_rows(_csv(f"pg{DELIMITER}app{DELIMITER}u"))
+
+    def test_values_arrive_verbatim(self) -> None:
+        """A password may legitimately begin or end with a space."""
+        rows = read_rows(_csv(f"pg{DELIMITER}app{DELIMITER}u{DELIMITER} pw "))
+        self.assertEqual(rows[0].password, " pw ")
+
+    def test_a_cluster_row_knows_itself(self) -> None:
+        rows = read_rows(
+            _csv(
+                f"pg{DELIMITER}{CLUSTER_ROW}{DELIMITER}postgres{DELIMITER}p",
+                f"pg{DELIMITER}app{DELIMITER}u{DELIMITER}p",
+            )
+        )
+        self.assertEqual([row.is_cluster for row in rows], [True, False])
+
+
+if __name__ == "__main__":
+    unittest.main()