Release version 1.5.0

- Handle missing or empty databases.csv gracefully with warnings and empty DataFrame
- Add unit tests for robust databases.csv loading behavior
- Adjust seed tests to assert warnings across multiple print calls
- Replace Debian docker.io with docker-ce-cli to avoid Docker API version mismatch
- Install required build tools (curl, gnupg) for Docker repo setup

https://chatgpt.com/share/697e6d9d-6458-800f-9d12-1e337509be4e

.gitignore

@@ -2,3 +2,4 @@ __pycache__
 artifacts/
 *.egg-info
 dist/
+build/

CHANGELOG.md

@@ -1,6 +1,32 @@
+## [1.5.0] - 2026-01-31
+
+* * Make `databases.csv` optional: missing or empty files now emit warnings and no longer break backups
+* Fix Docker CLI compatibility by switching to `docker-ce-cli` and required build tools
+
+
+## [1.4.0] - 2026-01-31
+
+* Baudolo now restarts Docker Compose stacks in a wrapper-aware way (with a `docker compose` fallback), ensuring that all Compose overrides and env files are applied identically to the Infinito.Nexus workflow.
+
+
+## [1.3.0] - 2026-01-10
+
+* Empty databases.csv no longer causes baudolo-seed to fail
+
+
+## [1.2.0] - 2025-12-29
+
+* * Introduced **`--dump-only-sql`** mode for reliable, SQL-only database backups (replaces `--dump-only`).
+* Database configuration in `databases.csv` is now **strict and explicit** (`*` or concrete database name only).
+* **PostgreSQL cluster backups** are supported via `*`.
+* SQL dumps are written **atomically** to avoid corrupted or empty files.
+* Backups are **smarter and faster**: ignored volumes are skipped early, file backups run only when needed.
+* Improved reliability through expanded end-to-end tests and safer defaults.
+
+
 ## [1.1.1] - 2025-12-28
 
-* * **Backup:** In ***--dump-only*** mode, fall back to file backups with a warning when no database dump can be produced (e.g. missing `databases.csv` entry).
+* * **Backup:** In ***--dump-only-sql*** mode, fall back to file backups with a warning when no database dump can be produced (e.g. missing `databases.csv` entry).
 
 
 ## [1.1.0] - 2025-12-28

Dockerfile

@@ -3,32 +3,35 @@ FROM python:3.11-slim
 
 WORKDIR /app
 
-# Runtime + build essentials:
-# - rsync: required for file backup/restore
-# - ca-certificates: TLS
-# - docker-cli: needed if you want to control the host Docker engine (via /var/run/docker.sock mount)
-# - make: to delegate install logic to Makefile
-#
-# Notes:
-# - On Debian slim, the docker client package is typically "docker.io".
-# - If you only want restore-without-docker, you can drop docker.io later.
+# Base deps for build/runtime + docker repo key
 RUN apt-get update && apt-get install -y --no-install-recommends \
     make \
     rsync \
     ca-certificates \
