diff --git a/README.md b/README.md index 4ab9356..407dc37 100644 --- a/README.md +++ b/README.md @@ -47,9 +47,13 @@ Backups are stored in a deterministic, fully nested structure: └── / ├── files/ └── sql/ - └── .backup.sql + └── .backup.sql.gz ``` +Dumps are gzip-compressed while they are written; no uncompressed copy +touches the disk. Generations written before compression hold +`.backup.sql`; `baudolo-restore` reads both. + ### Meaning of each level * `` diff --git a/src/baudolo/backup/db.py b/src/baudolo/backup/db.py index d4ea281..8636854 100644 --- a/src/baudolo/backup/db.py +++ b/src/baudolo/backup/db.py @@ -9,7 +9,7 @@ from baudolo.databases import CLUSTER_ROW, validate_database from baudolo.generation import CLUSTER_SUFFIX, DUMP_SUFFIX, SQL_DIR from .docker import docker_exec_argv -from .shell import BackupError, execute_to_file +from .shell import BackupError, execute_to_gzip if TYPE_CHECKING: import pandas as pd @@ -53,7 +53,7 @@ def fallback_pg_dumpall( """ Perform a full Postgres cluster dump using pg_dumpall. """ - execute_to_file( + execute_to_gzip( docker_exec_argv( container, ["pg_dumpall", "-U", username, "-h", "localhost"], @@ -122,7 +122,7 @@ def backup_database( if db_type == "mariadb": # Force TCP so auth matches ''@'%' instead of socket -> 'localhost'. - execute_to_file( + execute_to_gzip( docker_exec_argv( container, [ @@ -144,7 +144,7 @@ def backup_database( if db_type == "postgres": try: - execute_to_file( + execute_to_gzip( docker_exec_argv( container, [ diff --git a/src/baudolo/backup/shell.py b/src/baudolo/backup/shell.py index 400a436..07f8b31 100644 --- a/src/baudolo/backup/shell.py +++ b/src/baudolo/backup/shell.py @@ -7,14 +7,22 @@ string handed to ``shell=True`` allowed. from __future__ import annotations +import gzip import os import subprocess +import tempfile +import zlib +from functools import partial from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: from collections.abc import Mapping, Sequence +GZIP_LEVEL = 6 +READ_BYTES = 1 << 20 +SYNC_BYTES = 1 << 16 + class BackupError(Exception): """Generic exception for backup errors.""" @@ -53,23 +61,71 @@ def execute_shell_command( return [line.decode("utf-8") for line in out.splitlines()] -def execute_to_file( +def is_sync_point(piece: bytes) -> bool: + """ + Args: + piece: a line of the dump, or a READ_BYTES slice of a longer one. + + Returns: + True when a full flush follows *piece*, decided by its bytes alone, on + average once per SYNC_BYTES of input. + """ + return zlib.crc32(piece) % max(1, SYNC_BYTES // len(piece)) == 0 + + +def _abort(process: subprocess.Popen, tmp: Path) -> None: + process.kill() + process.wait() + tmp.unlink(missing_ok=True) + + +def execute_to_gzip( command: Sequence[str], out_file: str, *, env: Mapping[str, str] | None = None ) -> None: - """Run *command*, writing its stdout to *out_file* only once it succeeded. + """ + Args: + command: argv, the program first; its stdout is the dump. + out_file: the final gzip path. The stream is written to + ``.tmp`` and renamed only after the command exited 0. + env: variables added to the child's environment. - The output goes to a sibling temporary file first, so a partial or empty - stream from a failing dump never takes the place of a valid backup. + Raises: + BackupError: the command failed, or the file could not be written. """ command = list(command) print(" ".join(command), flush=True) tmp = Path(f"{out_file}.tmp") - with tmp.open("wb") as handle: + with tempfile.TemporaryFile() as err_sink: process = subprocess.Popen( - command, stdout=handle, stderr=subprocess.PIPE, env=_child_env(env) + command, stdout=subprocess.PIPE, stderr=err_sink, env=_child_env(env) ) - _, err = process.communicate() - if process.returncode != 0: + try: + with ( + tmp.open("wb") as raw, + gzip.GzipFile( + filename="", + mode="wb", + fileobj=raw, + compresslevel=GZIP_LEVEL, + mtime=0, + ) as packed, + ): + for piece in iter(partial(process.stdout.readline, READ_BYTES), b""): + packed.write(piece) + if is_sync_point(piece): + packed.flush(zlib.Z_FULL_FLUSH) + except OSError as error: + _abort(process, tmp) + raise BackupError(f"Error writing {out_file}: {error}") from error + except BaseException: + _abort(process, tmp) + raise + finally: + process.stdout.close() + returncode = process.wait() + err_sink.seek(0) + err = err_sink.read() + if returncode != 0: tmp.unlink() - _fail(command, process.returncode, b"", err) + _fail(command, returncode, b"", err) tmp.replace(out_file) diff --git a/src/baudolo/generation.py b/src/baudolo/generation.py index 65af72f..e0844b8 100644 --- a/src/baudolo/generation.py +++ b/src/baudolo/generation.py @@ -18,8 +18,12 @@ from __future__ import annotations FILES_DIR = "files" SQL_DIR = "sql" -DUMP_SUFFIX = ".backup.sql" -CLUSTER_SUFFIX = ".cluster.backup.sql" +LEGACY_DUMP_SUFFIX = ".backup.sql" +LEGACY_CLUSTER_SUFFIX = ".cluster.backup.sql" +COMPRESSED_SUFFIX = ".gz" +DUMP_COMPRESSION = "gzip" +DUMP_SUFFIX = LEGACY_DUMP_SUFFIX + COMPRESSED_SUFFIX +CLUSTER_SUFFIX = LEGACY_CLUSTER_SUFFIX + COMPRESSED_SUFFIX MANIFEST_FILE = "manifest.json" MANIFEST_SCHEMA = 1 @@ -42,6 +46,7 @@ def manifest_document(volumes: dict[str, object]) -> dict[str, object]: "sql_dir": SQL_DIR, "dump_suffix": DUMP_SUFFIX, "cluster_suffix": CLUSTER_SUFFIX, + "dump_compression": DUMP_COMPRESSION, }, "volumes": { name: { diff --git a/src/baudolo/restore/__main__.py b/src/baudolo/restore/__main__.py index c9bfae5..e597dc7 100644 --- a/src/baudolo/restore/__main__.py +++ b/src/baudolo/restore/__main__.py @@ -6,6 +6,7 @@ import sys from .db.cluster import restore_cluster_sql from .db.mariadb import restore_mariadb_sql from .db.postgres import restore_postgres_sql +from .dumpfile import locate from .files import restore_volume_files from .paths import BackupPaths @@ -106,7 +107,7 @@ def main(argv: list[str] | None = None) -> int: p_cluster.add_argument( "--instance", required=True, - help="Instance the dump was taken from; names .cluster.backup.sql", + help="Instance the dump was taken from; names .cluster.backup.sql.gz", ) p_cluster.add_argument( "--db-user", @@ -149,13 +150,15 @@ def main(argv: list[str] | None = None) -> int: db_name=args.db_name, user=user, password=password, - sql_path=BackupPaths( - args.volume_name, - args.backup_hash, - args.version, - repo_name=args.repo_name, - backups_dir=args.backups_dir, - ).sql_file(args.db_name), + sql_path=locate( + BackupPaths( + args.volume_name, + args.backup_hash, + args.version, + repo_name=args.repo_name, + backups_dir=args.backups_dir, + ).sql_file(args.db_name) + ), empty=args.empty, check_version=not args.no_version_check, ) @@ -166,13 +169,15 @@ def main(argv: list[str] | None = None) -> int: container=args.container, user=args.db_user, password=password, - sql_path=BackupPaths( - args.volume_name, - args.backup_hash, - args.version, - repo_name=args.repo_name, - backups_dir=args.backups_dir, - ).cluster_file(args.instance), + sql_path=locate( + BackupPaths( + args.volume_name, + args.backup_hash, + args.version, + repo_name=args.repo_name, + backups_dir=args.backups_dir, + ).cluster_file(args.instance) + ), empty=args.empty, check_version=not args.no_version_check, ) @@ -185,13 +190,15 @@ def main(argv: list[str] | None = None) -> int: db_name=args.db_name, user=user, password=password, - sql_path=BackupPaths( - args.volume_name, - args.backup_hash, - args.version, - repo_name=args.repo_name, - backups_dir=args.backups_dir, - ).sql_file(args.db_name), + sql_path=locate( + BackupPaths( + args.volume_name, + args.backup_hash, + args.version, + repo_name=args.repo_name, + backups_dir=args.backups_dir, + ).sql_file(args.db_name) + ), empty=args.empty, check_version=not args.no_version_check, ) diff --git a/src/baudolo/restore/db/cluster.py b/src/baudolo/restore/db/cluster.py index 928a1cd..5896581 100644 --- a/src/baudolo/restore/db/cluster.py +++ b/src/baudolo/restore/db/cluster.py @@ -24,6 +24,7 @@ import tempfile from pathlib import Path from typing import TYPE_CHECKING +from baudolo.restore.dumpfile import open_dump from baudolo.restore.run import docker_exec from .version import guard @@ -96,7 +97,7 @@ def dump_inventory(sql_path: str) -> tuple[list[str], list[str]]: """ databases: list[str] = [] roles: list[str] = [] - with Path(sql_path).open("rb") as handle: + with open_dump(sql_path) as handle: for raw in handle: line = raw.decode("utf-8", "replace") for pattern, sink, read in ( @@ -212,7 +213,8 @@ def restore_cluster_sql( user: a superuser of that instance; the dump creates roles and databases, which an application role may not do. password: its password, handed to psql through the container's env. - sql_path: the ``.cluster.backup.sql`` of a generation. + sql_path: the ``.cluster.backup.sql.gz`` of a generation, or + the uncompressed ``.cluster.backup.sql`` of an older one. empty: drop the cluster's databases and roles first. Without it the replay stops at the first object that already exists, which is the honest outcome: recreating a cluster over a populated one is a @@ -243,7 +245,7 @@ def restore_cluster_sql( docker_env=docker_env, ) - with Path(sql_path).open("rb") as src, tempfile.TemporaryFile() as filtered: + with open_dump(sql_path) as src, tempfile.TemporaryFile() as filtered: for line in filter_own_role_creation(src, user): filtered.write(line) filtered.seek(0) diff --git a/src/baudolo/restore/db/mariadb.py b/src/baudolo/restore/db/mariadb.py index 6f63277..5085049 100644 --- a/src/baudolo/restore/db/mariadb.py +++ b/src/baudolo/restore/db/mariadb.py @@ -3,7 +3,8 @@ from __future__ import annotations import sys from pathlib import Path -from baudolo.restore.run import docker_exec, docker_exec_sh +from baudolo.restore.dumpfile import open_dump, verify +from baudolo.restore.run import docker_exec, docker_exec_sh, docker_exec_stream from .version import guard @@ -47,6 +48,7 @@ def restore_mariadb_sql( if not Path(sql_path).is_file(): raise FileNotFoundError(sql_path) + verify(sql_path) docker_env = {"MYSQL_PWD": password} if check_version: @@ -98,9 +100,12 @@ def restore_mariadb_sql( docker_env=docker_env, ) - with Path(sql_path).open("rb") as f: - docker_exec( - container, [client, "-u", user, db_name], stdin=f, docker_env=docker_env + with open_dump(sql_path) as source: + docker_exec_stream( + container, + [client, "-u", user, db_name], + source=source, + docker_env=docker_env, ) print(f"MariaDB/MySQL restore complete for db '{db_name}'.") diff --git a/src/baudolo/restore/db/postgres.py b/src/baudolo/restore/db/postgres.py index a4013dd..48fbb50 100644 --- a/src/baudolo/restore/db/postgres.py +++ b/src/baudolo/restore/db/postgres.py @@ -4,6 +4,7 @@ import tempfile from pathlib import Path from typing import TYPE_CHECKING +from baudolo.restore.dumpfile import open_dump from baudolo.restore.run import docker_exec from .version import guard @@ -67,23 +68,22 @@ def restore_postgres_sql( docker_env = {"PGPASSWORD": password} - if empty: - with _EMPTY_PRECLEAN_SQL.open(encoding="utf-8") as preclean: - drop_sql = preclean.read() - docker_exec( - container, - ["psql", "-v", "ON_ERROR_STOP=1", "-U", user, "-d", db_name], - stdin=drop_sql.encode(), - docker_env=docker_env, - ) - # Filter into a spooled temp file instead of building the whole dump in # memory: production dumps reach many GB and the previous read/splitlines/ # join needed roughly three times the dump size in RSS. - with Path(sql_path).open("rb") as src, tempfile.TemporaryFile() as filtered: + with open_dump(sql_path) as src, tempfile.TemporaryFile() as filtered: for line in filter_superuser_only_lines(src): filtered.write(line) filtered.seek(0) + if empty: + with _EMPTY_PRECLEAN_SQL.open(encoding="utf-8") as preclean: + drop_sql = preclean.read() + docker_exec( + container, + ["psql", "-v", "ON_ERROR_STOP=1", "-U", user, "-d", db_name], + stdin=drop_sql.encode(), + docker_env=docker_env, + ) docker_exec( container, [ diff --git a/src/baudolo/restore/db/version.py b/src/baudolo/restore/db/version.py index 19a7388..7ee323c 100644 --- a/src/baudolo/restore/db/version.py +++ b/src/baudolo/restore/db/version.py @@ -23,9 +23,10 @@ with the cluster banner and the roles section, and the first from __future__ import annotations +import io import re -from pathlib import Path +from baudolo.restore.dumpfile import open_dump from baudolo.restore.run import docker_exec, stdout_of SCAN_LINES = 2000 @@ -69,7 +70,9 @@ def dump_version(sql_path: str, engine: str) -> str: VersionMismatchError: no version line within the first ``SCAN_LINES``. """ pattern = DUMP_VERSION[engine] - with Path(sql_path).open(encoding="utf-8", errors="replace") as handle: + with io.TextIOWrapper( + open_dump(sql_path), encoding="utf-8", errors="replace" + ) as handle: for _ in range(SCAN_LINES): line = handle.readline() if not line: diff --git a/src/baudolo/restore/dumpfile.py b/src/baudolo/restore/dumpfile.py new file mode 100644 index 0000000..a87bde1 --- /dev/null +++ b/src/baudolo/restore/dumpfile.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import gzip +from pathlib import Path +from typing import BinaryIO + +from baudolo.generation import COMPRESSED_SUFFIX + +GZIP_MAGIC = b"\x1f\x8b" +VERIFY_BYTES = 1 << 20 + + +def is_gzip(path: str) -> bool: + """ + Args: + path: a dump file. + + Returns: + True when its first bytes are the gzip magic, whatever its name says. + """ + with Path(path).open("rb") as handle: + return handle.read(len(GZIP_MAGIC)) == GZIP_MAGIC + + +def open_dump(path: str) -> BinaryIO: + """ + Args: + path: a dump, gzip-compressed or plain. + + Returns: + A binary reader yielding the SQL text either way. + """ + if is_gzip(path): + return gzip.open(path, "rb") + return Path(path).open("rb") + + +def verify(path: str) -> None: + """ + Args: + path: a dump; a gzip one is decompressed to its end, a plain one is + left alone. + + Raises: + EOFError: the gzip stream is truncated. + gzip.BadGzipFile: its header, CRC or length is wrong. + zlib.error: its deflate data is corrupt. + """ + if not is_gzip(path): + return + with gzip.open(path, "rb") as handle: + while handle.read(VERIFY_BYTES): + pass + + +def locate(path: str) -> str: + """ + Args: + path: where this version writes the dump, ending in COMPRESSED_SUFFIX. + + Returns: + *path*, or the uncompressed name earlier versions wrote, whichever + exists. + + Raises: + FileNotFoundError: neither exists. + FileExistsError: both exist. + """ + legacy = path.removesuffix(COMPRESSED_SUFFIX) + found = [ + candidate + for candidate in dict.fromkeys((path, legacy)) + if Path(candidate).is_file() + ] + if len(found) > 1: + raise FileExistsError( + f"both {path} and {legacy} exist; remove the one that must not be replayed" + ) + if not found: + raise FileNotFoundError(f"neither {path} nor {legacy} exists") + return found[0] diff --git a/src/baudolo/restore/run.py b/src/baudolo/restore/run.py index 105d353..b1da834 100644 --- a/src/baudolo/restore/run.py +++ b/src/baudolo/restore/run.py @@ -1,8 +1,13 @@ from __future__ import annotations +import contextlib import os +import shutil import subprocess import sys +from typing import BinaryIO + +STREAM_BYTES = 1 << 20 def run( @@ -96,6 +101,43 @@ def docker_exec( ) +def docker_exec_stream( + container: str, + argv: list[str], + *, + source: BinaryIO, + docker_env: dict[str, str] | None = None, +) -> None: + """ + Args: + container: the container to run in. + argv: the command, fed *source* on stdin. + source: a binary reader, copied in STREAM_BYTES chunks, so a reader + without a file descriptor of its own, such as a gzip one, works. + docker_env: as for :func:`docker_exec`. + + Raises: + subprocess.CalledProcessError: the command exited non-zero. + """ + cmd = docker_exec_cmd(container, argv, docker_env) + process = subprocess.Popen( + cmd, stdin=subprocess.PIPE, env=child_env(None, docker_env) + ) + try: + with contextlib.suppress(BrokenPipeError): + shutil.copyfileobj(source, process.stdin, STREAM_BYTES) + with contextlib.suppress(BrokenPipeError): + process.stdin.close() + except BaseException: + process.kill() + process.wait() + raise + returncode = process.wait() + if returncode != 0: + print(f"ERROR: command failed ({returncode}): {' '.join(cmd)}", file=sys.stderr) + raise subprocess.CalledProcessError(returncode, cmd) + + def docker_exec_sh( container: str, script: str, diff --git a/tests/e2e/helpers/fixtures.py b/tests/e2e/helpers/fixtures.py index e96fec7..2829f4e 100644 --- a/tests/e2e/helpers/fixtures.py +++ b/tests/e2e/helpers/fixtures.py @@ -2,10 +2,14 @@ from __future__ import annotations +import gzip import shutil import subprocess from pathlib import Path +from baudolo.generation import DUMP_SUFFIX, LEGACY_DUMP_SUFFIX +from baudolo.restore.dumpfile import open_dump + from .process import machine_hash, run # postgres 18+ mounts at /var/lib/postgresql, not /var/lib/postgresql/data. @@ -112,3 +116,54 @@ def ensure_empty_dir(path: str) -> None: if p.exists(): shutil.rmtree(p) p.mkdir(parents=True, exist_ok=True) + + +def dump_text(path: Path | str) -> str: + """ + Args: + path: a dump, gzip-compressed or plain. + + Returns: + Its SQL text. + """ + with open_dump(str(path)) as handle: + return handle.read().decode("utf-8", errors="replace") + + +def write_dump_text(path: Path | str, text: str) -> None: + """ + Args: + path: a gzip dump to overwrite. + text: the SQL text it carries afterwards. + """ + with gzip.open(str(path), "wb") as handle: + handle.write(text.encode("utf-8")) + + +def dumps_in(sql_dir: Path) -> list[Path]: + """ + Args: + sql_dir: a volume's sql directory, which may be absent. + + Returns: + Every dump in it, compressed or plain, single-database or cluster. + """ + if not sql_dir.is_dir(): + return [] + return sorted( + {*sql_dir.glob(f"*{DUMP_SUFFIX}"), *sql_dir.glob(f"*{LEGACY_DUMP_SUFFIX}")} + ) + + +def assert_dumps_in_sees_both_suffixes(case, scratch: str) -> None: + """ + Args: + case: the running ``unittest.TestCase``. + scratch: an empty directory to place the probe dumps in. + """ + probe = Path(scratch) / "dumps-in-probe" + probe.mkdir(parents=True, exist_ok=True) + expected = [probe / f"a{DUMP_SUFFIX}", probe / f"b{LEGACY_DUMP_SUFFIX}"] + for path in expected: + path.write_bytes(b"") + case.assertEqual(dumps_in(probe), expected) diff --git a/tests/e2e/test_e2e_app_container_ships_client_tools.py b/tests/e2e/test_e2e_app_container_ships_client_tools.py index bdb9f54..1678f18 100644 --- a/tests/e2e/test_e2e_app_container_ships_client_tools.py +++ b/tests/e2e/test_e2e_app_container_ships_client_tools.py @@ -26,6 +26,7 @@ from .helpers import ( backup_run, cleanup_docker, create_minimal_compose_dir, + dump_text, ensure_empty_dir, latest_version_dir, require_docker, @@ -140,7 +141,7 @@ class TestE2EAppContainerShipsClientTools(unittest.TestCase): def test_the_engine_volume_was_dumped(self) -> None: dump = self.volume_dir(self.engine_volume) / SQL_DIR / f"shopdb{DUMP_SUFFIX}" self.assertTrue(dump.is_file(), f"expected a dump at {dump}") - self.assertIn(PAYLOAD, dump.read_text(encoding="utf-8")) + self.assertIn(PAYLOAD, dump_text(dump)) def test_the_application_volume_produced_no_dump(self) -> None: """The collision this module exists for: the application container diff --git a/tests/e2e/test_e2e_engine_detection_by_tool.py b/tests/e2e/test_e2e_engine_detection_by_tool.py index cf00830..d775fde 100644 --- a/tests/e2e/test_e2e_engine_detection_by_tool.py +++ b/tests/e2e/test_e2e_engine_detection_by_tool.py @@ -13,6 +13,8 @@ inside Alpine and takes the whole run with it. import unittest +from baudolo.generation import DUMP_SUFFIX, SQL_DIR + from .helpers import ( POSTGRES_DATA_DIR, POSTGRES_IMAGE, @@ -20,6 +22,8 @@ from .helpers import ( backup_run, cleanup_docker, create_minimal_compose_dir, + dump_text, + dumps_in, ensure_empty_dir, latest_version_dir, require_docker, @@ -142,16 +146,20 @@ class TestE2EEngineDetectionByTool(unittest.TestCase): return backup_path(self.backups_dir, self.repo_name, self.version, volume) def test_an_engine_without_an_engine_name_is_still_dumped(self) -> None: - dump = self._volume_dir(self.engine_volume) / "sql" / "appdb.backup.sql" + dump = self._volume_dir(self.engine_volume) / SQL_DIR / f"appdb{DUMP_SUFFIX}" self.assertTrue( dump.is_file(), f"a Postgres tagged '{self.engine_image}' produced no dump at {dump}", ) - self.assertIn("Dumped by pg_dump", dump.read_text(encoding="utf-8")) + self.assertIn("Dumped by pg_dump", dump_text(dump)) def test_an_engine_name_without_an_engine_is_not_dumped(self) -> None: - sql_dir = self._volume_dir(self.impostor_volume) / "sql" - dumps = list(sql_dir.glob("*.sql")) if sql_dir.exists() else [] + self.assertEqual( + dumps_in(self._volume_dir(self.engine_volume) / SQL_DIR), + [self._volume_dir(self.engine_volume) / SQL_DIR / f"appdb{DUMP_SUFFIX}"], + "the glob must see the real engine's dump, or the check below is vacuous", + ) + dumps = dumps_in(self._volume_dir(self.impostor_volume) / SQL_DIR) self.assertEqual( dumps, [], diff --git a/tests/e2e/test_e2e_engine_detection_registry_prefix.py b/tests/e2e/test_e2e_engine_detection_registry_prefix.py index 9aa190f..fd6e4ee 100644 --- a/tests/e2e/test_e2e_engine_detection_registry_prefix.py +++ b/tests/e2e/test_e2e_engine_detection_registry_prefix.py @@ -1,5 +1,7 @@ import unittest +from baudolo.generation import DUMP_SUFFIX, SQL_DIR + from .helpers import ( POSTGRES_DATA_DIR, POSTGRES_IMAGE, @@ -7,6 +9,7 @@ from .helpers import ( backup_run, cleanup_docker, create_minimal_compose_dir, + dump_text, ensure_empty_dir, latest_version_dir, require_docker, @@ -94,11 +97,11 @@ class TestE2EEngineDetectionRegistryPrefix(unittest.TestCase): def test_the_registry_host_does_not_pick_the_engine(self) -> None: p = ( backup_path(self.backups_dir, self.repo_name, self.version, self.pg_volume) - / "sql" - / "appdb.backup.sql" + / SQL_DIR + / f"appdb{DUMP_SUFFIX}" ) self.assertTrue(p.is_file(), f"Expected a pg_dump at: {p}") - self.assertIn("Dumped by pg_dump", p.read_text(encoding="utf-8")) + self.assertIn("Dumped by pg_dump", dump_text(p)) if __name__ == "__main__": diff --git a/tests/e2e/test_e2e_mariadb_anonymous_preemption.py b/tests/e2e/test_e2e_mariadb_anonymous_preemption.py index a183db2..e44c2aa 100644 --- a/tests/e2e/test_e2e_mariadb_anonymous_preemption.py +++ b/tests/e2e/test_e2e_mariadb_anonymous_preemption.py @@ -34,6 +34,7 @@ from .helpers import ( MARIADB_DATA_DIR, MARIADB_IMAGE, cleanup_docker, + dump_text, require_docker, run, unique, @@ -159,8 +160,7 @@ class TestE2EMariaDBAnonymousPreemption(unittest.TestCase): self.assertTrue(produced, "backup_database did not produce a dump") dump_path = Path(volume_dir) / SQL_DIR / f"{self.db_name}{DUMP_SUFFIX}" self.assertTrue(dump_path.is_file(), f"expected dump at {dump_path}") - with dump_path.open(encoding="utf-8", errors="replace") as f: - content = f.read() + content = dump_text(dump_path) self.assertIn("INSERT INTO", content) self.assertIn("'ok'", content) diff --git a/tests/e2e/test_e2e_mariadb_full.py b/tests/e2e/test_e2e_mariadb_full.py index abf09af..1877498 100644 --- a/tests/e2e/test_e2e_mariadb_full.py +++ b/tests/e2e/test_e2e_mariadb_full.py @@ -1,5 +1,7 @@ import unittest +from baudolo.generation import DUMP_SUFFIX, SQL_DIR + from .helpers import ( MARIADB_DATA_DIR, MARIADB_IMAGE, @@ -7,6 +9,7 @@ from .helpers import ( backup_run, cleanup_docker, create_minimal_compose_dir, + dump_text, ensure_empty_dir, latest_version_dir, require_docker, @@ -150,10 +153,11 @@ class TestE2EMariaDBFull(unittest.TestCase): def test_dump_file_exists(self) -> None: p = ( backup_path(self.backups_dir, self.repo_name, self.version, self.db_volume) - / "sql" - / f"{self.db_name}.backup.sql" + / SQL_DIR + / f"{self.db_name}{DUMP_SUFFIX}" ) self.assertTrue(p.is_file(), f"Expected dump file at: {p}") + self.assertIn("INSERT INTO", dump_text(p)) def test_data_restored(self) -> None: p = run( diff --git a/tests/e2e/test_e2e_only_files.py b/tests/e2e/test_e2e_only_files.py index 91674cf..81ea78b 100644 --- a/tests/e2e/test_e2e_only_files.py +++ b/tests/e2e/test_e2e_only_files.py @@ -7,12 +7,16 @@ make the flag useless there. import unittest +from baudolo.generation import SQL_DIR + from .helpers import ( POSTGRES_DATA_DIR, POSTGRES_IMAGE, + assert_dumps_in_sees_both_suffixes, backup_path, cleanup_docker, create_minimal_compose_dir, + dumps_in, ensure_empty_dir, latest_version_dir, require_docker, @@ -104,8 +108,8 @@ class TestE2EOnlyFiles(unittest.TestCase): self.assertEqual(marker.read_text(encoding="utf-8").strip(), MARKER) def test_no_dump_is_written(self) -> None: - sql_dir = self._volume_dir() / "sql" - dumps = list(sql_dir.glob("*.sql")) if sql_dir.exists() else [] + assert_dumps_in_sees_both_suffixes(self, f"/tmp/{self.prefix}") + dumps = dumps_in(self._volume_dir() / SQL_DIR) self.assertEqual(dumps, [], f"did not expect any dump, found: {dumps}") def test_the_missing_databases_csv_is_not_reported(self) -> None: diff --git a/tests/e2e/test_e2e_only_sql_fallback_to_files.py b/tests/e2e/test_e2e_only_sql_fallback_to_files.py index 37ecd4b..5ba1eaa 100644 --- a/tests/e2e/test_e2e_only_sql_fallback_to_files.py +++ b/tests/e2e/test_e2e_only_sql_fallback_to_files.py @@ -1,14 +1,24 @@ import json import unittest -from baudolo.generation import FILES_DIR, MANIFEST_FILE, MANIFEST_SCHEMA, SQL_DIR +from baudolo.generation import ( + CLUSTER_SUFFIX, + DUMP_COMPRESSION, + DUMP_SUFFIX, + FILES_DIR, + MANIFEST_FILE, + MANIFEST_SCHEMA, + SQL_DIR, +) from .helpers import ( POSTGRES_DATA_DIR, POSTGRES_IMAGE, + assert_dumps_in_sees_both_suffixes, backup_path, cleanup_docker, create_minimal_compose_dir, + dumps_in, ensure_empty_dir, latest_version_dir, require_docker, @@ -143,7 +153,7 @@ class TestE2EOnlySqlFallbackToFiles(unittest.TestCase): self.assertTrue(p.is_dir(), f"Expected files backup dir at: {p}") def test_sql_dump_not_present(self) -> None: - # There should be no sql dumps because databases.csv had no matching entry. + assert_dumps_in_sees_both_suffixes(self, f"/tmp/{self.prefix}") sql_dir = ( backup_path( self.backups_dir, @@ -151,16 +161,10 @@ class TestE2EOnlySqlFallbackToFiles(unittest.TestCase): self.version, self.pg_volume, ) - / "sql" + / SQL_DIR ) - # Could exist (dir created) in some edge cases, but should contain no *.sql dumps. - if sql_dir.exists(): - dumps = list(sql_dir.glob("*.sql")) - self.assertEqual( - len(dumps), - 0, - f"Did not expect SQL dump files, found: {dumps}", - ) + dumps = dumps_in(sql_dir) + self.assertEqual(dumps, [], f"Did not expect SQL dump files, found: {dumps}") def manifest(self) -> dict: generation = backup_path( @@ -183,6 +187,9 @@ class TestE2EOnlySqlFallbackToFiles(unittest.TestCase): self.assertTrue((volume_dir / layout["files_dir"]).is_dir()) self.assertEqual(layout["files_dir"], FILES_DIR) self.assertEqual(layout["sql_dir"], SQL_DIR) + self.assertEqual(layout["dump_suffix"], DUMP_SUFFIX) + self.assertEqual(layout["cluster_suffix"], CLUSTER_SUFFIX) + self.assertEqual(layout["dump_compression"], DUMP_COMPRESSION) def test_the_manifest_states_a_schema_a_reader_can_check(self) -> None: self.assertEqual(self.manifest()["schema"], MANIFEST_SCHEMA) diff --git a/tests/e2e/test_e2e_postgres_cluster_restore.py b/tests/e2e/test_e2e_postgres_cluster_restore.py index 5aa264e..a912663 100644 --- a/tests/e2e/test_e2e_postgres_cluster_restore.py +++ b/tests/e2e/test_e2e_postgres_cluster_restore.py @@ -1,5 +1,7 @@ import unittest +from baudolo.generation import CLUSTER_SUFFIX, SQL_DIR + from .helpers import ( POSTGRES_DATA_DIR, POSTGRES_IMAGE, @@ -85,8 +87,8 @@ class TestE2EPostgresClusterRestore(unittest.TestCase): cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name) cls.dump = ( backup_path(cls.backups_dir, cls.repo_name, cls.version, cls.pg_volume) - / "sql" - / f"{cls.pg_container}.cluster.backup.sql" + / SQL_DIR + / f"{cls.pg_container}{CLUSTER_SUFFIX}" ) run(cls._restore_argv(), stdin_text="pgpw") diff --git a/tests/e2e/test_e2e_postgres_full.py b/tests/e2e/test_e2e_postgres_full.py index ede27a2..92d6d6b 100644 --- a/tests/e2e/test_e2e_postgres_full.py +++ b/tests/e2e/test_e2e_postgres_full.py @@ -1,5 +1,7 @@ import unittest +from baudolo.generation import DUMP_SUFFIX, SQL_DIR + from .helpers import ( POSTGRES_DATA_DIR, POSTGRES_IMAGE, @@ -7,6 +9,7 @@ from .helpers import ( backup_run, cleanup_docker, create_minimal_compose_dir, + dump_text, ensure_empty_dir, latest_version_dir, require_docker, @@ -124,10 +127,11 @@ class TestE2EPostgresFull(unittest.TestCase): def test_dump_file_exists(self) -> None: p = ( backup_path(self.backups_dir, self.repo_name, self.version, self.pg_volume) - / "sql" - / "appdb.backup.sql" + / SQL_DIR + / f"appdb{DUMP_SUFFIX}" ) self.assertTrue(p.is_file(), f"Expected dump file at: {p}") + self.assertIn("Dumped by pg_dump", dump_text(p)) def test_the_deprecated_password_argument_still_restores_but_warns(self) -> None: self.assertIn("visible in the process list", self.restored.stderr) diff --git a/tests/e2e/test_e2e_postgres_password_required.py b/tests/e2e/test_e2e_postgres_password_required.py index adab8bf..2330424 100644 --- a/tests/e2e/test_e2e_postgres_password_required.py +++ b/tests/e2e/test_e2e_postgres_password_required.py @@ -19,6 +19,7 @@ from .helpers import ( backup_run, cleanup_docker, create_minimal_compose_dir, + dump_text, ensure_empty_dir, latest_version_dir, require_docker, @@ -137,19 +138,19 @@ class TestE2EPostgresPasswordRequired(unittest.TestCase): def test_the_configured_database_was_dumped(self) -> None: dump = self.volume_dir() / SQL_DIR / f"appdb{DUMP_SUFFIX}" self.assertTrue(dump.is_file(), f"expected a dump at {dump}") - self.assertIn("Dumped by pg_dump", dump.read_text(encoding="utf-8")) + self.assertIn("Dumped by pg_dump", dump_text(dump)) def test_the_dump_carries_the_payload(self) -> None: """pg_dump emits table data as COPY ... FROM stdin, so the row reads as tab-separated values rather than as an INSERT literal.""" - dump = self.volume_dir() / SQL_DIR / f"appdb{DUMP_SUFFIX}" - self.assertIn("COPY public.t (id, v) FROM stdin;", dump.read_text("utf-8")) - self.assertIn("1\tok", dump.read_text(encoding="utf-8")) + text = dump_text(self.volume_dir() / SQL_DIR / f"appdb{DUMP_SUFFIX}") + self.assertIn("COPY public.t (id, v) FROM stdin;", text) + self.assertIn("1\tok", text) def test_the_cluster_row_was_dumped_too(self) -> None: cluster = self.volume_dir() / SQL_DIR / f"{self.pg_container}{CLUSTER_SUFFIX}" self.assertTrue(cluster.is_file(), f"expected a cluster dump at {cluster}") - self.assertIn("CREATE DATABASE", cluster.read_text(encoding="utf-8")) + self.assertIn("CREATE DATABASE", dump_text(cluster)) def test_only_sql_left_no_file_copy_behind(self) -> None: self.assertFalse((self.volume_dir() / FILES_DIR).exists()) diff --git a/tests/e2e/test_e2e_seed_star_and_db_entries_backup_postgres.py b/tests/e2e/test_e2e_seed_star_and_db_entries_backup_postgres.py index e4f1be3..04138da 100644 --- a/tests/e2e/test_e2e_seed_star_and_db_entries_backup_postgres.py +++ b/tests/e2e/test_e2e_seed_star_and_db_entries_backup_postgres.py @@ -1,11 +1,14 @@ import unittest +from baudolo.generation import CLUSTER_SUFFIX, DUMP_SUFFIX + from .helpers import ( POSTGRES_DATA_DIR, POSTGRES_IMAGE, backup_path, cleanup_docker, create_minimal_compose_dir, + dump_text, ensure_empty_dir, latest_version_dir, require_docker, @@ -197,15 +200,15 @@ class TestE2ESeedStarAndDbEntriesBackupPostgres(unittest.TestCase): ) # Cluster dump file produced by '*' entry - cluster = sql_dir / f"{self.pg_container}.cluster.backup.sql" + cluster = sql_dir / f"{self.pg_container}{CLUSTER_SUFFIX}" self.assertTrue(cluster.is_file(), f"Expected cluster dump file at: {cluster}") # Concrete DB dump produced by normal entry - db1 = sql_dir / f"{self.pg_db1}.backup.sql" + db1 = sql_dir / f"{self.pg_db1}{DUMP_SUFFIX}" self.assertTrue(db1.is_file(), f"Expected db dump file at: {db1}") # Basic sanity: cluster dump usually contains CREATE DATABASE statements - txt = cluster.read_text(encoding="utf-8", errors="ignore") + txt = dump_text(cluster) self.assertIn( "CREATE DATABASE", txt, diff --git a/tests/e2e/test_e2e_version_gate.py b/tests/e2e/test_e2e_version_gate.py index a4692b7..0e239f3 100644 --- a/tests/e2e/test_e2e_version_gate.py +++ b/tests/e2e/test_e2e_version_gate.py @@ -10,6 +10,8 @@ import re import unittest from pathlib import Path +from baudolo.generation import DUMP_SUFFIX, SQL_DIR + from .helpers import ( MARIADB_DATA_DIR, MARIADB_IMAGE, @@ -19,6 +21,7 @@ from .helpers import ( backup_run, cleanup_docker, create_minimal_compose_dir, + dump_text, ensure_empty_dir, latest_version_dir, require_docker, @@ -28,6 +31,7 @@ from .helpers import ( wait_for_mariadb_sql, wait_for_postgres, write_databases_csv, + write_dump_text, ) PAYLOAD = "gate-payload" @@ -36,14 +40,14 @@ FUTURE = "99.0" def rewrite_version(dump: Path, pattern: str, version: str) -> str: """Make the dump claim ``version``; return what it claimed before.""" - text = dump.read_text(encoding="utf-8", errors="replace") + text = dump_text(dump) found = re.search(pattern, text) if not found: raise AssertionError(f"{dump} carries no version header matching {pattern}") claimed = found.group(1) - dump.write_text( + write_dump_text( + dump, text.replace(found.group(0), found.group(0).replace(claimed, version), 1), - encoding="utf-8", ) return claimed @@ -103,8 +107,8 @@ class GateCase: cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name) cls.dump = ( backup_path(cls.backups_dir, cls.repo_name, cls.version, cls.volume) - / "sql" - / f"{cls.db_name}.backup.sql" + / SQL_DIR + / f"{cls.db_name}{DUMP_SUFFIX}" ) cls.truthful_version = rewrite_version(cls.dump, cls.pattern, FUTURE) diff --git a/tests/unit/backup/test_db_mariadb_dump.py b/tests/unit/backup/test_db_mariadb_dump.py index 597b727..afc05c5 100644 --- a/tests/unit/backup/test_db_mariadb_dump.py +++ b/tests/unit/backup/test_db_mariadb_dump.py @@ -20,7 +20,7 @@ def _capture_dumps(*, db_type, rows, container, dump_tool="mariadb-dump"): with ( tempfile.TemporaryDirectory() as td, - patch.object(db_mod, "execute_to_file", side_effect=_capture), + patch.object(db_mod, "execute_to_gzip", side_effect=_capture), ): db_mod.backup_database( container=container, @@ -94,6 +94,20 @@ class TestMariaDBDumpUsesTCP(unittest.TestCase): f"password reached the argv: {argv}", ) + def test_every_dump_is_written_gzip_named(self): + for db_type, container, database, suffix in ( + ("postgres", "pg", "appdb", ".backup.sql.gz"), + ("mariadb", "mariadb", "appdb", ".backup.sql.gz"), + ("postgres", "pg", "*", ".cluster.backup.sql.gz"), + ): + with self.subTest(engine=db_type, database=database): + _argv, _env, out = _capture_dumps( + db_type=db_type, + rows=[(container, database, "appuser", "s3cret")], + container=container, + )[0] + self.assertTrue(out.endswith(suffix), out) + def dump_tool_of(argv): return next(arg for arg in argv if arg in ("mariadb-dump", "mysqldump")) diff --git a/tests/unit/backup/test_docker_exec_env.py b/tests/unit/backup/test_docker_exec_env.py index 289ab0c..d2f0971 100644 --- a/tests/unit/backup/test_docker_exec_env.py +++ b/tests/unit/backup/test_docker_exec_env.py @@ -44,12 +44,12 @@ class TestPostgresDumpCarriesThePassword(unittest.TestCase): import baudolo.backup.db as db - original = db.execute_to_file - db.execute_to_file = fake + original = db.execute_to_gzip + db.execute_to_gzip = fake try: fallback_pg_dumpall("pg", "user", "secret", "/tmp/out.sql") finally: - db.execute_to_file = original + db.execute_to_gzip = original self.assertIn("-e", seen["command"]) self.assertEqual(seen["command"][seen["command"].index("-e") + 1], "PGPASSWORD") diff --git a/tests/unit/backup/test_shell_gzip.py b/tests/unit/backup/test_shell_gzip.py new file mode 100644 index 0000000..c75da16 --- /dev/null +++ b/tests/unit/backup/test_shell_gzip.py @@ -0,0 +1,126 @@ +import contextlib +import gzip +import io +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from baudolo.backup import shell +from baudolo.backup.shell import BackupError, execute_to_gzip + +CAT = ( + "import shutil, sys; shutil.copyfileobj(open(sys.argv[1], 'rb'), sys.stdout.buffer)" +) + + +def sql_lines(count: int, changed: int = -1) -> bytes: + return b"".join( + ( + b"INSERT INTO t VALUES (%d, 'changed');\n" % i + if i == changed + else b"INSERT INTO t VALUES (%d, 'name-%d');\n" % (i, i * 7919) + ) + for i in range(count) + ) + + +def common_suffix(a: bytes, b: bytes) -> int: + n = 0 + while n < min(len(a), len(b)) and a[-1 - n] == b[-1 - n]: + n += 1 + return n + + +class GzipDumpCase(unittest.TestCase): + def setUp(self): + self.dir = Path(tempfile.mkdtemp()) + + def dump(self, payload: bytes, name: str = "db.backup.sql.gz") -> bytes: + src = self.dir / f"{name}.src" + src.write_bytes(payload) + out = self.dir / name + with contextlib.redirect_stdout(io.StringIO()): + execute_to_gzip([sys.executable, "-c", CAT, str(src)], str(out)) + return out.read_bytes() + + +class TestRoundTrip(GzipDumpCase): + def test_the_dump_round_trips_through_gzip(self): + payload = sql_lines(5000) + b"-- " + b"x" * (3 * shell.READ_BYTES) + b"\n" + packed = self.dump(payload) + self.assertEqual(packed[:2], b"\x1f\x8b") + self.assertEqual(gzip.decompress(packed), payload) + + def test_an_empty_dump_is_a_valid_empty_gzip(self): + self.assertEqual(gzip.decompress(self.dump(b"")), b"") + + def test_no_temporary_file_is_left_behind(self): + self.dump(sql_lines(10)) + self.assertEqual(sorted(p.suffix for p in self.dir.iterdir()), [".gz", ".src"]) + + +class TestDeterminism(GzipDumpCase): + def test_identical_dumps_give_identical_bytes(self): + payload = sql_lines(20000) + self.assertEqual(self.dump(payload, "a.gz"), self.dump(payload, "b.gz")) + + def test_the_header_carries_neither_a_name_nor_a_time(self): + packed = self.dump(sql_lines(10)) + self.assertEqual(packed[3] & 0x08, 0, "FNAME flag set") + self.assertEqual(packed[4:8], b"\x00\x00\x00\x00", "MTIME set") + + +class TestResync(GzipDumpCase): + def test_a_change_near_the_start_leaves_the_tail_identical(self): + before = sql_lines(100000) + after = sql_lines(100000, changed=5) + ours = common_suffix( + self.dump(before, "a.gz")[:-8], self.dump(after, "b.gz")[:-8] + ) + plain = common_suffix(gzip.compress(before)[:-8], gzip.compress(after)[:-8]) + self.assertLess(plain, 1024, "control: plain gzip must not resync") + self.assertGreater(ours, len(gzip.compress(before)) // 2) + + +class TestFailure(GzipDumpCase): + def test_a_failing_command_leaves_no_file(self): + out = self.dir / "db.backup.sql.gz" + script = "import sys; sys.stdout.write('partial'); sys.exit(3)" + with ( + contextlib.redirect_stdout(io.StringIO()), + self.assertRaises(BackupError) as caught, + ): + execute_to_gzip([sys.executable, "-c", script], str(out)) + self.assertIn("Exit code: 3", str(caught.exception)) + self.assertEqual(list(self.dir.iterdir()), []) + + def test_a_write_error_kills_the_dump_and_leaves_no_file(self): + src = self.dir / "src" + src.write_bytes(sql_lines(200000)) + out = self.dir / "db.backup.sql.gz" + with ( + patch.object( + gzip.GzipFile, "write", side_effect=OSError(28, "No space left") + ), + contextlib.redirect_stdout(io.StringIO()), + self.assertRaises(BackupError), + ): + execute_to_gzip([sys.executable, "-c", CAT, str(src)], str(out)) + self.assertEqual(sorted(p.name for p in self.dir.iterdir()), ["src"]) + + def test_env_reaches_the_child_without_the_argv(self): + out = self.dir / "env.gz" + script = "import os, sys; sys.stdout.write(os.environ['MYSQL_PWD'])" + printed = io.StringIO() + with contextlib.redirect_stdout(printed): + execute_to_gzip( + [sys.executable, "-c", script], str(out), env={"MYSQL_PWD": "s3cret"} + ) + self.assertEqual(gzip.decompress(out.read_bytes()), b"s3cret") + self.assertNotIn("s3cret", printed.getvalue()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/restore/test_cli_password_stdin.py b/tests/unit/restore/test_cli_password_stdin.py index 080733a..78e8f35 100644 --- a/tests/unit/restore/test_cli_password_stdin.py +++ b/tests/unit/restore/test_cli_password_stdin.py @@ -24,6 +24,7 @@ class TestRestorePasswordChannel(unittest.TestCase): stderr = io.StringIO() with ( patch.object(cli, "restore_postgres_sql") as restore, + patch.object(cli, "locate", side_effect=lambda path: path), patch("sys.stdin", io.StringIO(stdin)), contextlib.redirect_stderr(stderr), ): diff --git a/tests/unit/restore/test_cli_version_flag.py b/tests/unit/restore/test_cli_version_flag.py index a849f18..dc6a97b 100644 --- a/tests/unit/restore/test_cli_version_flag.py +++ b/tests/unit/restore/test_cli_version_flag.py @@ -27,7 +27,10 @@ class TestVersionFlagReachesEveryEngine(unittest.TestCase): *required, *extra, ] - with patch.object(cli, target) as restore: + with ( + patch.object(cli, target) as restore, + patch.object(cli, "locate", side_effect=lambda path: path), + ): self.assertEqual(cli.main(argv), 0) return restore.call_args.kwargs diff --git a/tests/unit/restore/test_cluster_replay.py b/tests/unit/restore/test_cluster_replay.py index 0b49b7a..8e89de4 100644 --- a/tests/unit/restore/test_cluster_replay.py +++ b/tests/unit/restore/test_cluster_replay.py @@ -117,7 +117,7 @@ class TestClusterReplay(unittest.TestCase): paths = BackupPaths("vol", "hash", "v1", repo_name="repo", backups_dir="/B") self.assertEqual( paths.cluster_file("bigbluebutton"), - "/B/hash/repo/v1/vol/sql/bigbluebutton.cluster.backup.sql", + "/B/hash/repo/v1/vol/sql/bigbluebutton.cluster.backup.sql.gz", ) diff --git a/tests/unit/restore/test_compressed_restore.py b/tests/unit/restore/test_compressed_restore.py new file mode 100644 index 0000000..58b9058 --- /dev/null +++ b/tests/unit/restore/test_compressed_restore.py @@ -0,0 +1,242 @@ +import gzip +import sys +import tempfile +import unittest +import zlib +from pathlib import Path +from unittest.mock import MagicMock, patch + +from baudolo.restore import run as run_mod +from baudolo.restore.db import cluster as cluster_mod +from baudolo.restore.db import mariadb as mdb_mod +from baudolo.restore.db import postgres as pg_mod +from baudolo.restore.db import version as ver +from baudolo.restore.dumpfile import is_gzip, locate, open_dump, verify +from baudolo.restore.paths import BackupPaths + +from .test_version_gate import MARIADB_HEADER, POSTGRES_HEADER, cluster_header + +SQL = ( + POSTGRES_HEADER + "CREATE TABLE t (id int);\nINSERT INTO t VALUES (1);\n" +).encode() + + +def write(directory: Path, name: str, data: bytes) -> str: + path = directory / name + path.write_bytes(data) + return str(path) + + +class DumpCase(unittest.TestCase): + def setUp(self): + self.dir = Path(tempfile.mkdtemp()) + + +class TestOpenDump(DumpCase): + def test_gzip_and_plain_read_the_same_sql(self): + packed = write(self.dir, "a.backup.sql.gz", gzip.compress(SQL)) + plain = write(self.dir, "b.backup.sql", SQL) + for path in (packed, plain): + with self.subTest(path=path), open_dump(path) as handle: + self.assertEqual(handle.read(), SQL) + + def test_the_magic_decides_not_the_name(self): + misnamed = write(self.dir, "c.backup.sql", gzip.compress(SQL)) + self.assertTrue(is_gzip(misnamed)) + with open_dump(misnamed) as handle: + self.assertEqual(handle.read(), SQL) + + +class TestLocate(DumpCase): + def test_the_compressed_name_wins_when_alone(self): + path = write(self.dir, "db.backup.sql.gz", gzip.compress(SQL)) + self.assertEqual(locate(path), path) + + def test_an_old_uncompressed_dump_is_found(self): + legacy = write(self.dir, "db.backup.sql", SQL) + self.assertEqual(locate(str(self.dir / "db.backup.sql.gz")), legacy) + + def test_both_present_is_refused(self): + write(self.dir, "db.backup.sql", SQL) + path = write(self.dir, "db.backup.sql.gz", gzip.compress(SQL)) + with self.assertRaises(FileExistsError): + locate(path) + + def test_neither_present_names_both(self): + with self.assertRaises(FileNotFoundError) as caught: + locate(str(self.dir / "db.backup.sql.gz")) + self.assertIn("db.backup.sql.gz", str(caught.exception)) + self.assertIn("db.backup.sql ", str(caught.exception) + " ") + + def test_the_path_helper_names_what_this_version_writes(self): + paths = BackupPaths("vol", "h", "v", repo_name="r", backups_dir="/B") + self.assertEqual(paths.sql_file("app"), "/B/h/r/v/vol/sql/app.backup.sql.gz") + + +class TestVerify(DumpCase): + def test_a_sound_gzip_and_a_plain_file_pass(self): + verify(write(self.dir, "a.gz", gzip.compress(SQL))) + verify(write(self.dir, "b.sql", SQL[:10])) + + def test_a_truncated_gzip_is_refused(self): + packed = gzip.compress(SQL * 50) + with self.assertRaises(EOFError): + verify(write(self.dir, "t.gz", packed[: len(packed) // 2])) + + def test_a_corrupt_gzip_is_refused(self): + packed = bytearray(gzip.compress(SQL * 50)) + packed[-6] ^= 0xFF + with self.assertRaises((gzip.BadGzipFile, zlib.error, EOFError)): + verify(write(self.dir, "c.gz", bytes(packed))) + + +class TestReadersAcceptGzip(DumpCase): + def test_the_version_header_is_read_through_gzip(self): + for header, engine, expected in ( + (POSTGRES_HEADER, "postgres", "17.11"), + (MARIADB_HEADER, "mariadb", "11.8.8-MariaDB-ubu2404"), + ): + with self.subTest(engine=engine): + path = write(self.dir, f"{engine}.gz", gzip.compress(header.encode())) + self.assertEqual(ver.dump_version(path, engine), expected) + + def test_the_cluster_inventory_is_read_through_gzip(self): + data = (cluster_header(roles=2) + "CREATE DATABASE app2;\n").encode() + path = write(self.dir, "c.cluster.backup.sql.gz", gzip.compress(data)) + databases, roles = cluster_mod.dump_inventory(path) + self.assertIn("app2", databases) + self.assertEqual(roles, ["app0", "app1"]) + + +class TestReplayFeedsSql(DumpCase): + def test_postgres_replays_decompressed_sql(self): + fed = [] + + def capture(container, argv, **kwargs): + stdin = kwargs.get("stdin") + fed.append(stdin if isinstance(stdin, bytes) else stdin.read()) + return MagicMock() + + for name, data in ( + ("new.backup.sql.gz", gzip.compress(SQL)), + ("old.backup.sql", SQL), + ): + fed.clear() + path = write(self.dir, name, data) + with ( + self.subTest(name=name), + patch.object(pg_mod, "docker_exec", side_effect=capture), + ): + pg_mod.restore_postgres_sql( + container="db", + db_name="app", + user="app", + password="pw", + sql_path=path, + empty=False, + check_version=False, + ) + self.assertEqual(fed, [SQL]) + + def test_mariadb_streams_decompressed_sql(self): + fed = [] + + def capture(container, argv, *, source, docker_env): + fed.append(source.read()) + + for name, data in ( + ("new.backup.sql.gz", gzip.compress(SQL)), + ("old.backup.sql", SQL), + ): + fed.clear() + path = write(self.dir, name, data) + with ( + self.subTest(name=name), + patch.object(mdb_mod, "_pick_client", return_value="mariadb"), + patch.object(mdb_mod, "docker_exec_stream", side_effect=capture), + ): + mdb_mod.restore_mariadb_sql( + container="db", + db_name="app", + user="app", + password="pw", + sql_path=path, + empty=False, + check_version=False, + ) + self.assertEqual(fed, [SQL]) + + def test_the_stream_reaches_the_child_decompressed(self): + out = self.dir / "received.sql" + sink = "import shutil, sys; shutil.copyfileobj(sys.stdin.buffer, open(sys.argv[1], 'wb'))" + path = write(self.dir, "s.gz", gzip.compress(SQL * 2000)) + with ( + patch.object( + run_mod, + "docker_exec_cmd", + return_value=[sys.executable, "-c", sink, str(out)], + ), + open_dump(path) as source, + ): + run_mod.docker_exec_stream("db", ["ignored"], source=source) + self.assertEqual(out.read_bytes(), SQL * 2000) + + def test_a_child_that_dies_early_is_reported_by_its_exit_code(self): + path = write(self.dir, "s.gz", gzip.compress(SQL * 20000)) + with ( + patch.object( + run_mod, + "docker_exec_cmd", + return_value=[sys.executable, "-c", "import sys; sys.exit(3)"], + ), + open_dump(path) as source, + patch("sys.stderr"), + self.assertRaises(run_mod.subprocess.CalledProcessError) as caught, + ): + run_mod.docker_exec_stream("db", ["ignored"], source=source) + self.assertEqual(caught.exception.returncode, 3) + + +class TestCorruptionStopsBeforeDestroying(DumpCase): + def truncated(self) -> str: + packed = gzip.compress(SQL * 500) + return write(self.dir, "x.backup.sql.gz", packed[: len(packed) // 2]) + + def test_postgres_runs_no_preclean_on_a_truncated_dump(self): + with ( + patch.object(pg_mod, "docker_exec") as exec_, + self.assertRaises(EOFError), + ): + pg_mod.restore_postgres_sql( + container="db", + db_name="app", + user="app", + password="pw", + sql_path=self.truncated(), + empty=True, + check_version=False, + ) + exec_.assert_not_called() + + def test_mariadb_drops_nothing_on_a_truncated_dump(self): + with ( + patch.object(mdb_mod, "_pick_client", return_value="mariadb"), + patch.object(mdb_mod, "docker_exec") as exec_, + patch.object(mdb_mod, "docker_exec_stream") as stream, + self.assertRaises(EOFError), + ): + mdb_mod.restore_mariadb_sql( + container="db", + db_name="app", + user="app", + password="pw", + sql_path=self.truncated(), + empty=True, + check_version=False, + ) + exec_.assert_not_called() + stream.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/restore/test_mariadb_empty_drop.py b/tests/unit/restore/test_mariadb_empty_drop.py index a82ade6..3889c04 100644 --- a/tests/unit/restore/test_mariadb_empty_drop.py +++ b/tests/unit/restore/test_mariadb_empty_drop.py @@ -20,6 +20,7 @@ class TestMariadbEmptyDrop(unittest.TestCase): sql.flush() with ( patch.object(mariadb_mod, "docker_exec", side_effect=_capture), + patch.object(mariadb_mod, "docker_exec_stream"), patch.object(mariadb_mod, "_pick_client", return_value="mariadb"), ): mariadb_mod.restore_mariadb_sql( diff --git a/tests/unit/restore/test_password_off_argv.py b/tests/unit/restore/test_password_off_argv.py index c5b3cea..363c330 100644 --- a/tests/unit/restore/test_password_off_argv.py +++ b/tests/unit/restore/test_password_off_argv.py @@ -1,3 +1,4 @@ +import gzip import tempfile import unittest from pathlib import Path @@ -39,6 +40,20 @@ class TestDockerExecForwardsByName(unittest.TestCase): self.assertIsNone(run.call_args.kwargs["env"]) self.assertEqual(run.call_args.args[0], ["docker", "exec", "-i", "db", "true"]) + def test_the_stream_forwards_by_name_too(self): + with patch.object(run_mod.subprocess, "Popen") as popen: + popen.return_value.wait.return_value = 0 + run_mod.docker_exec_stream( + "db", + ["mariadb"], + source=MagicMock(read=MagicMock(return_value=b"")), + docker_env={"MYSQL_PWD": SECRET}, + ) + cmd = popen.call_args.args[0] + self.assertFalse([arg for arg in cmd if SECRET in arg]) + self.assertIn("MYSQL_PWD", cmd) + self.assertEqual(popen.call_args.kwargs["env"]["MYSQL_PWD"], SECRET) + class TestNoRestoreArgvCarriesThePassword(unittest.TestCase): def setUp(self): @@ -54,12 +69,20 @@ class TestNoRestoreArgvCarriesThePassword(unittest.TestCase): stdout = b"11.8.8-MariaDB\n" return MagicMock(stdout=stdout) + def fake_popen(cmd, *, stdin=None, env=None): + self.argvs.append(cmd) + self.envs.append(env or {}) + process = MagicMock() + process.wait.return_value = 0 + return process + patch.object(run_mod, "run", side_effect=fake_run).start() + patch.object(run_mod.subprocess, "Popen", side_effect=fake_popen).start() self.addCleanup(patch.stopall) def dump(self, name: str, text: str) -> str: path = self.dir / name - path.write_text(text, encoding="utf-8") + path.write_bytes(gzip.compress(text.encode())) return str(path) def assert_clean(self, variable: str): @@ -74,7 +97,7 @@ class TestNoRestoreArgvCarriesThePassword(unittest.TestCase): db_name="app", user="app", password=SECRET, - sql_path=self.dump("app.backup.sql", POSTGRES_HEADER), + sql_path=self.dump("app.backup.sql.gz", POSTGRES_HEADER), empty=True, ) self.assert_clean("PGPASSWORD") @@ -85,7 +108,7 @@ class TestNoRestoreArgvCarriesThePassword(unittest.TestCase): container="db", user="postgres", password=SECRET, - sql_path=self.dump("c.cluster.backup.sql", cluster_header(roles=1)), + sql_path=self.dump("c.cluster.backup.sql.gz", cluster_header(roles=1)), empty=True, ) self.assert_clean("PGPASSWORD") @@ -97,7 +120,7 @@ class TestNoRestoreArgvCarriesThePassword(unittest.TestCase): db_name="app", user="app", password=SECRET, - sql_path=self.dump("app.backup.sql", MARIADB_HEADER), + sql_path=self.dump("app.backup.sql.gz", MARIADB_HEADER), empty=True, ) self.assert_clean("MYSQL_PWD") diff --git a/tests/unit/test_generation.py b/tests/unit/test_generation.py index 96366d1..7d210f4 100644 --- a/tests/unit/test_generation.py +++ b/tests/unit/test_generation.py @@ -13,6 +13,8 @@ from baudolo.generation import ( CLUSTER_SUFFIX, DUMP_SUFFIX, FILES_DIR, + LEGACY_CLUSTER_SUFFIX, + LEGACY_DUMP_SUFFIX, MANIFEST_FILE, MANIFEST_SCHEMA, SQL_DIR, @@ -30,9 +32,19 @@ class TestManifestDocument(unittest.TestCase): "sql_dir": SQL_DIR, "dump_suffix": DUMP_SUFFIX, "cluster_suffix": CLUSTER_SUFFIX, + "dump_compression": "gzip", }, ) + def test_dumps_are_named_as_gzip_and_the_legacy_names_stay_known(self) -> None: + self.assertEqual(DUMP_SUFFIX, ".backup.sql.gz") + self.assertEqual(CLUSTER_SUFFIX, ".cluster.backup.sql.gz") + self.assertEqual(LEGACY_DUMP_SUFFIX, ".backup.sql") + self.assertEqual(LEGACY_CLUSTER_SUFFIX, ".cluster.backup.sql") + + def test_the_schema_stays_one_so_existing_readers_keep_reading(self) -> None: + self.assertEqual(MANIFEST_SCHEMA, 1) + def test_it_carries_a_schema_so_a_reader_can_refuse_a_newer_one(self) -> None: self.assertEqual(manifest_document({})["schema"], MANIFEST_SCHEMA)