mirror of
https://github.com/kevinveenbirkenbach/docker-volume-backup.git
synced 2026-08-20 13:12:48 +00:00
Kevin Veen-Birkenbach
03013b6c76
The schema lived three times in this repo alone: the seed and the backup each spelled out the column list and the semicolon, and _validate_database_value existed twice under one name with different strictness - the seed checked the character set, the backup only checked for empty. A hand-edited file therefore bypassed the only real validation on its way into a dump command. baudolo.databases now holds the columns, the delimiter, the cluster marker, the validator and a read_rows() for consumers that do not want pandas. Values come back verbatim: a password may begin or end with a space, so stripping belongs to the caller that compares, never to the reader. DatabasesCsvError subclasses ValueError, so callers that predate the module keep catching what they caught. The backup's call sites switch over in the next commit, which rewrites them anyway. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""Contract of databases.csv: the seed writes it, the backup and a restore read it."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from baudolo.databases import (
|
|
CLUSTER_ROW,
|
|
COLUMNS,
|
|
DELIMITER,
|
|
DatabasesCsvError,
|
|
Row,
|
|
read_rows,
|
|
validate_database,
|
|
)
|
|
|
|
HEADER = DELIMITER.join(COLUMNS)
|
|
|
|
|
|
def _csv(*lines: str) -> str:
|
|
path = Path(tempfile.mkdtemp()) / "databases.csv"
|
|
path.write_text("\n".join((HEADER, *lines)) + "\n", encoding="utf-8")
|
|
return str(path)
|
|
|
|
|
|
class TestValidateDatabase(unittest.TestCase):
|
|
def test_a_concrete_name_passes(self) -> None:
|
|
self.assertEqual(validate_database("app_db-1", instance="x"), "app_db-1")
|
|
|
|
def test_the_cluster_marker_passes(self) -> None:
|
|
self.assertEqual(validate_database(CLUSTER_ROW, instance="x"), CLUSTER_ROW)
|
|
|
|
def test_an_empty_column_is_rejected(self) -> None:
|
|
with self.assertRaises(DatabasesCsvError):
|
|
validate_database("", instance="x")
|
|
|
|
def test_the_string_nan_is_rejected(self) -> None:
|
|
"""pandas used to hand back NaN, which wrote a nan.backup.sql."""
|
|
with self.assertRaises(DatabasesCsvError):
|
|
validate_database("nan", instance="x")
|
|
|
|
def test_a_name_that_could_reach_a_shell_is_rejected(self) -> None:
|
|
for hostile in ("bad name", "a;rm -rf /", "$(id)", "a`id`", "a/b"):
|
|
with self.subTest(name=hostile), self.assertRaises(DatabasesCsvError):
|
|
validate_database(hostile, instance="x")
|
|
|
|
def test_the_error_is_a_value_error(self) -> None:
|
|
"""Callers predating the shared module catch ValueError."""
|
|
with self.assertRaises(ValueError):
|
|
validate_database("", instance="x")
|
|
|
|
|
|
class TestReadRows(unittest.TestCase):
|
|
def test_the_header_is_skipped(self) -> None:
|
|
rows = read_rows(_csv(f"pg{DELIMITER}app{DELIMITER}u{DELIMITER}p"))
|
|
self.assertEqual(rows, [Row("pg", "app", "u", "p")])
|
|
|
|
def test_a_blank_row_is_dropped(self) -> None:
|
|
rows = read_rows(_csv("", f"pg{DELIMITER}app{DELIMITER}u{DELIMITER}p", ""))
|
|
self.assertEqual(len(rows), 1)
|
|
|
|
def test_a_short_row_is_refused(self) -> None:
|
|
with self.assertRaises(DatabasesCsvError):
|
|
read_rows(_csv(f"pg{DELIMITER}app{DELIMITER}u"))
|
|
|
|
def test_values_arrive_verbatim(self) -> None:
|
|
"""A password may legitimately begin or end with a space."""
|
|
rows = read_rows(_csv(f"pg{DELIMITER}app{DELIMITER}u{DELIMITER} pw "))
|
|
self.assertEqual(rows[0].password, " pw ")
|
|
|
|
def test_a_cluster_row_knows_itself(self) -> None:
|
|
rows = read_rows(
|
|
_csv(
|
|
f"pg{DELIMITER}{CLUSTER_ROW}{DELIMITER}postgres{DELIMITER}p",
|
|
f"pg{DELIMITER}app{DELIMITER}u{DELIMITER}p",
|
|
)
|
|
)
|
|
self.assertEqual([row.is_cluster for row in rows], [True, False])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|