mirror of
https://github.com/kevinveenbirkenbach/docker-volume-backup.git
synced 2026-08-20 21:22:54 +00:00
The package carried no ruff configuration at all, so it ran on the defaults (E4+E7+E9+F) while infinito-nexus-core, its only consumer, holds itself to a far wider selection. Measured against that selection this tree had 224 findings. It now has none. The selector list is core's verbatim so both repositories answer to one bar. target-version stays py39 rather than core's py311, because requires-python still declares >=3.9 and pyupgrade would otherwise propose syntax the declared minimum cannot run. Every ignore carries its reason: S603/S607 in particular, since running docker and dump binaries from PATH in list form is this tool's whole job and is already injection-safe. Two conversions are judgement rather than mechanics. os.path.join(dir, '') was the rsync idiom for a trailing separator, which Path drops, so it becomes an explicit os.sep. os.path.abspath stays where Path.resolve() would follow symlinks and let a symlinked volume test as inside the snapshot subject. BREAKING CHANGE: VersionMismatch is renamed VersionMismatchError. 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 tempfile
|
|
import unittest
|
|
from contextlib import redirect_stderr
|
|
from pathlib import Path
|
|
|
|
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 = str(Path(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 = Path(td) / "databases.csv"
|
|
with empty_path.open("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 = Path(td) / "databases.csv"
|
|
|
|
content = "instance;database;username;password\nmyapp;*;dbuser;secret\n"
|
|
with csv_path.open("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()
|