mirror of
https://github.com/kevinveenbirkenbach/docker-volume-backup.git
synced 2026-08-20 13:12:48 +00:00
Kevin Veen-Birkenbach
2129c5e362
ruff was never wired into this repository: no target, no CI step, no pin. It reported 45 findings across sources and tests, so nothing enforced what the codebase already mostly followed. Adds `make ruff` (check + format --check), `make ruff-fix`, and `make lint` as its alias, and makes `make test` run lint as a fourth parallel spur. The CI workflow calls `make test`, so it is covered there too. The linter is pinned in a `lint` extra: a ruff minor bump changes which rules fire, and with the suite gating on a clean run an unpinned linter would fail it on an unrelated day. The 45 findings are fixed rather than configured away. Three needed a decision instead of the mechanical fix: - The generation timestamp keeps its local wall clock (DTZ005 waived). Generations sort by that name, and UTC would order new ones before the existing ones wherever the offset is positive - "newest generation" is what every restore path selects on. - The per-volume `copy` closure now binds volume_name and vol_dir as default arguments (B023). It only worked because it is called inside the same iteration. - The two CLI top-level handlers keep their blind except (BLE001 waived): turning any failure into exit 1 is what a CLI boundary is for. The two in run.py did not need it and were narrowed to what they actually catch. Also drops the comments that restate the code: the section banners in restore/__main__.py, the filename repeated as line 1 of nine test files, step narration above the statement it narrates, and a block in app.py documenting parameters that had moved to another module. What names a trip-wire stays - the snapshot destination rule, the mysql-binary absence in MariaDB 11 images, the session-scoped FOREIGN_KEY_CHECKS, the spooled temp file for multi-GB dumps, and the negative control that loses its discriminating power if it ever passes. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
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.dumps 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")
|
|
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()
|