Add some very basic unit tests to ensure backups complete successfully

db-auto-backup.py

@@ -53,28 +53,28 @@ def temp_backup_file_name() -> str:
     return ".auto-backup-" + secrets.token_hex(4)
 
 
-def open_file_compressed(file_path: Path) -> IO[bytes]:
-    if COMPRESSION == "gzip":
+def open_file_compressed(file_path: Path, algorithm: str) -> IO[bytes]:
+    if algorithm == "gzip":
         return gzip.open(file_path, mode="wb")  # type:ignore
-    elif COMPRESSION in ["lzma", "xz"]:
+    elif algorithm in ["lzma", "xz"]:
         return lzma.open(file_path, mode="wb")
-    elif COMPRESSION == "bz2":
+    elif algorithm == "bz2":
         return bz2.open(file_path, mode="wb")
-    elif COMPRESSION == "plain":
+    elif algorithm == "plain":
         return file_path.open(mode="wb")
-    raise ValueError(f"Unknown compression method {COMPRESSION}")
+    raise ValueError(f"Unknown compression method {algorithm}")
 
 
-def get_compressed_file_extension() -> str:
-    if COMPRESSION == "gzip":
+def get_compressed_file_extension(algorithm: str) -> str:
+    if algorithm == "gzip":
         return ".gz"
-    elif COMPRESSION in ["lzma", "xz"]:
+    elif algorithm in ["lzma", "xz"]:
         return ".xz"
-    elif COMPRESSION == "bz2":
+    elif algorithm == "bz2":
         return ".bz2"
-    elif COMPRESSION == "plain":
+    elif algorithm == "plain":
         return ""
-    raise ValueError(f"Unknown compression method {COMPRESSION}")
+    raise ValueError(f"Unknown compression method {algorithm}")
 
 
 def backup_psql(container: Container) -> str:
@@ -154,14 +154,16 @@ def backup(now: datetime) -> None:
 
         backup_file = (
             BACKUP_DIR
-            / f"{container.name}.{backup_provider.file_extension}{get_compressed_file_extension()}"
+            / f"{container.name}.{backup_provider.file_extension}{get_compressed_file_extension(COMPRESSION)}"
         )
         backup_temp_file_path = BACKUP_DIR / temp_backup_file_name()
 
         backup_command = backup_provider.backup_method(container)
         _, output = container.exec_run(backup_command, stream=True, demux=True)
 
-        with open_file_compressed(backup_temp_file_path) as backup_temp_file:
+        with open_file_compressed(
+            backup_temp_file_path, COMPRESSION
+        ) as backup_temp_file:
             with tqdm.wrapattr(
                 backup_temp_file,
                 method="write",

dev-requirements.txt

@@ -4,3 +4,4 @@ black==23.1.0
 ruff==0.0.256
 mypy==1.1.1
 types-requests
+pytest==7.4.0

docker-compose.yml

@@ -3,6 +3,7 @@ services:
     build:
       context: .
     restart: unless-stopped
+    command: "sleep infinity"
     volumes:
       - "/var/run/docker.sock:/var/run/docker.sock:ro"
       - ./backups:/var/backups

scripts/fix.sh

@@ -4,5 +4,5 @@ set -e
 
 export PATH=env/bin:${PATH}
 
-black db-auto-backup.py
-ruff --fix db-auto-backup.py
+black db-auto-backup.py tests
+ruff --fix db-auto-backup.py tests

scripts/lint.sh

@@ -6,8 +6,8 @@ export PATH=env/bin:${PATH}
 
 set -x
 
-black db-auto-backup.py --check
+black db-auto-backup.py tests --check
 
-ruff check db-auto-backup.py
+ruff check db-auto-backup.py tests
 
-mypy db-auto-backup.py
+mypy db-auto-backup.py tests

scripts/test.sh

@@ -2,26 +2,24 @@
 
 set -e
 
+export PATH=env/bin:${PATH}
+
 echo "> Start all containers..."
 docker-compose up -d
 
-echo "> Stop backup container..."
-docker-compose stop backup
-
 echo "> Await postgres..."
 until docker-compose exec -T psql pg_isready -U postgres
 do
   sleep 1
 done
+
 echo "> Await mysql..."
 until docker-compose exec -T mysql bash -c 'mysqladmin ping --protocol tcp -p$MYSQL_ROOT_PASSWORD'
 do
-  sleep 1
+  sleep 3
 done
 
-echo "> Run backups..."
-# Unset `$SCHEDULE` to run just once
-docker-compose run -e "SCHEDULE=" backup ./db-auto-backup.py
+pytest -v
 
 echo "> Clean up..."
 docker-compose down

setup.cfg

@@ -17,3 +17,6 @@ ignore_missing_imports = True
 disallow_untyped_calls = True
 disallow_untyped_defs = True
 disallow_incomplete_defs = True
+
+[tool:pytest]
+python_files = tests.py

tests/conftest.py

@@ -0,0 +1,29 @@
+from pathlib import Path
+from typing import Any, Callable
+
+import docker
+import pytest
+
+BACKUP_DIR = Path.cwd() / "backups"
+
+
+@pytest.fixture
+def run_backup(request: Any) -> Callable:
+    docker_client = docker.from_env()
+    backup_container = docker_client.containers.get("docker-db-auto-backup-backup-1")
+
+    def clean_backups() -> None:
+        # HACK: Remove files from inside container to avoid permissions issue
+        backup_container.exec_run(["rm", "-rf", "/var/backups"])
+
+    def _run_backup(env: dict) -> Any:
+        return backup_container.exec_run(
+            [
+                "./db-auto-backup.py",
+            ],
+            environment={**env, "SCHEDULE": ""},
+        )
+
+    request.addfinalizer(clean_backups)
+
+    return _run_backup

tests/tests.py

@@ -0,0 +1,46 @@
+from importlib.machinery import SourceFileLoader
+from importlib.util import module_from_spec, spec_from_loader
+from pathlib import Path
+from typing import Any, Callable
+
+import pytest
+
+BACKUP_DIR = Path.cwd() / "backups"
+
+
+def import_file(path: Path) -> Any:
+    """
+    Import a module from a file path, returning its contents.
+    """
+    loader = SourceFileLoader(path.name, str(path))
+    spec = spec_from_loader(path.name, loader)
+    assert spec is not None
+    mod = module_from_spec(spec)
+    loader.exec_module(mod)
+    return mod
+
+
+# HACK: The filename isn't compatible with `import foo` syntax
+db_auto_backup = import_file(Path.cwd() / "db-auto-backup.py")
+
+
+def test_backup_runs(run_backup: Callable) -> None:
+    exit_code, _ = run_backup({})
+    assert exit_code == 0
+    assert BACKUP_DIR.is_dir()
+    assert sorted(f.name for f in BACKUP_DIR.glob("*")) == [
+        "docker-db-auto-backup-mariadb-1.sql",
+        "docker-db-auto-backup-mysql-1.sql",
+        "docker-db-auto-backup-psql-1.sql",
+        "docker-db-auto-backup-redis-1.rdb",
+    ]
+    for backup_file in BACKUP_DIR.glob("*"):
+        assert backup_file.stat().st_size > 0
+
+
+@pytest.mark.parametrize(
+    "algorithm,extension",
+    [("gzip", ".gz"), ("lzma", ".xz"), ("xz", ".xz"), ("bz2", ".bz2"), ("plain", "")],
+)
+def test_compressed_file_extension(algorithm: str, extension: str) -> None:
+    assert db_auto_backup.get_compressed_file_extension(algorithm) == extension