-    docker-cli \
+    bash \
+    curl \
+    gnupg \
   && rm -rf /var/lib/apt/lists/*
 
+# Install Docker CLI (docker-ce-cli) from Docker's official apt repo
+RUN bash -lc "set -euo pipefail \
+  && install -m 0755 -d /etc/apt/keyrings \
+  && curl -fsSL https://download.docker.com/linux/debian/gpg \
+     | gpg --dearmor -o /etc/apt/keyrings/docker.gpg \
+  && chmod a+r /etc/apt/keyrings/docker.gpg \
+  && . /etc/os-release \
+  && echo \"deb [arch=\$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/debian \${VERSION_CODENAME} stable\" \
+     > /etc/apt/sources.list.d/docker.list \
+  && apt-get update \
+  && apt-get install -y --no-install-recommends docker-ce-cli \
+  && rm -rf /var/lib/apt/lists/*"
+
 # Fail fast if docker client is missing
+RUN docker version || true
 RUN command -v docker
 
 COPY . .
-
-# All install decisions are handled by the Makefile.
 RUN make install
 
-# Sensible defaults (can be overridden at runtime)
 ENV PYTHONUNBUFFERED=1
-
-# Default: show CLI help
 CMD ["baudolo", "--help"]

Makefile

@@ -49,9 +49,9 @@ test: test-unit test-integration test-e2e
 test-unit: clean build
 	@echo ">> Running unit tests"
 	@docker run --rm -t $(IMAGE) \
-	  sh -lc 'python -m unittest discover -t . -s tests/unit -p "test_*.py" -v'
+	  bash -lc 'python -m unittest discover -t . -s tests/unit -p "test_*.py" -v'
 
 test-integration: clean build
 	@echo ">> Running integration tests"
 	@docker run --rm -t $(IMAGE) \
-	  sh -lc 'python -m unittest discover -t . -s tests/integration -p "test_*.py" -v'
+	  bash -lc 'python -m unittest discover -t . -s tests/integration -p "test_*.py" -v'

README.md

@@ -134,7 +134,7 @@ baudolo \
 | Flag            | Description                                 |
 | --------------- | ------------------------------------------- |
 | `--everything`  | Always stop containers and re-run rsync     |
-| `--dump-only`   | Only create SQL dumps, skip file backups    |
+| `--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.    |
 | `--shutdown`    | Do not restart containers after backup      |
 | `--backups-dir` | Backup root directory (default: `/Backups`) |
 | `--repo-name`   | Backup namespace under machine hash         |

pyproject.toml

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

scripts/test-e2e.sh

@@ -83,7 +83,7 @@ dump_debug() {
   docker -H "${DIND_HOST}" run --rm \
     -v "${E2E_TMP_VOL}:/tmp" \
     alpine:3.20 \
-    sh -lc 'cd /tmp && tar -czf /out.tar.gz . || true' \
+    bash -lc 'cd /tmp && tar -czf /out.tar.gz . || true' \
     >/dev/null 2>&1 || true
 
   # The above writes inside the container FS, not to host. So do it properly:
@@ -91,7 +91,7 @@ dump_debug() {
   local tmpc="baudolo-e2e-tmpdump-${TS}"
   docker -H "${DIND_HOST}" rm -f "${tmpc}" >/dev/null 2>&1 || true
   docker -H "${DIND_HOST}" create --name "${tmpc}" -v "${E2E_TMP_VOL}:/tmp" alpine:3.20 \
-    sh -lc 'cd /tmp && tar -czf /tmpdump.tar.gz . || true' >/dev/null
+    bash -lc 'cd /tmp && tar -czf /tmpdump.tar.gz . || true' >/dev/null
   docker -H "${DIND_HOST}" start -a "${tmpc}" >/dev/null 2>&1 || true
   docker -H "${DIND_HOST}" cp "${tmpc}:/tmpdump.tar.gz" "${ARTIFACTS_DIR}/e2e-tmp-${TS}.tar.gz" >/dev/null 2>&1 || true
   docker -H "${DIND_HOST}" rm -f "${tmpc}" >/dev/null 2>&1 || true
@@ -187,7 +187,7 @@ if [ "${DEBUG_SHELL}" = "1" ]; then
     -v "${DIND_VOL}:/var/lib/docker:ro" \
     -v "${E2E_TMP_VOL}:/tmp" \
     "${IMG}" \
-    sh -lc '
+    bash -lc '
       set -e
       if [ ! -f /etc/machine-id ]; then
         mkdir -p /etc
@@ -195,7 +195,7 @@ if [ "${DEBUG_SHELL}" = "1" ]; then
       fi
       echo ">> DOCKER_HOST=${DOCKER_HOST}"
       docker ps -a || true
-      exec sh
+      exec bash
     '
   rc=$?
 else
@@ -206,7 +206,7 @@ else
     -v "${DIND_VOL}:/var/lib/docker:ro" \
     -v "${E2E_TMP_VOL}:/tmp" \
     "${IMG}" \
-    sh -lc '
+    bash -lc '
     set -euo pipefail
     set -x
     export PYTHONUNBUFFERED=1

src/baudolo/backup/app.py

@@ -2,10 +2,12 @@ from __future__ import annotations
 
 import os
 import pathlib
+import sys
 from datetime import datetime
 
 import pandas
 from dirval import create_stamp_file
+from pandas.errors import EmptyDataError
 
 from .cli import parse_args
 from .compose import handle_docker_compose_services
@@ -72,6 +74,7 @@ def requires_stop(containers: list[str], images_no_stop_required: list[str]) ->
             return True
     return False
 
+
 def backup_mariadb_or_postgres(
     *,
     container: str,
@@ -95,6 +98,42 @@ def backup_mariadb_or_postgres(
     return False, False
 
 
+def _empty_databases_df() -> "pandas.DataFrame":
+    """
+    Create an empty DataFrame with the expected schema for databases.csv.
+
+    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"])
+
+
+def _load_databases_df(csv_path: str) -> "pandas.DataFrame":
+    """
+    Load databases.csv robustly.
+
+    - Missing file     -> warn, continue with empty df
+    - Empty file       -> warn, continue with empty df
+    - Valid CSV        -> return dataframe
+    """
+    try:
+        return pandas.read_csv(csv_path, sep=";", keep_default_na=False, dtype=str)
+    except FileNotFoundError:
+        print(
+            f"WARNING: databases.csv not found: {csv_path}. Continuing without database dumps.",
+            file=sys.stderr,
+            flush=True,
+        )
+        return _empty_databases_df()
+    except EmptyDataError:
+        print(
+            f"WARNING: databases.csv exists but is empty: {csv_path}. Continuing without database dumps.",
+            file=sys.stderr,
+            flush=True,
+        )
+        return _empty_databases_df()
+
+
 def _backup_dumps_for_volume(
     *,
     containers: list[str],
@@ -132,7 +171,13 @@ 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 = pandas.read_csv(args.databases_csv, sep=";")
+    # 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)
 
@@ -140,6 +185,14 @@ def main() -> int:
         print(f"Start backup routine for volume: {volume_name}", flush=True)
         containers = containers_using_volume(volume_name)
 
+        # EARLY SKIP: if all linked containers are ignored, do not create any dirs
+        if volume_is_fully_ignored(containers, args.images_no_backup_required):
+            print(
+                f"Skipping volume '{volume_name}' entirely (all linked containers are ignored).",
+                flush=True,
+            )
+            continue
+
         vol_dir = create_volume_directory(version_dir, volume_name)
 
         found_db, dumped_any = _backup_dumps_for_volume(
@@ -149,26 +202,20 @@ def main() -> int:
             database_containers=args.database_containers,
         )
 
-        # dump-only logic:
-        if args.dump_only:
-            if found_db and not dumped_any:
-                print(
-                    f"WARNING: dump-only requested but no DB dump was produced for DB volume '{volume_name}'. Falling back to file backup.",
-                    flush=True,
-                )
-                # continue to file backup below
-            else:
-                # keep old behavior: skip file backups
-                continue
-
-
-        # skip file backup if all linked containers are ignored
-        if volume_is_fully_ignored(containers, args.images_no_backup_required):
-            print(
-                f"Skipping file backup for volume '{volume_name}' (all linked containers are ignored).",
-                flush=True,
-            )
-            continue
+        # dump-only-sql logic:
+        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,
+                    )
+                    # fall through to file backup below
+                else:
+                    # DB volume successfully dumped -> skip file backup
+                    continue
+            # Non-DB volume -> always do file backup (fall through)
 
         if args.everything:
             # "everything": always do pre-rsync, then stop + rsync again

src/baudolo/backup/cli.py

@@ -68,10 +68,15 @@ def parse_args() -> argparse.Namespace:
         action="store_true",
         help="Do not restart containers after backup",
     )
-    p.add_argument(
-        "--dump-only",
-        action="store_true",
-        help="Only create DB dumps (skip ALL file rsync backups)",
-    )
 
+    p.add_argument(
+        "--dump-only-sql",
+        action="store_true",
+        help=(
+            "Create database dumps only for DB volumes. "
+            "File backups are skipped for DB volumes if a dump succeeds, "
+            "but non-DB volumes are still backed up. "
+            "If a DB dump cannot be produced, baudolo falls back to a file backup."
+        ),
+    )
     return p.parse_args()

src/baudolo/backup/compose.py

@@ -1,13 +1,104 @@
 from __future__ import annotations
 
 import os
+import shutil
 import subprocess
+from pathlib import Path
+from typing import List, Optional
+
+
+def _detect_env_file(project_dir: Path) -> Optional[Path]:
+    """
+    Detect Compose env file in a directory.
+    Preference (same as Infinito.Nexus wrapper):
+      1) <dir>/.env (file)
+      2) <dir>/.env/env (file)  (legacy layout)
+    """
+    c1 = project_dir / ".env"
+    if c1.is_file():
+        return c1
+
+    c2 = project_dir / ".env" / "env"
+    if c2.is_file():
+        return c2
+
+    return None
+
+
+def _detect_compose_files(project_dir: Path) -> List[Path]:
+    """
+    Detect Compose file stack in a directory (same as Infinito.Nexus wrapper).
+    Always requires docker-compose.yml.
+    Optionals:
+      - docker-compose.override.yml
+      - docker-compose.ca.override.yml
+    """
+    base = project_dir / "docker-compose.yml"
+    if not base.is_file():
+        raise FileNotFoundError(f"Missing docker-compose.yml in: {project_dir}")
+
+    files = [base]
+
+    override = project_dir / "docker-compose.override.yml"
+    if override.is_file():
+        files.append(override)
+
+    ca_override = project_dir / "docker-compose.ca.override.yml"
+    if ca_override.is_file():
+        files.append(ca_override)
+
+    return files
+
+
+def _compose_wrapper_path() -> Optional[str]:
+    """
+    Prefer the Infinito.Nexus compose wrapper if present.
+    Equivalent to: `which compose`
+    """
+    return shutil.which("compose")
+
+
+def _build_compose_cmd(project_dir: str, passthrough: List[str]) -> List[str]:
+    """
+    Build the compose command for this project directory.
+
+    Behavior:
+    - If `compose` wrapper exists: use it with --chdir (so it resolves -f/--env-file itself)
+    - Else: use `docker compose` and replicate wrapper's file/env detection.
+    """
+    pdir = Path(project_dir).resolve()
+
+    wrapper = _compose_wrapper_path()
+    if wrapper:
+        # Wrapper defaults project name to basename of --chdir.
+        # "--" ensures wrapper stops parsing its own args.
+        return [wrapper, "--chdir", str(pdir), "--", *passthrough]
+
+    # Fallback: pure docker compose, but mirror wrapper behavior.
+    files = _detect_compose_files(pdir)
+    env_file = _detect_env_file(pdir)
+
+    cmd: List[str] = ["docker", "compose"]
+    for f in files:
+        cmd += ["-f", str(f)]
+    if env_file:
+        cmd += ["--env-file", str(env_file)]
+
+    cmd += passthrough
+    return cmd
 
 
 def hard_restart_docker_services(dir_path: str) -> None:
-    print(f"Hard restart docker-compose services in: {dir_path}", flush=True)
-    subprocess.run(["docker-compose", "down"], cwd=dir_path, check=True)
-    subprocess.run(["docker-compose", "up", "-d"], cwd=dir_path, check=True)
+    print(f"Hard restart compose services in: {dir_path}", flush=True)
+
+    down_cmd = _build_compose_cmd(dir_path, ["down"])
+    up_cmd = _build_compose_cmd(dir_path, ["up", "-d"])
+
+    print(">>> " + " ".join(down_cmd), flush=True)
+    subprocess.run(down_cmd, check=True)
+
+    print(">>> " + " ".join(up_cmd), flush=True)
+    subprocess.run(up_cmd, check=True)
 
 
 def handle_docker_compose_services(

src/baudolo/backup/db.py

@@ -4,6 +4,8 @@ import os
 import pathlib
 import re
 import logging
+from typing import Optional
+
 import pandas
 
 from .shell import BackupException, execute_shell_command
@@ -12,17 +14,55 @@ log = logging.getLogger(__name__)
 
 
 def get_instance(container: str, database_containers: list[str]) -> str:
+    """
+    Derive a stable instance name from the container name.
+    """
     if container in database_containers:
         return container
     return re.split(r"(_|-)(database|db|postgres)", container)[0]
 
 
-def fallback_pg_dumpall(container: str, username: str, password: str, out_file: str) -> None:
+def _validate_database_value(value: Optional[str], *, 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 > {out_file}"
+        f"pg_dumpall -U {username} -h localhost"
     )
-    execute_shell_command(cmd)
+    _atomic_write_cmd(cmd, out_file)
 
 
 def backup_database(
@@ -34,12 +74,15 @@ def backup_database(
     database_containers: list[str],
 ) -> bool:
     """
-    Returns True if at least one dump file was produced, else False.
+    Backup databases for a given DB container.
+
+    Returns True if at least one dump was produced.
     """
     instance_name = get_instance(container, database_containers)
-    entries = databases_df.loc[databases_df["instance"] == instance_name]
+
+    entries = databases_df[databases_df["instance"] == instance_name]
     if entries.empty:
-        log.warning("No entry found for instance '%s' (skipping DB dump)", instance_name)
+        log.debug("No database entries for instance '%s'", instance_name)
         return False
 
     out_dir = os.path.join(volume_dir, "sql")
@@ -48,43 +91,54 @@ def backup_database(
     produced = False
 
     for row in entries.itertuples(index=False):
-        db_name = row.database
-        user = row.username
-        password = row.password
+        raw_db = getattr(row, "database", "")
+        user = (getattr(row, "username", "") or "").strip()
+        password = (getattr(row, "password", "") or "").strip()
 
+        db_value = _validate_database_value(raw_db, instance=instance_name)
+
+        # Explicit: dump ALL databases
+        if db_value == "*":
+            if db_type != "postgres":
+                raise ValueError(
+                    f"databases.csv entry for instance '{instance_name}': "
+                    "'*' is currently only supported for Postgres."
+                )
+
+            cluster_file = os.path.join(out_dir, f"{instance_name}.cluster.backup.sql")
+            fallback_pg_dumpall(container, user, password, cluster_file)
+            produced = True
+            continue
+
+        # Concrete database dump
+        db_name = db_value
         dump_file = os.path.join(out_dir, f"{db_name}.backup.sql")
 
         if db_type == "mariadb":
             cmd = (
                 f"docker exec {container} /usr/bin/mariadb-dump "
-                f"-u {user} -p{password} {db_name} > {dump_file}"
+                f"-u {user} -p{password} {db_name}"
             )
-            execute_shell_command(cmd)
+            _atomic_write_cmd(cmd, dump_file)
             produced = True
             continue
 
         if db_type == "postgres":
-            cluster_file = os.path.join(out_dir, f"{instance_name}.cluster.backup.sql")
-
-            if not db_name:
-                fallback_pg_dumpall(container, user, password, cluster_file)
-                return True
-
             try:
                 cmd = (
                     f"PGPASSWORD={password} docker exec -i {container} "
-                    f"pg_dump -U {user} -d {db_name} -h localhost > {dump_file}"
+                    f"pg_dump -U {user} -d {db_name} -h localhost"
                 )
-                execute_shell_command(cmd)
+                _atomic_write_cmd(cmd, dump_file)
                 produced = True
             except BackupException as e:
-                print(f"pg_dump failed: {e}", flush=True)
-                print(
-                    f"Falling back to pg_dumpall for instance '{instance_name}'",
-                    flush=True,
+                # 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 "
+                    "and therefore must succeed.\n"
+                    f"{e}"
                 )
-                fallback_pg_dumpall(container, user, password, cluster_file)
-                produced = True
             continue
 
     return produced

src/baudolo/seed/__main__.py

@@ -1,67 +1,116 @@
-import pandas as pd
+#!/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 check_and_add_entry(file_path, instance, database, username, password):
-    # Check if the file exists and is not empty
-    if os.path.exists(file_path) and os.path.getsize(file_path) > 0:
-        # Read the existing CSV file with header
-        df = pd.read_csv(file_path, sep=";")
-    else:
-        # Create a new DataFrame with columns if file does not exist
-        df = pd.DataFrame(columns=["instance", "database", "username", "password"])
-
-    # Check if the entry exists and remove it
-    mask = (
-        (df["instance"] == instance)
-        & (
-            (df["database"] == database)
-            | (((df["database"].isna()) | (df["database"] == "")) & (database == ""))
+def _validate_database_value(value: Optional[str], *, 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)."
         )
-        & (df["username"] == username)
-    )
+    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
 
-    if not df[mask].empty:
-        print("Replacing existing entry.")
-        df = df[~mask]
+
+def _empty_df() -> pd.DataFrame:
+    return pd.DataFrame(columns=["instance", "database", "username", "password"])
+
+
+def check_and_add_entry(
+    file_path: str,
+    instance: str,
+    database: Optional[str],
+    username: str,
+    password: str,
+) -> None:
+    """
+    Add or update an entry in databases.csv.
+
+    The function enforces strict validation:
+    - database MUST be set
+    - database MUST be '*' or a valid database name
+    """
+    database = _validate_database_value(database, instance=instance)
+
+    if os.path.exists(file_path):
+        try:
+            df = pd.read_csv(
+                file_path,
+                sep=";",
+                dtype=str,
+                keep_default_na=False,
+            )
+        except EmptyDataError:
+            print(
+                f"WARNING: databases.csv exists but is empty: {file_path}. Creating header columns.",
+                file=sys.stderr,
+            )
+            df = _empty_df()
+    else:
+        df = _empty_df()
+    mask = (df["instance"] == instance) & (df["database"] == database)
+
+    if mask.any():
+        print("Updating existing entry.")
+        df.loc[mask, ["username", "password"]] = [username, password]
     else:
         print("Adding new entry.")
+        new_entry = pd.DataFrame(
+            [[instance, database, username, password]],
+            columns=["instance", "database", "username", "password"],
+        )
+        df = pd.concat([df, new_entry], ignore_index=True)
 
-    # Create a new DataFrame for the new entry
-    new_entry = pd.DataFrame(
-        [
-            {
-                "instance": instance,
-                "database": database,
-                "username": username,
-                "password": password,
-            }
-        ]
-    )
-
-    # Add (or replace) the entry using concat
-    df = pd.concat([df, new_entry], ignore_index=True)
-
-    # Save the updated CSV file
     df.to_csv(file_path, sep=";", index=False)
 
 
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser(
-        description="Check and replace (or add) a database entry in a CSV file."
+        description="Seed or update databases.csv for backup configuration."
     )
-    parser.add_argument("file_path", help="Path to the CSV file")
-    parser.add_argument("instance", help="Database instance")
-    parser.add_argument("database", help="Database name")
-    parser.add_argument("username", help="Username")
-    parser.add_argument("password", nargs="?", default="", help="Password (optional)")
+    parser.add_argument("file", help="Path to databases.csv")
+    parser.add_argument("instance", help="Instance name (e.g. bigbluebutton)")
+    parser.add_argument(
+        "database",
+        help="Database name or '*' to dump all databases",
+    )
+    parser.add_argument("username", help="Database username")
+    parser.add_argument("password", help="Database password")
 
     args = parser.parse_args()
 
-    check_and_add_entry(
-        args.file_path, args.instance, args.database, args.username, args.password
-    )
+    try:
+        check_and_add_entry(
+            file_path=args.file,
+            instance=args.instance,
+            database=args.database,
+            username=args.username,
+            password=args.password,
+        )
+    except Exception as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        sys.exit(1)
 
 
 if __name__ == "__main__":

tests/e2e/helpers.py

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

tests/e2e/test_e2e_cli_contract_dump_only_sql.py

@@ -0,0 +1,29 @@
+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_dump_only_fallback_to_files.py

@@ -19,7 +19,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None:
         require_docker()
-        cls.prefix = unique("baudolo-e2e-dump-only-fallback")
+        cls.prefix = unique("baudolo-e2e-dump-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-fallback-marker"
+        cls.marker = "dump-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 and a DB container present:
+        # Run baudolo with --dump-only-sql and a DB container present:
         # Expected: WARNING + FALLBACK to file backup (files/ must exist)
         cmd = [
             "baudolo",
@@ -94,7 +94,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
             "mariadb",
             "mysql",
             "alpine",
-            "--dump-only",
+            "--dump-only-sql",
         ]
         cp = run(cmd, capture=True, check=True)
 
@@ -127,28 +127,34 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
 
     def test_warns_about_missing_dump_in_dump_only_mode(self) -> None:
         self.assertIn(
-            "WARNING: dump-only requested but no DB dump was produced",
+            "WARNING: dump-only-sql requested but no DB dump was produced",
             self.stdout,
             f"Expected warning in baudolo output. STDOUT:\n{self.stdout}",
         )
 
     def test_files_backup_exists_due_to_fallback(self) -> None:
-        p = backup_path(
-            self.backups_dir,
-            self.repo_name,
-            self.version,
-            self.pg_volume,
-        ) / "files"
+        p = (
+            backup_path(
+                self.backups_dir,
+                self.repo_name,
+                self.version,
+                self.pg_volume,
+            )
+            / "files"
+        )
         self.assertTrue(p.is_dir(), f"Expected files backup dir at: {p}")
 
     def test_sql_dump_not_present(self) -> None:
         # There should be no sql dumps because databases.csv had no matching entry.
-        sql_dir = backup_path(
-            self.backups_dir,
-            self.repo_name,
-            self.version,
-            self.pg_volume,
-        ) / "sql"
+        sql_dir = (
+            backup_path(
+                self.backups_dir,
+                self.repo_name,
+                self.version,
+                self.pg_volume,
+            )
+            / "sql"
+        )
         # Could exist (dir created) in some edge cases, but should contain no *.sql dumps.
         if sql_dir.exists():
             dumps = list(sql_dir.glob("*.sql"))

tests/e2e/test_e2e_dump_only_sql_mixed_run.py

@@ -0,0 +1,184 @@
+import unittest
+
+from .helpers import (
+    backup_path,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    wait_for_postgres,
+    write_databases_csv,
+)
+
+
+class TestE2EDumpOnlySqlMixedRun(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-dump-only-sql-mixed-run")
+        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
+
+        # --- Volumes ---
+        cls.db_volume = f"{cls.prefix}-vol-db"
+        cls.files_volume = f"{cls.prefix}-vol-files"
+
+        # Track for cleanup
+        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])
+
+        # Put a marker into the non-db volume
+        run(
+            [
+                "docker",
+                "run",
+                "--rm",
+                "-v",
+                f"{cls.files_volume}:/data",
+                "alpine:3.20",
+                "sh",
+                "-lc",
+                "echo 'hello-non-db' > /data/hello.txt",
+            ]
+        )
+
+        # --- Start Postgres container using the DB volume ---
+        cls.pg_container = f"{cls.prefix}-pg"
+        cls.containers.append(cls.pg_container)
+
+        cls.pg_password = "postgres"
+        cls.pg_db = "testdb"
+        cls.pg_user = "postgres"
+
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.pg_container,
+                "-e",
+                f"POSTGRES_PASSWORD={cls.pg_password}",
+                "-v",
+                f"{cls.db_volume}:/var/lib/postgresql/data",
+                "postgres:16-alpine",
+            ]
+        )
+        wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
+
+        # Create deterministic content in DB so dump is non-empty
+        run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                f'psql -U postgres -c "CREATE DATABASE {cls.pg_db};" || true',
+            ],
+            check=True,
+        )
+        run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                (
+                    f"psql -U postgres -d {cls.pg_db} -c "
+                    '"CREATE TABLE IF NOT EXISTS t (id INT PRIMARY KEY, v TEXT);'
+                    "INSERT INTO t(id,v) VALUES (1,'hello-db') "
+                    'ON CONFLICT (id) DO UPDATE SET v=EXCLUDED.v;"'
+                ),
+            ],
+            check=True,
+        )
+
+        # databases.csv with an entry => dump should succeed
+        cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
+        write_databases_csv(
+            cls.databases_csv,
+            [(cls.pg_container, cls.pg_db, cls.pg_user, cls.pg_password)],
+        )
+
+        # Run baudolo with dump-only-sql
+        cmd = [
+            "baudolo",
+            "--compose-dir",
+            cls.compose_dir,
+            "--databases-csv",
+            cls.databases_csv,
+            "--database-containers",
+            cls.pg_container,
+            "--images-no-stop-required",
+            "alpine",
+            "postgres",
+            "mariadb",
+            "mysql",
+            "--dump-only-sql",
+            "--backups-dir",
+            cls.backups_dir,
+            "--repo-name",
+            cls.repo_name,
+        ]
+        cp = run(cmd, capture=True, check=True)
+        cls.stdout = cp.stdout
+        cls.stderr = cp.stderr
+
+        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 test_db_volume_has_dump_and_no_files_dir(self) -> None:
+        base = backup_path(
+            self.backups_dir, self.repo_name, self.version, self.db_volume
+        )
+
+        dumps = base / "sql"
+        files = base / "files"
+
+        self.assertTrue(dumps.exists(), f"Expected dumps dir for DB volume at: {dumps}")
+        self.assertFalse(
+            files.exists(),
+            f"Did not expect files dir for DB volume when dump succeeded at: {files}",
+        )
+
+        # Optional: at least one dump file exists
+        dump_files = list(dumps.glob("*.sql")) + list(dumps.glob("*.sql.gz"))
+        self.assertTrue(
+            dump_files,
+            f"Expected at least one SQL dump file in {dumps}, found none.",
+        )
+
+    def test_non_db_volume_has_files_dir(self) -> None:
+        base = backup_path(
+            self.backups_dir, self.repo_name, self.version, self.files_volume
+        )
+        files = base / "files"
+        self.assertTrue(
+            files.exists(),
+            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
+        base = backup_path(
+            self.backups_dir, self.repo_name, self.version, self.files_volume
+        )
+        self.assertTrue(
+            (base / "files").exists(),
+            f"Expected non-DB volume files backup to exist at: {base / 'files'}",
+        )

tests/e2e/test_e2e_files_no_copy.py

@@ -26,10 +26,10 @@ class TestE2EFilesNoCopy(unittest.TestCase):
         cls.repo_name = cls.prefix
 
         cls.volume_src = f"{cls.prefix}-vol-src"
-        cls.volume_dst = f"{cls.prefix}-vol-dst"
-        cls.containers = []
-        cls.volumes = [cls.volume_src, cls.volume_dst]
+        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(
             [
@@ -48,7 +48,7 @@ class TestE2EFilesNoCopy(unittest.TestCase):
         cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
         write_databases_csv(cls.databases_csv, [])
 
-        # dump-only => NO file rsync backups
+        # dump-only-sql => non-DB volumes are STILL backed up as files
         backup_run(
             backups_dir=cls.backups_dir,
             repo_name=cls.repo_name,
@@ -56,28 +56,32 @@ class TestE2EFilesNoCopy(unittest.TestCase):
             databases_csv=cls.databases_csv,
             database_containers=["dummy-db"],
             images_no_stop_required=["alpine", "postgres", "mariadb", "mysql"],
-            dump_only=True,
+            dump_only_sql=True,
         )
 
         cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
 
+        # Wipe the volume to ensure restore actually restores something
+        run(["docker", "volume", "rm", "-f", cls.volume_src])
+        run(["docker", "volume", "create", cls.volume_src])
+
     @classmethod
     def tearDownClass(cls) -> None:
         cleanup_docker(containers=cls.containers, volumes=cls.volumes)
 
-    def test_files_backup_not_present(self) -> None:
+    def test_files_backup_present_for_non_db_volume(self) -> None:
         p = (
             backup_path(self.backups_dir, self.repo_name, self.version, self.volume_src)
             / "files"
         )
-        self.assertFalse(p.exists(), f"Did not expect files backup dir at: {p}")
+        self.assertTrue(p.exists(), f"Expected files backup dir at: {p}")
 
-    def test_restore_files_fails_expected(self) -> None:
+    def test_restore_files_succeeds_and_restores_content(self) -> None:
         p = run(
             [
                 "baudolo-restore",
                 "files",
-                self.volume_dst,
+                self.volume_src,
                 self.hash,
                 self.version,
                 "--backups-dir",
@@ -89,6 +93,27 @@ class TestE2EFilesNoCopy(unittest.TestCase):
         )
         self.assertEqual(
             p.returncode,
-            2,
-            f"Expected exitcode 2, got {p.returncode}\nSTDOUT={p.stdout}\nSTDERR={p.stderr}",
+            0,
+            f"Expected exitcode 0, got {p.returncode}\nSTDOUT={p.stdout}\nSTDERR={p.stderr}",
+        )
+
+        cp = run(
+            [
+                "docker",
+                "run",
+                "--rm",
+                "-v",
+                f"{self.volume_src}:/data",
+                "alpine:3.20",
+                "sh",
+                "-lc",
+                "cat /data/hello.txt",
+            ],
+            capture=True,
+            check=True,
+        )
+        self.assertEqual(
+            cp.stdout.strip(),
+            "hello",
+            f"Unexpected restored content. STDOUT={cp.stdout}\nSTDERR={cp.stderr}",
         )

tests/e2e/test_e2e_images_no_backup_required_early_skip.py

@@ -0,0 +1,131 @@
+# tests/e2e/test_e2e_images_no_backup_required_early_skip.py
+import unittest
+
+from .helpers import (
+    backup_path,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    write_databases_csv,
+)
+
+
+class TestE2EImagesNoBackupRequiredEarlySkip(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+
+        cls.prefix = unique("baudolo-e2e-early-skip-no-backup-required")
+        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
+
+        # --- Docker resources ---
+        cls.redis_container = f"{cls.prefix}-redis"
+        cls.ignored_volume = f"{cls.prefix}-redis-vol"
+        cls.normal_volume = f"{cls.prefix}-files-vol"
+
+        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",
+                "run",
+                "-d",
+                "--name",
+                cls.redis_container,
+                "-v",
+                f"{cls.ignored_volume}:/data",
+                "redis:alpine",
+            ]
+        )
+
+        # Put deterministic content into the normal volume
+        run(
+            [
+                "docker",
+                "run",
+                "--rm",
+                "-v",
+                f"{cls.normal_volume}:/data",
+                "alpine:3.20",
+                "sh",
+                "-lc",
+                "mkdir -p /data && echo 'hello' > /data/hello.txt",
+            ]
+        )
+
+        # databases.csv required by CLI (can be empty)
+        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",
+            cls.compose_dir,
+            "--docker-compose-hard-restart-required",
+            "mailu",
+            "--repo-name",
+            cls.repo_name,
+            "--databases-csv",
+            cls.databases_csv,
+            "--backups-dir",
+            cls.backups_dir,
+            "--database-containers",
+            "dummy-db",
+            "--images-no-stop-required",
+            "alpine",
+            "redis",
+            "postgres",
+            "mariadb",
+            "mysql",
+            "--images-no-backup-required",
+            "redis",
+        ]
+        cp = run(cmd, capture=True, check=True)
+        cls.stdout = cp.stdout or ""
+        cls.stderr = cp.stderr 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 test_ignored_volume_has_no_backup_directory_at_all(self) -> None:
+        p = backup_path(
+            self.backups_dir,
+            self.repo_name,
+            self.version,
+            self.ignored_volume,
+        )
+        self.assertFalse(
+            p.exists(),
+            f"Expected NO backup directory to be created for ignored volume, but found: {p}",
+        )
+
+    def test_normal_volume_is_still_backed_up(self) -> None:
+        p = (
+            backup_path(
+                self.backups_dir,
+                self.repo_name,
+                self.version,
+                self.normal_volume,
+            )
+            / "files"
+            / "hello.txt"
+        )
+        self.assertTrue(p.is_file(), f"Expected backed up file at: {p}")

tests/e2e/test_e2e_mariadb_no_copy.py

@@ -87,7 +87,7 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
             [(cls.db_container, cls.db_name, cls.db_user, cls.db_password)],
         )
 
-        # dump-only => no files
+        # dump-only-sql => no files
         backup_run(
             backups_dir=cls.backups_dir,
             repo_name=cls.repo_name,
@@ -95,7 +95,7 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
             databases_csv=cls.databases_csv,
             database_containers=[cls.db_container],
             images_no_stop_required=["mariadb", "mysql", "alpine", "postgres"],
-            dump_only=True,
+            dump_only_sql=True,
         )
 
         cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)

tests/e2e/test_e2e_postgres_no_copy.py

@@ -75,7 +75,7 @@ class TestE2EPostgresNoCopy(unittest.TestCase):
             databases_csv=cls.databases_csv,
             database_containers=[cls.pg_container],
             images_no_stop_required=["postgres", "mariadb", "mysql", "alpine"],
-            dump_only=True,
+            dump_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

@@ -0,0 +1,232 @@
+import unittest
+
+from .helpers import (
+    backup_path,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    wait_for_postgres,
+)
+
+
+class TestE2ESeedStarAndDbEntriesBackupPostgres(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+
+        cls.prefix = unique("baudolo-e2e-seed-star-and-db")
+        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
+
+        # --- Volumes ---
+        cls.db_volume = f"{cls.prefix}-vol-db"
+        cls.files_volume = f"{cls.prefix}-vol-files"
+        cls.volumes = [cls.db_volume, cls.files_volume]
+
+        run(["docker", "volume", "create", cls.db_volume])
+        run(["docker", "volume", "create", cls.files_volume])
+
+        # Put a marker into the non-db volume
+        cls.marker = "hello-non-db-seed-star"
+        run(
+            [
+                "docker",
+                "run",
+                "--rm",
+                "-v",
+                f"{cls.files_volume}:/data",
+                "alpine:3.20",
+                "sh",
+                "-lc",
+                f"echo '{cls.marker}' > /data/hello.txt",
+            ]
+        )
+
+        # --- Start Postgres container using the DB volume ---
+        cls.pg_container = f"{cls.prefix}-pg"
+        cls.containers = [cls.pg_container]
+
+        cls.pg_password = "postgres"
+        cls.pg_user = "postgres"
+
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.pg_container,
+                "-e",
+                f"POSTGRES_PASSWORD={cls.pg_password}",
+                "-v",
+                f"{cls.db_volume}:/var/lib/postgresql/data",
+                "postgres:16-alpine",
+            ]
+        )
+        wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
+
+        # Create two DBs and deterministic content, so pg_dumpall is meaningful
+        cls.pg_db1 = "testdb1"
+        cls.pg_db2 = "testdb2"
+
+        run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                (
+                    f'psql -U {cls.pg_user} -c "CREATE DATABASE {cls.pg_db1};" || true; '
+                    f'psql -U {cls.pg_user} -c "CREATE DATABASE {cls.pg_db2};" || true; '
+                ),
+            ],
+            check=True,
+        )
+
+        run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                (
+                    f"psql -U {cls.pg_user} -d {cls.pg_db1} -c "
+                    '"CREATE TABLE IF NOT EXISTS t (id INT PRIMARY KEY, v TEXT);'
+                    "INSERT INTO t(id,v) VALUES (1,'hello-db1') "
+                    'ON CONFLICT (id) DO UPDATE SET v=EXCLUDED.v;"'
+                ),
+            ],
+            check=True,
+        )
+        run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                (
+                    f"psql -U {cls.pg_user} -d {cls.pg_db2} -c "
+                    '"CREATE TABLE IF NOT EXISTS t (id INT PRIMARY KEY, v TEXT);'
+                    "INSERT INTO t(id,v) VALUES (1,'hello-db2') "
+                    'ON CONFLICT (id) DO UPDATE SET v=EXCLUDED.v;"'
+                ),
+            ],
+            check=True,
+        )
+
+        # --- Seed databases.csv using CLI (star + concrete db) ---
+        cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
+
+        # IMPORTANT: because we pass --database-containers <container>,
+        # get_instance() will use the container name as instance key.
+        instance = cls.pg_container
+
+        # Seed star entry (pg_dumpall)
+        run(
+            [
+                "baudolo-seed",
+                cls.databases_csv,
+                instance,
+                "*",
+                cls.pg_user,
+                cls.pg_password,
+            ]
+        )
+
+        # Seed concrete DB entry (pg_dump)
+        run(
+            [
+                "baudolo-seed",
+                cls.databases_csv,
+                instance,
+                cls.pg_db1,
+                cls.pg_user,
+                cls.pg_password,
+            ]
+        )
+
+        # --- Run baudolo with dump-only-sql ---
+        cmd = [
+            "baudolo",
+            "--compose-dir",
+            cls.compose_dir,
+            "--databases-csv",
+            cls.databases_csv,
+            "--database-containers",
+            cls.pg_container,
+            "--images-no-stop-required",
+            "alpine",
+            "postgres",
+            "mariadb",
+            "mysql",
+            "--dump-only-sql",
+            "--backups-dir",
+            cls.backups_dir,
+            "--repo-name",
+            cls.repo_name,
+        ]
+        cp = run(cmd, capture=True, check=True)
+        cls.stdout = cp.stdout or ""
+        cls.stderr = cp.stderr 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 test_db_volume_has_cluster_dump_and_concrete_db_dump_and_no_files(self) -> None:
+        base = backup_path(
+            self.backups_dir, self.repo_name, self.version, self.db_volume
+        )
+        sql_dir = base / "sql"
+        files_dir = base / "files"
+
+        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}",
+        )
+
+        # Cluster dump file produced by '*' entry
+        cluster = sql_dir / f"{self.pg_container}.cluster.backup.sql"
+        self.assertTrue(cluster.is_file(), f"Expected cluster dump file at: {cluster}")
+
+        # Concrete DB dump produced by normal entry
+        db1 = sql_dir / f"{self.pg_db1}.backup.sql"
+        self.assertTrue(db1.is_file(), f"Expected db dump file at: {db1}")
+
+        # Basic sanity: cluster dump usually contains CREATE DATABASE statements
+        txt = cluster.read_text(encoding="utf-8", errors="ignore")
+        self.assertIn(
+            "CREATE DATABASE",
+            txt,
+            "Expected cluster dump to contain CREATE DATABASE statements",
+        )
+
+    def test_non_db_volume_still_has_files_backup(self) -> None:
+        base = backup_path(
+            self.backups_dir, self.repo_name, self.version, self.files_volume
+        )
+        files_dir = base / "files"
+
+        self.assertTrue(
+            files_dir.exists(), f"Expected files dir for non-DB volume at: {files_dir}"
+        )
+
+        marker = files_dir / "hello.txt"
+        self.assertTrue(marker.is_file(), f"Expected marker file at: {marker}")
+        self.assertEqual(
+            marker.read_text(encoding="utf-8").strip(),
+            self.marker,
+        )

tests/integration/test_seed_integration.py

@@ -7,9 +7,47 @@ from pathlib import Path
 
 
 def run_seed(
-    csv_path: Path, instance: str, database: str, username: str, password: str = ""
+    csv_path: Path, instance: str, database: str, username: str, password: str
 ) -> subprocess.CompletedProcess:
-    # Run the real CLI module (integration-style).
+    """
+    Run the real CLI module (E2E-style) using subprocess.
+
+    Seed contract (current):
+    - database must be "*" or a valid name (non-empty, matches allowed charset)
+    - password is required
+    - entry is keyed by (instance, database); username/password get updated
+    """
+    cp = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "baudolo.seed",
+            str(csv_path),
+            instance,
+            database,
+            username,
+            password,
+        ],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    if cp.returncode != 0:
+        raise AssertionError(
+            "seed command failed unexpectedly.\n"
+            f"returncode: {cp.returncode}\n"
+            f"stdout:\n{cp.stdout}\n"
+            f"stderr:\n{cp.stderr}\n"
+        )
+    return cp
+
+
+def run_seed_expect_fail(
+    csv_path: Path, instance: str, database: str, username: str, password: str
+) -> subprocess.CompletedProcess:
+    """
+    Same as run_seed, but expects non-zero exit. Returns CompletedProcess for inspection.
+    """
     return subprocess.run(
         [
             sys.executable,
@@ -23,7 +61,7 @@ def run_seed(
         ],
         text=True,
         capture_output=True,
-        check=True,
+        check=False,
     )
 
 
@@ -33,6 +71,10 @@ def read_csv_semicolon(path: Path) -> list[dict]:
         return list(reader)
 
 
+def read_text(path: Path) -> str:
+    return path.read_text(encoding="utf-8")
+
+
 class TestSeedIntegration(unittest.TestCase):
     def test_creates_file_and_adds_entry_when_missing(self) -> None:
         with tempfile.TemporaryDirectory() as td:
@@ -41,7 +83,7 @@ class TestSeedIntegration(unittest.TestCase):
 
             cp = run_seed(p, "docker.test", "appdb", "alice", "secret")
 
-            self.assertEqual(cp.returncode, 0, cp.stderr)
+            self.assertEqual(cp.returncode, 0)
             self.assertTrue(p.exists())
 
             rows = read_csv_semicolon(p)
@@ -51,40 +93,121 @@ class TestSeedIntegration(unittest.TestCase):
             self.assertEqual(rows[0]["username"], "alice")
             self.assertEqual(rows[0]["password"], "secret")
 
-    def test_replaces_existing_entry_same_keys(self) -> None:
+    def test_replaces_existing_entry_same_instance_and_database_updates_username_and_password(
+        self,
+    ) -> None:
+        """
+        Replacement semantics:
+        - Key is (instance, database)
+        - username/password are updated in-place
+        """
         with tempfile.TemporaryDirectory() as td:
             p = Path(td) / "databases.csv"
 
-            # First add
             run_seed(p, "docker.test", "appdb", "alice", "oldpw")
             rows = read_csv_semicolon(p)
             self.assertEqual(len(rows), 1)
+            self.assertEqual(rows[0]["username"], "alice")
             self.assertEqual(rows[0]["password"], "oldpw")
 
-            # Replace (same instance+database+username)
-            run_seed(p, "docker.test", "appdb", "alice", "newpw")
+            run_seed(p, "docker.test", "appdb", "bob", "newpw")
             rows = read_csv_semicolon(p)
 
             self.assertEqual(len(rows), 1, "Expected replacement, not a duplicate row")
             self.assertEqual(rows[0]["instance"], "docker.test")
             self.assertEqual(rows[0]["database"], "appdb")
-            self.assertEqual(rows[0]["username"], "alice")
+            self.assertEqual(rows[0]["username"], "bob")
             self.assertEqual(rows[0]["password"], "newpw")
 
-    def test_database_empty_string_matches_existing_empty_database(self) -> None:
+    def test_allows_star_database_for_dump_all(self) -> None:
         with tempfile.TemporaryDirectory() as td:
             p = Path(td) / "databases.csv"
 
-            # Add with empty database
-            run_seed(p, "docker.test", "", "alice", "pw1")
+            cp = run_seed(p, "bigbluebutton", "*", "postgres", "pw")
+            self.assertEqual(cp.returncode, 0)
+
             rows = read_csv_semicolon(p)
             self.assertEqual(len(rows), 1)
-            self.assertEqual(rows[0]["database"], "")
+            self.assertEqual(rows[0]["instance"], "bigbluebutton")
+            self.assertEqual(rows[0]["database"], "*")
+            self.assertEqual(rows[0]["username"], "postgres")
+            self.assertEqual(rows[0]["password"], "pw")
+
+    def test_replaces_existing_star_entry(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            p = Path(td) / "databases.csv"
+
+            run_seed(p, "bigbluebutton", "*", "postgres", "pw1")
+            run_seed(p, "bigbluebutton", "*", "postgres", "pw2")
 
-            # Replace with empty database again
-            run_seed(p, "docker.test", "", "alice", "pw2")
             rows = read_csv_semicolon(p)
-
             self.assertEqual(len(rows), 1)
-            self.assertEqual(rows[0]["database"], "")
+            self.assertEqual(rows[0]["database"], "*")
             self.assertEqual(rows[0]["password"], "pw2")
+
+    def test_rejects_empty_database_value(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            p = Path(td) / "databases.csv"
+
+            cp = run_seed_expect_fail(p, "docker.test", "", "alice", "pw")
+            self.assertNotEqual(cp.returncode, 0)
+
+            combined = ((cp.stdout or "") + "\n" + (cp.stderr or "")).lower()
+            self.assertIn("error:", combined)
+            self.assertIn("database", combined)
+            self.assertIn("not empty", combined)
+
+            self.assertFalse(p.exists(), "Should not create file on invalid input")
+
+    def test_rejects_invalid_database_name_characters(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            p = Path(td) / "databases.csv"
+
+            cp = run_seed_expect_fail(p, "docker.test", "app db", "alice", "pw")
+            self.assertNotEqual(cp.returncode, 0)
+
+            combined = ((cp.stdout or "") + "\n" + (cp.stderr or "")).lower()
+            self.assertIn("error:", combined)
+            self.assertIn("invalid database name", combined)
+
+            self.assertFalse(p.exists(), "Should not create file on invalid input")
+
+    def test_rejects_nan_database_name(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            p = Path(td) / "databases.csv"
+
+            cp = run_seed_expect_fail(p, "docker.test", "nan", "alice", "pw")
+            self.assertNotEqual(cp.returncode, 0)
+
+            combined = ((cp.stdout or "") + "\n" + (cp.stderr or "")).lower()
+            self.assertIn("error:", combined)
+            self.assertIn("must not be 'nan'", combined)
+
+            self.assertFalse(p.exists(), "Should not create file on invalid input")
+
+    def test_accepts_hyphen_and_underscore_database_names(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            p = Path(td) / "databases.csv"
+
+            run_seed(p, "docker.test", "my_db-1", "alice", "pw")
+
+            rows = read_csv_semicolon(p)
+            self.assertEqual(len(rows), 1)
+            self.assertEqual(rows[0]["database"], "my_db-1")
+
+    def test_file_is_semicolon_delimited_and_has_header(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            p = Path(td) / "databases.csv"
+
+            run_seed(p, "docker.test", "appdb", "alice", "pw")
+
+            txt = read_text(p)
+            self.assertTrue(
+                txt.startswith("instance;database;username;password"),
+                f"Unexpected header / delimiter in file:\n{txt}",
+            )
+            self.assertIn(";", txt)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_app_databases_csv.py

@@ -0,0 +1,77 @@
+import io
+import os
+import tempfile
+import unittest
+from contextlib import redirect_stderr
+
+import pandas as pd
+
+# Adjust if your package name/import path differs.
+from baudolo.backup.app import _load_databases_df
+
+
+EXPECTED_COLUMNS = ["instance", "database", "username", "password"]
+
+
+class TestLoadDatabasesDf(unittest.TestCase):
+    def test_missing_csv_is_handled_with_warning_and_empty_df(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            missing_path = os.path.join(td, "does-not-exist.csv")
+
+            buf = io.StringIO()
+            with redirect_stderr(buf):
+                df = _load_databases_df(missing_path)
+
+            stderr = buf.getvalue()
+            self.assertIn("WARNING:", stderr)
+            self.assertIn("databases.csv not found", stderr)
+
+            self.assertIsInstance(df, pd.DataFrame)
+            self.assertListEqual(list(df.columns), EXPECTED_COLUMNS)
+            self.assertTrue(df.empty)
+
+    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("")
+
+            buf = io.StringIO()
+            with redirect_stderr(buf):
+                df = _load_databases_df(empty_path)
+
+            stderr = buf.getvalue()
+            self.assertIn("WARNING:", stderr)
+            self.assertIn("exists but is empty", stderr)
+
+            self.assertIsInstance(df, pd.DataFrame)
+            self.assertListEqual(list(df.columns), EXPECTED_COLUMNS)
+            self.assertTrue(df.empty)
+
+    def test_valid_csv_loads_without_warning(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            csv_path = os.path.join(td, "databases.csv")
+
+            content = "instance;database;username;password\nmyapp;*;dbuser;secret\n"
+            with open(csv_path, "w", encoding="utf-8") as f:
+                f.write(content)
+
+            buf = io.StringIO()
+            with redirect_stderr(buf):
+                df = _load_databases_df(csv_path)
+
+            stderr = buf.getvalue()
+            self.assertEqual(stderr, "")  # no warning expected
+
+            self.assertIsInstance(df, pd.DataFrame)
+            self.assertListEqual(list(df.columns), EXPECTED_COLUMNS)
+            self.assertEqual(len(df), 1)
+            self.assertEqual(df.loc[0, "instance"], "myapp")
+            self.assertEqual(df.loc[0, "database"], "*")
+            self.assertEqual(df.loc[0, "username"], "dbuser")
+            self.assertEqual(df.loc[0, "password"], "secret")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_compose.py

@@ -0,0 +1,239 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+from typing import List
+from unittest.mock import patch
+
+
+def _touch(p: Path) -> None:
+    p.parent.mkdir(parents=True, exist_ok=True)
+
+    # If the path already exists as a directory (e.g. ".env" created by ".env/env"),
+    # remove it so we can create a file with the same name.
+    if p.exists() and p.is_dir():
+        shutil.rmtree(p)
+
+    p.write_text("x", encoding="utf-8")
+
+
+def _setup_compose_dir(
+    tmp_path: Path,
+    name: str = "mailu",
+    *,
+    with_override: bool = False,
+    with_ca_override: bool = False,
+    env_layout: str | None = None,  # None | ".env" | ".env/env"
+) -> Path:
+    d = tmp_path / name
+    d.mkdir(parents=True, exist_ok=True)
+
+    _touch(d / "docker-compose.yml")
+
+    if with_override:
+        _touch(d / "docker-compose.override.yml")
+
+    if with_ca_override:
+        _touch(d / "docker-compose.ca.override.yml")
+
+    if env_layout == ".env":
+        _touch(d / ".env")
+    elif env_layout == ".env/env":
+        _touch(d / ".env" / "env")
+
+    return d
+
+
+class TestCompose(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        from baudolo.backup import compose as mod
+
+        cls.compose_mod = mod
+
+    def test_detect_env_file_prefers_dotenv_over_legacy(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            d = _setup_compose_dir(tmp_path, env_layout=".env/env")
+            # Also create .env file -> should be preferred
+            _touch(d / ".env")
+
+            env_file = self.compose_mod._detect_env_file(d)
+            self.assertEqual(env_file, d / ".env")
+
+    def test_detect_env_file_uses_legacy_if_no_dotenv(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            d = _setup_compose_dir(tmp_path, env_layout=".env/env")
+
+            env_file = self.compose_mod._detect_env_file(d)
+            self.assertEqual(env_file, d / ".env" / "env")
+
+    def test_detect_compose_files_requires_base(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            d = tmp_path / "stack"
+            d.mkdir()
+
+            with self.assertRaises(FileNotFoundError):
+                self.compose_mod._detect_compose_files(d)
+
+    def test_detect_compose_files_includes_optional_overrides(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            d = _setup_compose_dir(
+                tmp_path,
+                with_override=True,
+                with_ca_override=True,
+            )
+
+            files = self.compose_mod._detect_compose_files(d)
+            self.assertEqual(
+                files,
+                [
+                    d / "docker-compose.yml",
+                    d / "docker-compose.override.yml",
+                    d / "docker-compose.ca.override.yml",
+                ],
+            )
+
+    def test_build_cmd_uses_wrapper_when_present(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            d = _setup_compose_dir(
+                tmp_path, with_override=True, with_ca_override=True, env_layout=".env"
+            )
+
+            with patch.object(
+                self.compose_mod.shutil, "which", lambda name: "/usr/local/bin/compose"
+            ):
+                cmd = self.compose_mod._build_compose_cmd(str(d), ["up", "-d"])
+
+            self.assertEqual(
+                cmd,
+                [
+                    "/usr/local/bin/compose",
+                    "--chdir",
+                    str(d.resolve()),
+                    "--",
+                    "up",
+                    "-d",
+                ],
+            )
+
+    def test_build_cmd_fallback_docker_compose_with_all_files_and_env(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            d = _setup_compose_dir(
+                tmp_path,
+                with_override=True,
+                with_ca_override=True,
+                env_layout=".env",
+            )
+
+            with patch.object(self.compose_mod.shutil, "which", lambda name: None):
+                cmd = self.compose_mod._build_compose_cmd(
+                    str(d), ["up", "-d", "--force-recreate"]
+                )
+
+            expected: List[str] = [
+                "docker",
+                "compose",
+                "-f",
+                str((d / "docker-compose.yml").resolve()),
+                "-f",
+                str((d / "docker-compose.override.yml").resolve()),
+                "-f",
+                str((d / "docker-compose.ca.override.yml").resolve()),
+                "--env-file",
+                str((d / ".env").resolve()),
+                "up",
+                "-d",
+                "--force-recreate",
+            ]
+            self.assertEqual(cmd, expected)
+
+    def test_hard_restart_calls_run_twice_with_correct_cmds_wrapper(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            d = _setup_compose_dir(tmp_path, name="mailu", env_layout=".env")
+
+            with patch.object(
+                self.compose_mod.shutil, "which", lambda name: "/usr/local/bin/compose"
+            ):
+                calls = []
+
+                def fake_run(cmd, check: bool):
+                    calls.append((cmd, check))
+                    return 0
+
+                with patch.object(self.compose_mod.subprocess, "run", fake_run):
+                    self.compose_mod.hard_restart_docker_services(str(d))
+
+            self.assertEqual(
+                calls,
+                [
+                    (
+                        [
+                            "/usr/local/bin/compose",
+                            "--chdir",
+                            str(d.resolve()),
+                            "--",
+                            "down",
+                        ],
+                        True,
+                    ),
+                    (
+                        [
+                            "/usr/local/bin/compose",
+                            "--chdir",
+                            str(d.resolve()),
+                            "--",
+                            "up",
+                            "-d",
+                        ],
+                        True,
+                    ),
+                ],
+            )
+
+    def test_hard_restart_calls_run_twice_with_correct_cmds_fallback(self) -> None:
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            d = _setup_compose_dir(
+                tmp_path,
+                name="mailu",
+                with_override=True,
+                with_ca_override=True,
+                env_layout=".env/env",
+            )
+
+            with patch.object(self.compose_mod.shutil, "which", lambda name: None):
+                calls = []
+
+                def fake_run(cmd, check: bool):
+                    calls.append((cmd, check))
+                    return 0
+
+                with patch.object(self.compose_mod.subprocess, "run", fake_run):
+                    self.compose_mod.hard_restart_docker_services(str(d))
+
+            down_cmd = calls[0][0]
+            up_cmd = calls[1][0]
+
+            self.assertTrue(calls[0][1] is True)
+            self.assertTrue(calls[1][1] is True)
+
+            self.assertEqual(down_cmd[0:2], ["docker", "compose"])
+            self.assertEqual(down_cmd[-1], "down")
+            self.assertIn("--env-file", down_cmd)
+
+            self.assertEqual(up_cmd[0:2], ["docker", "compose"])
+            self.assertTrue(up_cmd[-2:] == ["up", "-d"] or up_cmd[-3:] == ["up", "-d"])
+            self.assertIn("--env-file", up_cmd)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tests/unit/seed/test_main.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+import pandas as pd
+from pandas.errors import EmptyDataError
+
+import baudolo.seed.__main__ as seed_main
+
+
+class TestSeedMain(unittest.TestCase):
+    @patch("baudolo.seed.__main__.pd.DataFrame")
+    def test_empty_df_creates_expected_columns(self, df_ctor: MagicMock) -> None:
+        seed_main._empty_df()
+        df_ctor.assert_called_once_with(
+            columns=["instance", "database", "username", "password"]
+        )
+
+    def test_validate_database_value_rejects_empty(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")
+
+    def _mock_df_mask_any(self, *, any_value: bool) -> MagicMock:
+        """
+        Build a DataFrame-like mock such that:
+          mask = (df["instance"] == instance) & (df["database"] == database)
+          mask.any() returns any_value
+        """
+        df = MagicMock(spec=pd.DataFrame)
+
+        left = MagicMock()
+        right = MagicMock()
+
+        mask = MagicMock()
+        mask.any.return_value = any_value
+
+        # (left & right) => mask
+        left.__and__.return_value = mask
+
+        # df["instance"] / df["database"] => return objects whose == produces left/right
+        col = MagicMock()
+        col.__eq__.side_effect = [left, right]
+        df.__getitem__.return_value = col
+
+        return df
+
+    @patch("baudolo.seed.__main__.os.path.exists", return_value=False)
+    @patch("baudolo.seed.__main__.pd.read_csv")
+    @patch("baudolo.seed.__main__._empty_df")
+    @patch("baudolo.seed.__main__.pd.concat")
+    def test_check_and_add_entry_file_missing_adds_entry(
+        self,
+        concat: MagicMock,
+        empty_df: MagicMock,
+        read_csv: MagicMock,
+        exists: MagicMock,
+    ) -> None:
+        df_existing = self._mock_df_mask_any(any_value=False)
+        empty_df.return_value = df_existing
+
+        df_out = MagicMock(spec=pd.DataFrame)
+        concat.return_value = df_out
+
+        seed_main.check_and_add_entry(
+            file_path="/tmp/databases.csv",
+            instance="inst",
+            database="db",
+            username="user",
+            password="pass",
+        )
+
+        read_csv.assert_not_called()
+        empty_df.assert_called_once()
+        concat.assert_called_once()
+        df_out.to_csv.assert_called_once_with(
+            "/tmp/databases.csv", sep=";", index=False
+        )
+
+    @patch("baudolo.seed.__main__.os.path.exists", return_value=True)
+    @patch("baudolo.seed.__main__.pd.read_csv", side_effect=EmptyDataError("empty"))
+    @patch("baudolo.seed.__main__._empty_df")
+    @patch("baudolo.seed.__main__.pd.concat")
+    @patch("baudolo.seed.__main__.print")
+    def test_check_and_add_entry_empty_file_warns_and_creates_columns_and_adds(
+        self,
+        print_: MagicMock,
+        concat: MagicMock,
+        empty_df: MagicMock,
+        read_csv: MagicMock,
+        exists: MagicMock,
+    ) -> None:
+        df_existing = self._mock_df_mask_any(any_value=False)
+        empty_df.return_value = df_existing
+
+        df_out = MagicMock(spec=pd.DataFrame)
+        concat.return_value = df_out
+
+        seed_main.check_and_add_entry(
+            file_path="/tmp/databases.csv",
+            instance="inst",
+            database="db",
+            username="user",
+            password="pass",
+        )
+
+        exists.assert_called_once_with("/tmp/databases.csv")
+        read_csv.assert_called_once()
+        empty_df.assert_called_once()
+        concat.assert_called_once()
+
+        # Assert: at least one print call contains the WARNING and prints to stderr
+        warning_calls = []
+        for call in print_.call_args_list:
+            args, kwargs = call
+            if args and "WARNING: databases.csv exists but is empty" in str(args[0]):
+                warning_calls.append((args, kwargs))
+
+        self.assertTrue(
+            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)
+
+        df_out.to_csv.assert_called_once_with(
+            "/tmp/databases.csv", sep=";", index=False
+        )
+
+    @patch("baudolo.seed.__main__.os.path.exists", return_value=True)
+    @patch("baudolo.seed.__main__.pd.read_csv")
+    def test_check_and_add_entry_updates_existing_row(
+        self,
+        read_csv: MagicMock,
+        exists: MagicMock,
+    ) -> None:
+        df = self._mock_df_mask_any(any_value=True)
+        read_csv.return_value = df
+
+        seed_main.check_and_add_entry(
+            file_path="/tmp/databases.csv",
+            instance="inst",
+            database="db",
+            username="user",
+            password="pass",
+        )
+
+        df.to_csv.assert_called_once_with("/tmp/databases.csv", sep=";", index=False)
+
+    @patch("baudolo.seed.__main__.check_and_add_entry")
+    @patch("baudolo.seed.__main__.argparse.ArgumentParser.parse_args")
+    def test_main_calls_check_and_add_entry(
+        self, parse_args: MagicMock, cae: MagicMock
+    ) -> None:
+        ns = MagicMock()
+        ns.file = "/tmp/databases.csv"
+        ns.instance = "inst"
+        ns.database = "db"
+        ns.username = "user"
+        ns.password = "pass"
+        parse_args.return_value = ns
+
+        seed_main.main()
+
+        cae.assert_called_once_with(
+            file_path="/tmp/databases.csv",
+            instance="inst",
+            database="db",
+            username="user",
+            password="pass",
+        )
+
+    @patch("baudolo.seed.__main__.sys.exit")
+    @patch("baudolo.seed.__main__.print")
+    @patch(
+        "baudolo.seed.__main__.check_and_add_entry", side_effect=RuntimeError("boom")
+    )
+    @patch("baudolo.seed.__main__.argparse.ArgumentParser.parse_args")
+    def test_main_exits_nonzero_on_error(
+        self,
+        parse_args: MagicMock,
+        cae: MagicMock,
+        print_: MagicMock,
+        exit_: MagicMock,
+    ) -> None:
+        ns = MagicMock()
+        ns.file = "/tmp/databases.csv"
+        ns.instance = "inst"
+        ns.database = "db"
+        ns.username = "user"
+        ns.password = "pass"
+        parse_args.return_value = ns
+
+        seed_main.main()
+
+        self.assertTrue(print_.called)
+        _, kwargs = print_.call_args
+        self.assertEqual(kwargs.get("file"), seed_main.sys.stderr)
+        exit_.assert_called_once_with(1)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)