mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5
575a5a87ac
Every script a role ships now lives under `files/<language>/`, and the unit
suite is split the same way: `tests/unit/python/` beside new sibling trees for
the other languages. `tests/lint/filesystem/test_role_files_layout.py` keeps it
that way; directories whose shape is dictated from outside carry a `.nocheck`
naming the constraint (Docker build contexts, WordPress mu-plugins, the Joomla
extension, Odoo addons, Playwright suites, the `files/test/test.sh` harness
that `test-e2e-cli` resolves generically).
This cannot be applied in halves. Moving a file and repointing what names it
are the same change: 14 references resolve only at deploy time, so the tree is
green either before or after, never between.
Those 14 were found by static resolution over every reference class, not by the
test suite -- a full `make quality-high` was green while all of them were
broken:
* 7 `src:` values built from a role variable holding a bare filename
(svc-bkp-remote-2-local, decidim, espocrm, friendica, joomla, pixelfed x2).
The variable doubles as the destination name inside the container, so the
language segment belongs at the call site, not in the variable.
* `sys-service/tasks/04_base.yml` resolved every consumer's systemd script
through `first_found(..., errors='strict')` over `files/script.{sh,py}`.
18 roles ship exactly that and have no template fallback: the play would
have died on each.
* `sys-front-inj-javascript` fell back to `<role>/files/javascript.js`; the
interface now reads `files/javascript/javascript.js`.
* `sys-front-inj-logout/templates/head_sub.j2` built the cache-busting mtime
path through `path_absolute(... 'files', ...)`.
* 4 `ansible.builtin.script:` invocations naming a bare command
(pull.py, prune.sh, set_postgres_superuser_password.sh, ganesha-serve.sh).
* `dev-python/files/shell/install.sh` pointed at a sibling role's script.
Depth-dependent paths inside the moved trees were bumped accordingly: 59
`parents[N]` computations, 12 relative climbs, and the per-role recover CLI,
whose path `cli/administration/recover/recoverers.py` still assembled from the
pre-move location.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
538 lines
20 KiB
Python
Executable File
538 lines
20 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest import TestCase, main, mock
|
|
|
|
from baudolo.restore.paths import BackupPaths
|
|
|
|
from utils.recovery import databases
|
|
from utils.recovery import docker as recovery_docker
|
|
from utils.recovery.layout import MANIFEST_FILE, MANIFEST_SCHEMA
|
|
|
|
APPLICATIONS = {
|
|
"web-app-zammad": {"services": {"postgres": {"enabled": True}}},
|
|
"web-app-nextcloud": {"services": {"mariadb": {"enabled": True}}},
|
|
"web-app-static": {"services": {"redis": {"enabled": True}}},
|
|
}
|
|
|
|
|
|
def generation(root: Path) -> Path:
|
|
path = root / "Backups" / "abc123" / "backup-docker-to-local" / "20260816190906"
|
|
path.mkdir(parents=True)
|
|
return path
|
|
|
|
|
|
class TestEngineResolution(TestCase):
|
|
def setUp(self):
|
|
self.engines = databases.engine_by_key(APPLICATIONS)
|
|
|
|
def test_central_instances_are_their_own_engine(self):
|
|
self.assertEqual(self.engines["postgres"], "postgres")
|
|
self.assertEqual(self.engines["mariadb"], "mariadb")
|
|
|
|
def test_both_volume_spellings_resolve(self):
|
|
self.assertEqual(self.engines["zammad"], "postgres")
|
|
self.assertEqual(self.engines["zammad_database"], "postgres")
|
|
self.assertEqual(self.engines["nextcloud_database"], "mariadb")
|
|
|
|
def test_an_app_without_a_database_contributes_nothing(self):
|
|
self.assertNotIn("static", self.engines)
|
|
|
|
def test_engine_of_prefers_the_volume_then_the_database(self):
|
|
dump = databases.Dump("nextcloud_database", "nextcloud", Path("/x"))
|
|
self.assertEqual(databases.engine_of(dump, self.engines), "mariadb")
|
|
by_name = databases.Dump("unknown_volume", "zammad", Path("/x"))
|
|
self.assertEqual(databases.engine_of(by_name, self.engines), "postgres")
|
|
|
|
def test_an_unknown_dump_aborts_instead_of_guessing(self):
|
|
dump = databases.Dump("mystery_data", "mystery", Path("/x"))
|
|
with self.assertRaises(databases.RecoveryError):
|
|
databases.engine_of(dump, self.engines)
|
|
|
|
def test_the_real_repository_resolves_its_own_apps(self):
|
|
engines = databases.engine_by_key()
|
|
self.assertEqual(engines["postgres"], "postgres")
|
|
self.assertEqual(engines["zammad"], "postgres")
|
|
|
|
|
|
class TestLayoutFollowsBaudolo(TestCase):
|
|
"""baudolo writes the generation, so core must read exactly what it writes."""
|
|
|
|
def setUp(self):
|
|
self.backups = Path(tempfile.mkdtemp())
|
|
self.paths = BackupPaths(
|
|
"postgres",
|
|
"abc123",
|
|
"20260816190906",
|
|
repo_name="a-repo",
|
|
backups_dir=str(self.backups),
|
|
)
|
|
for written in (self.paths.sql_file("zammad"), self.paths.cluster_file("all")):
|
|
Path(written).parent.mkdir(parents=True, exist_ok=True)
|
|
Path(written).write_text("")
|
|
self.generation = Path(self.paths.root()).parent
|
|
|
|
def test_generation_of_recovers_what_backup_paths_composed(self):
|
|
parts = databases.generation_of(self.generation)
|
|
self.assertEqual(parts.backups_dir, str(self.backups))
|
|
self.assertEqual(parts.machine_hash, "abc123")
|
|
self.assertEqual(parts.repo_name, "a-repo")
|
|
self.assertEqual(parts.name, "20260816190906")
|
|
|
|
def test_dumps_of_finds_the_files_backup_paths_names(self):
|
|
dumps, clusters = databases.dumps_of(self.generation)
|
|
self.assertEqual(
|
|
[(d.volume, d.database) for d in dumps], [("postgres", "zammad")]
|
|
)
|
|
self.assertEqual(
|
|
[(c.volume, c.instance) for c in clusters], [("postgres", "all")]
|
|
)
|
|
|
|
|
|
class TestGenerationLayout(TestCase):
|
|
def setUp(self):
|
|
self.root = Path(tempfile.mkdtemp())
|
|
self.generation = generation(self.root)
|
|
(self.generation / "postgres/sql").mkdir(parents=True)
|
|
(self.generation / "postgres/sql/zammad.backup.sql").write_text("")
|
|
(self.generation / "postgres/sql/all.cluster.backup.sql").write_text("")
|
|
(self.generation / "openldap_data/files").mkdir(parents=True)
|
|
|
|
def test_parts_come_from_the_path(self):
|
|
parts = databases.generation_of(self.generation)
|
|
self.assertEqual(parts.backups_dir, str(self.root / "Backups"))
|
|
self.assertEqual(parts.machine_hash, "abc123")
|
|
self.assertEqual(parts.repo_name, "backup-docker-to-local")
|
|
self.assertEqual(parts.name, "20260816190906")
|
|
|
|
def test_each_dump_is_sorted_into_the_subcommand_that_replays_it(self):
|
|
dumps, clusters = databases.dumps_of(self.generation)
|
|
self.assertEqual(
|
|
[(d.volume, d.database) for d in dumps], [("postgres", "zammad")]
|
|
)
|
|
self.assertEqual(
|
|
[(c.volume, c.instance) for c in clusters], [("postgres", "all")]
|
|
)
|
|
|
|
def test_cluster_argv_matches_the_baudolo_contract(self):
|
|
cluster = databases.Cluster("postgres", "central-postgres", Path("/x"))
|
|
argv = databases.cluster_argv(
|
|
cluster,
|
|
databases.generation_of(self.generation),
|
|
"postgres-1",
|
|
"postgres",
|
|
"s3cret",
|
|
)
|
|
self.assertEqual(
|
|
argv[:5],
|
|
["baudolo-restore", "cluster", "postgres", "abc123", "20260816190906"],
|
|
)
|
|
self.assertEqual(argv[argv.index("--instance") + 1], "central-postgres")
|
|
self.assertEqual(argv[argv.index("--db-user") + 1], "postgres")
|
|
self.assertIn("--empty", argv)
|
|
self.assertNotIn(
|
|
"--db-name", argv, "a cluster dump names an instance, not a database"
|
|
)
|
|
|
|
def test_restore_argv_matches_the_baudolo_contract(self):
|
|
dump = databases.Dump("postgres", "zammad", Path("/x"))
|
|
argv = databases.restore_argv(
|
|
dump,
|
|
databases.generation_of(self.generation),
|
|
"postgres",
|
|
"postgres-1",
|
|
"zammad",
|
|
"s3cret",
|
|
)
|
|
self.assertEqual(
|
|
argv[:5],
|
|
[
|
|
"baudolo-restore",
|
|
"postgres",
|
|
"postgres",
|
|
"abc123",
|
|
"20260816190906",
|
|
],
|
|
)
|
|
self.assertIn("--empty", argv)
|
|
self.assertEqual(argv[argv.index("--container") + 1], "postgres-1")
|
|
self.assertEqual(argv[argv.index("--db-name") + 1], "zammad")
|
|
|
|
|
|
POSTGRES_HEADER = """--
|
|
-- PostgreSQL database dump
|
|
--
|
|
|
|
\\restrict BbyzwODc1rWKL3rDyLhEjgCF0Kf2TU5ma7gcTs8eQI7copLtydXkc61zdULsPav
|
|
|
|
-- Dumped from database version 17.11
|
|
-- Dumped by pg_dump version 17.11
|
|
|
|
SET statement_timeout = 0;
|
|
"""
|
|
|
|
MARIADB_HEADER = """/*M!999999\\- enable the sandbox mode */
|
|
-- MariaDB dump 10.19-11.8.8-MariaDB, for debian-linux-gnu (x86_64)
|
|
--
|
|
-- Host: 127.0.0.1 Database: mysql
|
|
-- ------------------------------------------------------
|
|
-- Server version\t11.8.8-MariaDB-ubu2404
|
|
|
|
/*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
|
|
"""
|
|
|
|
|
|
CLUSTER_DUMP = """--
|
|
-- PostgreSQL database cluster dump
|
|
--
|
|
|
|
CREATE ROLE greenlight;
|
|
ALTER ROLE greenlight WITH LOGIN;
|
|
|
|
\\connect template1
|
|
|
|
\\connect postgres
|
|
|
|
\\connect greenlight
|
|
|
|
-- Dumped from database version 17.11
|
|
CREATE TABLE t (v text);
|
|
|
|
\\connect keycloak
|
|
|
|
-- Dumped from database version 17.11
|
|
CREATE TABLE t (v text);
|
|
"""
|
|
|
|
|
|
class TestClusterDumps(TestCase):
|
|
def cluster(self, text=CLUSTER_DUMP):
|
|
path = Path(tempfile.mkdtemp()) / "central-postgres.cluster.backup.sql"
|
|
path.write_text(text, encoding="utf-8")
|
|
return databases.Cluster("postgres", "central-postgres", path)
|
|
|
|
def test_the_connect_lines_are_the_inventory(self):
|
|
self.assertEqual(
|
|
databases.databases_in(self.cluster()), ["greenlight", "keycloak"]
|
|
)
|
|
|
|
def test_the_control_databases_are_not_applications(self):
|
|
found = databases.databases_in(self.cluster())
|
|
for control in ("postgres", "template0", "template1"):
|
|
self.assertNotIn(control, found)
|
|
|
|
def test_a_database_created_but_never_reconnected_still_counts(self):
|
|
"""pg_dumpall names a database twice; a stream that only creates it
|
|
would otherwise leave it unseeded and the drill would pass regardless."""
|
|
cluster = self.cluster('CREATE DATABASE "late" WITH TEMPLATE = template0;\n')
|
|
self.assertEqual(databases.databases_in(cluster), ["late"])
|
|
|
|
def test_a_quoted_name_loses_its_quotes(self):
|
|
cluster = self.cluster('\\connect "odd-name"\n')
|
|
self.assertEqual(databases.databases_in(cluster), ["odd-name"])
|
|
|
|
def test_a_quoted_name_keeps_its_spaces(self):
|
|
cluster = self.cluster('\\connect "odd name"\n')
|
|
self.assertEqual(databases.databases_in(cluster), ["odd name"])
|
|
|
|
def test_psql_options_are_not_mistaken_for_the_database(self):
|
|
cluster = self.cluster("\\connect -reuse-previous=on dbname=appdb\n")
|
|
self.assertEqual(databases.databases_in(cluster), ["appdb"])
|
|
|
|
def test_a_database_reconnected_twice_is_listed_once(self):
|
|
cluster = self.cluster("\\connect app\nSELECT 1;\n\\connect app\n")
|
|
self.assertEqual(databases.databases_in(cluster), ["app"])
|
|
|
|
|
|
class TestCredentials(TestCase):
|
|
def write(self, text):
|
|
path = Path(tempfile.mkdtemp()) / "databases.csv"
|
|
path.write_text(text, encoding="utf-8")
|
|
return path
|
|
|
|
def test_first_row_per_database_wins(self):
|
|
path = self.write(
|
|
"instance;database;username;password\n"
|
|
"postgres;zammad;zammad;secret\n"
|
|
"postgres;zammad;other;later\n"
|
|
"\n"
|
|
)
|
|
self.assertEqual(
|
|
databases.credentials_of(path), {"zammad": ("zammad", "secret")}
|
|
)
|
|
|
|
def test_short_row_aborts(self):
|
|
path = self.write("instance;database;username;password\npostgres;zammad\n")
|
|
with self.assertRaises(databases.RecoveryError):
|
|
databases.credentials_of(path)
|
|
|
|
def test_missing_file_aborts(self):
|
|
with self.assertRaises(databases.RecoveryError):
|
|
databases.credentials_of(Path("/nonexistent/databases.csv"))
|
|
|
|
def test_a_cluster_row_is_keyed_by_its_instance(self):
|
|
path = self.write(
|
|
"instance;database;username;password\n"
|
|
"central-postgres;*;postgres;super\n"
|
|
"bbb-postgres;*;postgres;other\n"
|
|
"central-postgres;zammad;zammad;secret\n"
|
|
)
|
|
self.assertEqual(
|
|
databases.cluster_credentials_of(path),
|
|
{
|
|
"central-postgres": ("postgres", "super"),
|
|
"bbb-postgres": ("postgres", "other"),
|
|
},
|
|
)
|
|
|
|
def test_two_instances_would_collide_under_the_database_key(self):
|
|
path = self.write(
|
|
"instance;database;username;password\n"
|
|
"central-postgres;*;postgres;super\n"
|
|
"bbb-postgres;*;postgres;other\n"
|
|
)
|
|
self.assertEqual(databases.credentials_of(path), {"*": ("postgres", "super")})
|
|
|
|
|
|
class TestReplay(TestCase):
|
|
def setUp(self):
|
|
self.root = Path(tempfile.mkdtemp())
|
|
self.generation = generation(self.root)
|
|
(self.generation / "postgres/sql").mkdir(parents=True)
|
|
(self.generation / "postgres/sql/zammad.backup.sql").write_text(POSTGRES_HEADER)
|
|
self.csv = self.root / "databases.csv"
|
|
self.csv.write_text(
|
|
"instance;database;username;password\npostgres;zammad;zammad;s3cret\n",
|
|
encoding="utf-8",
|
|
)
|
|
self.engines = {"postgres": "postgres"}
|
|
|
|
def replay(self, running):
|
|
calls = []
|
|
|
|
def fake_run(argv, secret=""):
|
|
calls.append(argv)
|
|
return ""
|
|
|
|
with (
|
|
mock.patch.object(recovery_docker, "_run", fake_run),
|
|
mock.patch.object(
|
|
recovery_docker, "consumers_running", lambda *a, **k: running
|
|
),
|
|
mock.patch.object(
|
|
recovery_docker, "container_of_volume", lambda *a, **k: "postgres-1"
|
|
),
|
|
):
|
|
replayed = databases.replay(self.generation, self.csv, engines=self.engines)
|
|
return replayed, calls
|
|
|
|
def test_a_running_consumer_aborts_before_anything_is_restored(self):
|
|
with self.assertRaises(databases.RecoveryError) as raised:
|
|
self.replay(["zammad-railsserver"])
|
|
self.assertIn("zammad-railsserver", str(raised.exception))
|
|
|
|
def test_the_engine_itself_does_not_block_its_own_replay(self):
|
|
replayed, _ = self.replay(["postgres-1"])
|
|
self.assertEqual(replayed, 1)
|
|
|
|
def test_a_consumer_beside_the_engine_still_aborts(self):
|
|
with self.assertRaises(databases.RecoveryError) as raised:
|
|
self.replay(["postgres-1", "zammad-railsserver"])
|
|
self.assertIn("zammad-railsserver", str(raised.exception))
|
|
self.assertNotIn("postgres-1", str(raised.exception))
|
|
|
|
def test_a_quiesced_host_replays_the_dump(self):
|
|
replayed, calls = self.replay([])
|
|
self.assertEqual(replayed, 1)
|
|
restores = [argv for argv in calls if argv[0] == "baudolo-restore"]
|
|
self.assertEqual(len(restores), 1)
|
|
self.assertEqual(restores[0][:2], ["baudolo-restore", "postgres"])
|
|
|
|
def test_the_version_check_is_left_to_baudolo(self):
|
|
_, calls = self.replay([])
|
|
self.assertEqual(
|
|
[argv for argv in calls if "exec" in argv],
|
|
[],
|
|
"the replay must not query the engine version itself",
|
|
)
|
|
|
|
def test_a_generation_without_dumps_is_not_an_error(self):
|
|
empty = generation(Path(tempfile.mkdtemp()))
|
|
self.assertEqual(databases.replay(empty, self.csv, engines=self.engines), 0)
|
|
|
|
def test_a_missing_generation_aborts(self):
|
|
with self.assertRaises(databases.RecoveryError):
|
|
databases.replay(self.generation / "nope", self.csv, engines=self.engines)
|
|
|
|
def test_a_stopped_database_container_says_to_start_it(self):
|
|
with mock.patch.object(
|
|
recovery_docker, "_run", lambda argv, secret="": "postgres\n"
|
|
):
|
|
self.assertEqual(
|
|
recovery_docker.container_of_volume("postgres"), "postgres"
|
|
)
|
|
|
|
def only_ps_all(argv, secret=""):
|
|
return "postgres\n" if "-a" in argv else ""
|
|
|
|
with (
|
|
mock.patch.object(recovery_docker, "_run", only_ps_all),
|
|
self.assertRaises(databases.RecoveryError) as raised,
|
|
):
|
|
recovery_docker.container_of_volume("postgres")
|
|
self.assertIn("is not running", str(raised.exception))
|
|
|
|
def test_no_container_at_all_says_to_deploy_first(self):
|
|
with (
|
|
mock.patch.object(recovery_docker, "_run", lambda argv, secret="": ""),
|
|
self.assertRaises(databases.RecoveryError) as raised,
|
|
):
|
|
recovery_docker.container_of_volume("postgres")
|
|
self.assertIn("deploy the stack first", str(raised.exception))
|
|
|
|
def test_a_cluster_dump_is_replayed_as_a_whole(self):
|
|
(self.generation / "postgres/sql/bbb-postgres.cluster.backup.sql").write_text(
|
|
CLUSTER_DUMP, encoding="utf-8"
|
|
)
|
|
self.csv.write_text(
|
|
"instance;database;username;password\n"
|
|
"postgres;zammad;zammad;s3cret\n"
|
|
"bbb-postgres;*;postgres;super\n",
|
|
encoding="utf-8",
|
|
)
|
|
replayed, calls = self.replay([])
|
|
self.assertEqual(replayed, 2)
|
|
subcommands = [argv[1] for argv in calls if argv[0] == "baudolo-restore"]
|
|
self.assertEqual(
|
|
subcommands,
|
|
["cluster", "postgres"],
|
|
"the cluster goes in first; a single-database dump of the same "
|
|
"instance is the more specific statement and must land after it",
|
|
)
|
|
|
|
def test_a_cluster_without_a_superuser_row_aborts(self):
|
|
(self.generation / "postgres/sql/bbb-postgres.cluster.backup.sql").write_text(
|
|
CLUSTER_DUMP, encoding="utf-8"
|
|
)
|
|
with self.assertRaises(databases.RecoveryError) as raised:
|
|
self.replay([])
|
|
self.assertIn("bbb-postgres", str(raised.exception))
|
|
|
|
def test_a_consumer_of_a_clustered_database_aborts_the_replay(self):
|
|
(self.generation / "postgres/sql/bbb-postgres.cluster.backup.sql").write_text(
|
|
CLUSTER_DUMP, encoding="utf-8"
|
|
)
|
|
self.csv.write_text(
|
|
"instance;database;username;password\n"
|
|
"postgres;zammad;zammad;s3cret\n"
|
|
"bbb-postgres;*;postgres;super\n",
|
|
encoding="utf-8",
|
|
)
|
|
with (
|
|
mock.patch.object(recovery_docker, "_run", lambda argv, secret="": ""),
|
|
mock.patch.object(
|
|
recovery_docker,
|
|
"consumers_running",
|
|
lambda project, *a, **k: (
|
|
["greenlight-web"] if project == "greenlight" else []
|
|
),
|
|
),
|
|
mock.patch.object(
|
|
recovery_docker, "container_of_volume", lambda *a, **k: "postgres-1"
|
|
),
|
|
self.assertRaises(databases.RecoveryError) as raised,
|
|
):
|
|
databases.replay(self.generation, self.csv, engines=self.engines)
|
|
self.assertIn("greenlight-web", str(raised.exception))
|
|
|
|
def cluster_only(self, running_in_project):
|
|
"""A generation holding nothing but one instance's cluster dump."""
|
|
(self.generation / "postgres/sql/zammad.backup.sql").unlink()
|
|
(self.generation / "postgres/sql/bbb-postgres.cluster.backup.sql").write_text(
|
|
CLUSTER_DUMP, encoding="utf-8"
|
|
)
|
|
self.csv.write_text(
|
|
"instance;database;username;password\nbbb-postgres;*;postgres;super\n",
|
|
encoding="utf-8",
|
|
)
|
|
return (
|
|
mock.patch.object(
|
|
recovery_docker,
|
|
"_run",
|
|
lambda argv, secret="": "bigbluebutton\n" if "inspect" in argv else "",
|
|
),
|
|
mock.patch.object(
|
|
recovery_docker,
|
|
"consumers_running",
|
|
lambda project, *a, **k: running_in_project.get(project, []),
|
|
),
|
|
mock.patch.object(
|
|
recovery_docker, "container_of_volume", lambda *a, **k: "bbb-postgres-1"
|
|
),
|
|
)
|
|
|
|
def test_a_consumer_in_the_engines_own_project_aborts_the_cluster_replay(self):
|
|
patches = self.cluster_only({"bigbluebutton": ["bigbluebutton-etherpad-1"]})
|
|
with (
|
|
patches[0],
|
|
patches[1],
|
|
patches[2],
|
|
self.assertRaises(databases.RecoveryError) as raised,
|
|
):
|
|
databases.replay(self.generation, self.csv, engines=self.engines)
|
|
self.assertIn("bigbluebutton-etherpad-1", str(raised.exception))
|
|
|
|
def test_the_engine_is_not_counted_as_its_own_consumer(self):
|
|
patches = self.cluster_only({"bigbluebutton": ["bbb-postgres-1"]})
|
|
with patches[0], patches[1], patches[2]:
|
|
self.assertEqual(
|
|
databases.replay(self.generation, self.csv, engines=self.engines), 1
|
|
)
|
|
|
|
def test_a_dump_only_the_manifest_can_place_is_replayed(self):
|
|
root = Path(tempfile.mkdtemp())
|
|
gen = generation(root)
|
|
(gen / "postgres_data/sql").mkdir(parents=True)
|
|
(gen / "postgres_data/sql/mautrix_meta_bridge.backup.sql").write_text(
|
|
POSTGRES_HEADER
|
|
)
|
|
(gen / MANIFEST_FILE).write_text(
|
|
json.dumps(
|
|
{
|
|
"schema": MANIFEST_SCHEMA,
|
|
"volumes": {
|
|
"postgres_data": {
|
|
"database": True,
|
|
"dumped": True,
|
|
"engine": "postgres",
|
|
}
|
|
},
|
|
}
|
|
)
|
|
)
|
|
csv = root / "databases.csv"
|
|
csv.write_text(
|
|
"instance;database;username;password\n"
|
|
"postgres;mautrix_meta_bridge;bridge;pw\n",
|
|
encoding="utf-8",
|
|
)
|
|
with (
|
|
mock.patch.object(recovery_docker, "_run", lambda argv, secret="": ""),
|
|
mock.patch.object(recovery_docker, "consumers_running", lambda *a, **k: []),
|
|
mock.patch.object(
|
|
recovery_docker, "container_of_volume", lambda *a, **k: "postgres-1"
|
|
),
|
|
):
|
|
self.assertEqual(databases.replay(gen, csv), 1)
|
|
|
|
def test_the_password_is_redacted_from_a_failure(self):
|
|
with self.assertRaises(databases.RecoveryError) as raised:
|
|
recovery_docker._run(["false", "--db-password", "s3cret"], secret="s3cret")
|
|
self.assertNotIn("s3cret", str(raised.exception))
|
|
self.assertIn("***", str(raised.exception))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|