Kevin Veen-BirkenbachandClaude Opus 5
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>
117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
import pandas as pd
|
|
from pandas.errors import EmptyDataError
|
|
|
|
DB_NAME_RE = re.compile(r"^[a-zA-Z0-9_][a-zA-Z0-9_-]*$")
|
|
|
|
|
|
def _validate_database_value(value: str | None, *, 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)."
|
|
)
|
|
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
|
|
|
|
|
|
def _empty_df() -> pd.DataFrame:
|
|
return pd.DataFrame(columns=["instance", "database", "username", "password"])
|
|
|
|
|
|
def check_and_add_entry(
|
|
file_path: str,
|
|
instance: str,
|
|
database: str | None,
|
|
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)
|
|
|
|
df.to_csv(file_path, sep=";", index=False)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Seed or update databases.csv for backup configuration."
|
|
)
|
|
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()
|
|
|
|
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: # noqa: BLE001 - CLI boundary: any failure becomes exit 1
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|