feat(executors): add pg_dump executor with per-database restore verification
postgres-shared had no consistent backup. The only thing that copied it was config_backup_executor, which tars /data/docker-data -- including the live data directory -- while the server runs, without pg_backup_start: a file copy of a running cluster, with no per-database restore. config_backup_executor is left untouched; retiring it is a separate decision. Dumps each database with `docker exec postgres-shared pg_dump -Fc <db>`, server-side inside the PG16 container rather than through this image's own client. Confirmed live: postgres-shared runs pg_dump 16.15, this image carries pg_restore/pg_dump 17.10 -- a 17 custom archive would not read back on a 16 pg_restore, and 17's plain SQL emits SET transaction_timeout, which 16 rejects. Dumping inside the 16 container is version-matched by construction; the mismatch runs the other way for verification, so pg_restore --list runs directly against the finished archive with no docker exec needed. Every archive is written to a .partial file, verified with pg_restore --list (zero TOC entries counts as a failed dump, not a kept one), then renamed and chmod 0600 -- the dump contains table data and the globals-only dump contains role password hashes. One database failing does not stop the others; their files are kept, every failure is named in the result string, and the run is still reported CRITICAL to check_history (D-33). Not yet registered as a scheduled task, and nothing about config_backup_executor or the deploy surface changes here (T-7). tests/test_pg_dump_executor.py: 32 new tests, all subprocess calls mocked -- hermetic per D-26, no live services touched by `make test`. Mutation-checked: dropping the chmod(0o600) fails the mode test; relaxing the TOC check to accept zero entries fails test_verify_archive_raises_on_zero_entries. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -6,6 +6,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
- `pg_dump_executor`: dumps every postgres-shared database individually (`docker exec
|
||||
postgres-shared pg_dump -Fc`, version-matched inside that container rather than through this
|
||||
image's PG17 client), plus a globals-only dump of roles. Each archive is verified with
|
||||
`pg_restore --list` before being promoted — a zero-TOC archive is treated as a failed dump,
|
||||
not a kept one. One database failing does not stop the others; the run is still reported
|
||||
CRITICAL and every failure is named. Not yet registered as a scheduled task (T-7).
|
||||
|
||||
## [1.9.0] - 2026-08-11
|
||||
|
||||
### Changed
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
"""
|
||||
Postgres Dump Executor
|
||||
|
||||
Backs up every database on postgres-shared with `pg_dump`, one file per
|
||||
database plus a globals-only dump of roles, so any single database can be
|
||||
restored on its own rather than only as part of a full-cluster file copy.
|
||||
|
||||
Why this exists: the only thing that currently copies postgres-shared is
|
||||
config_backup_executor, which tars `/data/docker-data` — including the live
|
||||
data directory — while the server runs, without `pg_backup_start`. That is a
|
||||
file copy of a running cluster, with no per-database restore and no
|
||||
verification that the copy is even usable. This executor is the proven
|
||||
replacement; config_backup_executor is left untouched until it is retired by
|
||||
a separate decision.
|
||||
|
||||
Why the dump runs server-side, through `docker exec`, instead of this
|
||||
image's own pg_dump: postgres-shared is PostgreSQL 16.15; this scheduler
|
||||
image carries the client tools for PostgreSQL 17.10. A 17 custom-format
|
||||
archive cannot be read by a 16 `pg_restore`, and 17's plain SQL dumps emit
|
||||
`SET transaction_timeout`, which 16 rejects outright. `docker exec
|
||||
postgres-shared pg_dump ...` runs version 16's own pg_dump inside that
|
||||
container, so the archive is version-matched by construction. The mismatch
|
||||
runs the other way for verification: 17's `pg_restore --list` reads a 16
|
||||
archive without trouble, so it is run directly — no `docker exec` needed,
|
||||
because the scheduler container already has it.
|
||||
|
||||
Config schema:
|
||||
{
|
||||
"container": "postgres-shared", # required; the server's container name
|
||||
"databases": "all", # "all", or an explicit list of names
|
||||
"exclude": ["template0", "template1",
|
||||
"test_scheduler", "core_api_test"],
|
||||
"globals": true, # also dump roles via pg_dumpall --globals-only
|
||||
"output_dir": "/backups/postgres",
|
||||
"retention_days": 14,
|
||||
"timeout_per_db": 900 # seconds; applies to each dump individually
|
||||
}
|
||||
|
||||
"all" is discovered with `select datname from pg_database where not
|
||||
datistemplate`, then `exclude` is subtracted — so the exclude list is a
|
||||
safety net even when the template databases are already filtered out
|
||||
upstream.
|
||||
|
||||
Every archive is written to `<name>-<stamp>.dump.partial`, verified, then
|
||||
renamed and chmod 0600 — the same partial-then-rename-then-chmod shape as
|
||||
portainer_backup_executor, because a PostgreSQL dump contains table data and
|
||||
the globals dump contains role password hashes.
|
||||
|
||||
Failure semantics: one database failing does not stop the others, and their
|
||||
files are kept. The run as a whole is still reported CRITICAL, and every
|
||||
failed target is named in the result string, by construction — that string
|
||||
is built by joining each target's own outcome, not a separate summary of it.
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from src.config import Settings
|
||||
from src.executors import health_report
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_OUTPUT_DIR = "/backups/postgres"
|
||||
DEFAULT_RETENTION_DAYS = 14
|
||||
DEFAULT_TIMEOUT_PER_DB = 900
|
||||
|
||||
DISCOVER_SQL = "select datname from pg_database where not datistemplate"
|
||||
|
||||
# Matches every filename this executor writes and nothing else: "<name>" is
|
||||
# either a database name (letters, digits, underscore, hyphen, dot — real
|
||||
# names here include "sonarr-log") or the literal "globals", followed by the
|
||||
# UTC run stamp and the right extension. `_prune` uses this to delete only
|
||||
# what it wrote, never a file placed here by something else.
|
||||
FILENAME_RE = re.compile(r"^(?:globals|[A-Za-z0-9_.\-]+)-\d{8}T\d{6}Z\.(?:dump|sql)$")
|
||||
|
||||
|
||||
def _toc_entry_count(list_output: str) -> int:
|
||||
"""Count TOC entry lines from `pg_restore --list` output.
|
||||
|
||||
Every header and comment line in that output starts with ';'; the actual
|
||||
entries pg_restore would restore do not. Counting on that distinction is
|
||||
more direct than parsing the "; TOC Entries: N" summary line, and does not
|
||||
depend on that line's exact wording surviving a pg_restore version bump.
|
||||
"""
|
||||
return sum(
|
||||
1 for line in list_output.splitlines()
|
||||
if line.strip() and not line.lstrip().startswith(";")
|
||||
)
|
||||
|
||||
|
||||
async def _verify_archive(path: Path) -> int:
|
||||
"""Run the image's own pg_restore --list against a finished dump.
|
||||
|
||||
A 200-looking response that is not a usable archive is worse than no
|
||||
backup at all, because it looks fine until the day it is needed — the
|
||||
same reasoning portainer_backup_executor applies to its own archive. The
|
||||
only signal pg_restore --list gives back is the TOC; zero entries is
|
||||
treated the same as a file that would not open.
|
||||
"""
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
"pg_restore", "--list", str(path),
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await proc.communicate()
|
||||
if proc.returncode != 0:
|
||||
raise Exception(f"pg_restore --list failed: {stderr.decode(errors='replace').strip()[:300]}")
|
||||
entries = _toc_entry_count(stdout.decode(errors="replace"))
|
||||
if entries <= 0:
|
||||
raise Exception("archive has zero TOC entries")
|
||||
return entries
|
||||
|
||||
|
||||
async def _discover_databases(container: str) -> List[str]:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
"docker", "exec", container, "psql", "-U", "postgres", "-tAc", DISCOVER_SQL,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await proc.communicate()
|
||||
if proc.returncode != 0:
|
||||
raise Exception(f"database discovery failed: {stderr.decode(errors='replace').strip()[:300]}")
|
||||
return [line.strip() for line in stdout.decode().splitlines() if line.strip()]
|
||||
|
||||
|
||||
async def _stream_subprocess_to_file(args: List[str], destination: Path, timeout: int) -> None:
|
||||
"""Run args, writing its stdout to destination as it arrives, within timeout.
|
||||
|
||||
Same create-subprocess-then-wait_for-then-kill shape as
|
||||
docker_prune_executor._run; the body differs because a dump's stdout is a
|
||||
binary stream that can run into the hundreds of MB, not the few lines of
|
||||
text `communicate()` holds comfortably — so this reads and writes it in
|
||||
chunks instead of buffering it whole.
|
||||
"""
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*args,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
|
||||
async def _drain() -> bytes:
|
||||
with open(destination, "wb") as f:
|
||||
while True:
|
||||
chunk = await proc.stdout.read(1 << 20)
|
||||
if not chunk:
|
||||
break
|
||||
f.write(chunk)
|
||||
return await proc.stderr.read()
|
||||
|
||||
try:
|
||||
stderr = await asyncio.wait_for(_drain(), timeout=timeout)
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
await proc.wait()
|
||||
raise Exception(f"timed out after {timeout}s: {' '.join(args)}")
|
||||
|
||||
rc = await proc.wait()
|
||||
if rc != 0:
|
||||
raise Exception(f"exited {rc}: {stderr.decode(errors='replace').strip()[:300]}")
|
||||
|
||||
|
||||
async def _dump_database(container: str, db: str, destination: Path, timeout: int) -> None:
|
||||
await _stream_subprocess_to_file(
|
||||
["docker", "exec", container, "pg_dump", "-U", "postgres", "-Fc", db],
|
||||
destination, timeout,
|
||||
)
|
||||
|
||||
|
||||
async def _dump_globals(container: str, destination: Path, timeout: int) -> None:
|
||||
await _stream_subprocess_to_file(
|
||||
["docker", "exec", container, "pg_dumpall", "-U", "postgres", "--globals-only"],
|
||||
destination, timeout,
|
||||
)
|
||||
|
||||
|
||||
def _prune(output_dir: Path, retention_days: int) -> int:
|
||||
"""Delete this executor's own files past the retention window."""
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
|
||||
removed = 0
|
||||
for path in output_dir.iterdir():
|
||||
if not path.is_file() or not FILENAME_RE.match(path.name):
|
||||
continue
|
||||
if datetime.fromtimestamp(path.stat().st_mtime, timezone.utc) < cutoff:
|
||||
path.unlink()
|
||||
removed += 1
|
||||
logger.info("pruned old postgres dump: %s", path.name)
|
||||
return removed
|
||||
|
||||
|
||||
async def _run(config: dict, settings: Settings) -> str:
|
||||
container = config.get("container", "")
|
||||
if not container:
|
||||
raise ValueError("Missing required config: 'container'")
|
||||
|
||||
output_dir = Path(config.get("output_dir", DEFAULT_OUTPUT_DIR))
|
||||
exclude = set(config.get("exclude") or [])
|
||||
want_globals = bool(config.get("globals", True))
|
||||
|
||||
retention_days = config.get("retention_days", DEFAULT_RETENTION_DAYS)
|
||||
if not isinstance(retention_days, int) or isinstance(retention_days, bool) or retention_days < 1:
|
||||
raise ValueError(f"retention_days must be a positive integer, got {retention_days!r}")
|
||||
|
||||
timeout_per_db = config.get("timeout_per_db", DEFAULT_TIMEOUT_PER_DB)
|
||||
if not isinstance(timeout_per_db, int) or isinstance(timeout_per_db, bool) or timeout_per_db < 1:
|
||||
raise ValueError(f"timeout_per_db must be a positive integer, got {timeout_per_db!r}")
|
||||
|
||||
databases_cfg = config.get("databases", "all")
|
||||
if databases_cfg == "all":
|
||||
discovered = await _discover_databases(container)
|
||||
elif isinstance(databases_cfg, list) and all(isinstance(d, str) for d in databases_cfg):
|
||||
discovered = databases_cfg
|
||||
else:
|
||||
raise ValueError(f"'databases' must be \"all\" or a list of names, got {databases_cfg!r}")
|
||||
|
||||
targets = [d for d in discovered if d not in exclude]
|
||||
if not targets and not want_globals:
|
||||
raise ValueError("no databases selected (all excluded) and globals disabled; nothing to do")
|
||||
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
||||
started = time.monotonic()
|
||||
|
||||
succeeded: List[str] = []
|
||||
failed: List[str] = []
|
||||
|
||||
for db in targets:
|
||||
final = output_dir / f"{db}-{stamp}.dump"
|
||||
partial = output_dir / f"{final.name}.partial"
|
||||
try:
|
||||
await _dump_database(container, db, partial, timeout_per_db)
|
||||
entries = await _verify_archive(partial)
|
||||
partial.replace(final)
|
||||
final.chmod(0o600)
|
||||
size_mb = final.stat().st_size / 1_048_576
|
||||
succeeded.append(f"{db} ({entries} entries, {size_mb:.2f} MB)")
|
||||
except Exception as exc: # noqa: BLE001 - one database's failure must not stop the rest
|
||||
logger.error("pg_dump failed for %s: %s", db, exc)
|
||||
failed.append(f"FAILED {db}: {exc}")
|
||||
finally:
|
||||
if partial.exists():
|
||||
partial.unlink()
|
||||
|
||||
if want_globals:
|
||||
final = output_dir / f"globals-{stamp}.sql"
|
||||
partial = output_dir / f"{final.name}.partial"
|
||||
try:
|
||||
await _dump_globals(container, partial, timeout_per_db)
|
||||
# pg_restore --list cannot read a plain-SQL globals dump; a
|
||||
# non-empty file is the only check available for it.
|
||||
if partial.stat().st_size == 0:
|
||||
raise Exception("globals dump is empty")
|
||||
partial.replace(final)
|
||||
final.chmod(0o600) # role password hashes live in here
|
||||
succeeded.append(f"globals ({final.stat().st_size} bytes)")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.error("globals dump failed: %s", exc)
|
||||
failed.append(f"FAILED globals: {exc}")
|
||||
finally:
|
||||
if partial.exists():
|
||||
partial.unlink()
|
||||
|
||||
removed = _prune(output_dir, retention_days)
|
||||
kept = len([p for p in output_dir.iterdir() if p.is_file() and FILENAME_RE.match(p.name)])
|
||||
elapsed = time.monotonic() - started
|
||||
total = len(targets) + (1 if want_globals else 0)
|
||||
|
||||
summary = (
|
||||
f"dumped {len(succeeded)}/{total} targets in {elapsed:.1f}s "
|
||||
f"({'; '.join(succeeded + failed)}); kept {kept}, pruned {removed} older than {retention_days}d"
|
||||
)
|
||||
logger.info(summary)
|
||||
|
||||
if failed:
|
||||
raise Exception(summary)
|
||||
return summary
|
||||
|
||||
|
||||
async def execute(config: dict, settings: Settings) -> str:
|
||||
"""Run the dump and report its own outcome to check_history (D-33, T-69).
|
||||
|
||||
The report wraps the work rather than living inside it, so the failure
|
||||
path cannot be forgotten: an exception is reported as critical and then
|
||||
re-raised, leaving the task's own status untouched. Reporting only
|
||||
success would reproduce exactly the blind spot this replaces — a monitor
|
||||
that cannot tell a failed backup from one that has not run.
|
||||
"""
|
||||
try:
|
||||
output = await _run(config, settings)
|
||||
except Exception as exc:
|
||||
await health_report.report_async(
|
||||
settings,
|
||||
domain="backup",
|
||||
status=health_report.CRITICAL,
|
||||
source="scheduler/pg_dump_executor",
|
||||
summary=f"backup failed: {str(exc)[:300]}",
|
||||
metrics={"job": "scheduler/pg_dump_executor", "error": str(exc)[:400]},
|
||||
)
|
||||
raise
|
||||
await health_report.report_async(
|
||||
settings,
|
||||
domain="backup",
|
||||
status=health_report.OK,
|
||||
source="scheduler/pg_dump_executor",
|
||||
summary=output[:400],
|
||||
metrics={"job": "scheduler/pg_dump_executor"},
|
||||
)
|
||||
return output
|
||||
@@ -0,0 +1,364 @@
|
||||
"""
|
||||
Tests for the postgres dump executor.
|
||||
|
||||
The point of this executor is a backup that is actually restorable, so most
|
||||
of these cover the failure paths: an empty TOC behind a successful dump, one
|
||||
database failing while the others are kept, retention touching only its own
|
||||
files, and a per-database timeout actually firing. Every subprocess call is
|
||||
mocked — this suite must pass with postgres-shared unreachable (D-26).
|
||||
"""
|
||||
import asyncio
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from src.config import Settings
|
||||
from src.executors import pg_dump_executor as pde
|
||||
|
||||
|
||||
def _config(tmp_path, **over):
|
||||
cfg = {
|
||||
"container": "postgres-shared",
|
||||
"databases": "all",
|
||||
"exclude": [],
|
||||
"globals": False, # most tests don't exercise the globals path; see TestGlobals
|
||||
"output_dir": str(tmp_path),
|
||||
"retention_days": 14,
|
||||
"timeout_per_db": 900,
|
||||
}
|
||||
cfg.update(over)
|
||||
return cfg
|
||||
|
||||
|
||||
async def _writing_dump(container, db, destination, timeout):
|
||||
"""A fake _dump_database that behaves like the real one: it writes bytes
|
||||
to the destination it is given."""
|
||||
destination.write_bytes(f"dump-of-{db}".encode())
|
||||
|
||||
|
||||
@pytest.mark.executor
|
||||
@pytest.mark.unit
|
||||
class TestConfigValidation:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_container_rejected(self, test_settings: Settings, tmp_path):
|
||||
with pytest.raises(ValueError, match="container"):
|
||||
await pde.execute(_config(tmp_path, container=""), test_settings)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("bad", [0, -1, "14", None, True])
|
||||
async def test_bad_retention_rejected(self, bad, test_settings: Settings, tmp_path):
|
||||
with pytest.raises(ValueError, match="retention_days"):
|
||||
await pde.execute(_config(tmp_path, retention_days=bad), test_settings)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("bad", [0, -1, "900", None, True])
|
||||
async def test_bad_timeout_per_db_rejected(self, bad, test_settings: Settings, tmp_path):
|
||||
with pytest.raises(ValueError, match="timeout_per_db"):
|
||||
await pde.execute(_config(tmp_path, timeout_per_db=bad), test_settings)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bad_databases_value_rejected(self, test_settings: Settings, tmp_path):
|
||||
with pytest.raises(ValueError, match="databases"):
|
||||
await pde.execute(_config(tmp_path, databases={"not": "valid"}), test_settings)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_everything_excluded_and_no_globals_rejected(self, test_settings: Settings, tmp_path):
|
||||
with patch.object(pde, "_discover_databases", AsyncMock(return_value=["a", "b"])):
|
||||
with pytest.raises(ValueError, match="nothing to do"):
|
||||
await pde.execute(
|
||||
_config(tmp_path, exclude=["a", "b"], globals=False), test_settings
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.executor
|
||||
@pytest.mark.unit
|
||||
class TestDiscovery:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_databases_parses_rows(self):
|
||||
proc = MagicMock()
|
||||
proc.communicate = AsyncMock(return_value=(b"alpha\nbeta\n\n", b""))
|
||||
proc.returncode = 0
|
||||
with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)) as mock_exec:
|
||||
result = await pde._discover_databases("postgres-shared")
|
||||
assert result == ["alpha", "beta"]
|
||||
args = mock_exec.call_args.args
|
||||
assert args[:3] == ("docker", "exec", "postgres-shared")
|
||||
assert "psql" in args
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_databases_failure_raises(self):
|
||||
proc = MagicMock()
|
||||
proc.communicate = AsyncMock(return_value=(b"", b"connection refused"))
|
||||
proc.returncode = 1
|
||||
with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
|
||||
with pytest.raises(Exception, match="database discovery failed"):
|
||||
await pde._discover_databases("postgres-shared")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_all_minus_excludes_selects_correct_targets(self, test_settings: Settings, tmp_path):
|
||||
attempted = []
|
||||
|
||||
async def fake_dump(container, db, destination, timeout):
|
||||
attempted.append(db)
|
||||
destination.write_bytes(b"x")
|
||||
|
||||
with patch.object(pde, "_discover_databases", AsyncMock(
|
||||
return_value=["keep", "template0", "test_scheduler"])), \
|
||||
patch.object(pde, "_dump_database", fake_dump), \
|
||||
patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
|
||||
await pde.execute(
|
||||
_config(tmp_path, exclude=["template0", "test_scheduler"]), test_settings
|
||||
)
|
||||
|
||||
assert attempted == ["keep"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_explicit_database_list_also_respects_exclude(self, test_settings: Settings, tmp_path):
|
||||
attempted = []
|
||||
|
||||
async def fake_dump(container, db, destination, timeout):
|
||||
attempted.append(db)
|
||||
destination.write_bytes(b"x")
|
||||
|
||||
with patch.object(pde, "_discover_databases", AsyncMock()) as discover, \
|
||||
patch.object(pde, "_dump_database", fake_dump), \
|
||||
patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
|
||||
await pde.execute(
|
||||
_config(tmp_path, databases=["one", "two"], exclude=["two"]), test_settings
|
||||
)
|
||||
|
||||
discover.assert_not_called() # explicit list means no discovery call at all
|
||||
assert attempted == ["one"]
|
||||
|
||||
|
||||
@pytest.mark.executor
|
||||
@pytest.mark.unit
|
||||
class TestDumpBehaviour:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_writes_verified_archive_per_database_mode_0600(self, test_settings: Settings, tmp_path):
|
||||
with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha", "beta"])), \
|
||||
patch.object(pde, "_dump_database", _writing_dump), \
|
||||
patch.object(pde, "_verify_archive", AsyncMock(return_value=5)):
|
||||
result = await pde.execute(_config(tmp_path), test_settings)
|
||||
|
||||
files = sorted(tmp_path.glob("*.dump"))
|
||||
assert [f.name.split("-")[0] for f in files] == ["alpha", "beta"]
|
||||
assert "5 entries" in result
|
||||
for f in files:
|
||||
assert oct(f.stat().st_mode)[-3:] == "600"
|
||||
# no .partial left behind
|
||||
assert list(tmp_path.glob("*.partial")) == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_one_failing_database_fails_run_but_keeps_others(self, test_settings: Settings, tmp_path):
|
||||
async def fake_dump(container, db, destination, timeout):
|
||||
if db == "bad":
|
||||
raise Exception("pg_dump exploded")
|
||||
destination.write_bytes(b"x")
|
||||
|
||||
with patch.object(pde, "_discover_databases", AsyncMock(return_value=["good", "bad"])), \
|
||||
patch.object(pde, "_dump_database", fake_dump), \
|
||||
patch.object(pde, "_verify_archive", AsyncMock(return_value=3)):
|
||||
with pytest.raises(Exception, match="FAILED bad: pg_dump exploded"):
|
||||
await pde.execute(_config(tmp_path), test_settings)
|
||||
|
||||
files = list(tmp_path.glob("*.dump"))
|
||||
assert len(files) == 1
|
||||
assert files[0].name.startswith("good-")
|
||||
assert list(tmp_path.glob("*.partial")) == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_toc_deletes_partial_and_counts_as_failure(self, test_settings: Settings, tmp_path):
|
||||
async def fake_verify(path):
|
||||
raise Exception("archive has zero TOC entries")
|
||||
|
||||
with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
|
||||
patch.object(pde, "_dump_database", _writing_dump), \
|
||||
patch.object(pde, "_verify_archive", fake_verify):
|
||||
with pytest.raises(Exception, match="FAILED alpha: archive has zero TOC entries"):
|
||||
await pde.execute(_config(tmp_path), test_settings)
|
||||
|
||||
assert list(tmp_path.glob("*.dump")) == []
|
||||
assert list(tmp_path.glob("*.partial")) == []
|
||||
|
||||
|
||||
@pytest.mark.executor
|
||||
@pytest.mark.unit
|
||||
class TestGlobals:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_globals_dump_written_and_mode_0600(self, test_settings: Settings, tmp_path):
|
||||
async def fake_globals(container, destination, timeout):
|
||||
destination.write_bytes(b"CREATE ROLE postgres;")
|
||||
|
||||
with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
|
||||
patch.object(pde, "_dump_database", _writing_dump), \
|
||||
patch.object(pde, "_verify_archive", AsyncMock(return_value=2)), \
|
||||
patch.object(pde, "_dump_globals", fake_globals):
|
||||
result = await pde.execute(_config(tmp_path, globals=True), test_settings)
|
||||
|
||||
files = list(tmp_path.glob("globals-*.sql"))
|
||||
assert len(files) == 1
|
||||
assert oct(files[0].stat().st_mode)[-3:] == "600"
|
||||
assert "globals" in result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_globals_dump_is_a_failure_but_keeps_db_dumps(self, test_settings: Settings, tmp_path):
|
||||
async def fake_globals(container, destination, timeout):
|
||||
destination.write_bytes(b"") # empty
|
||||
|
||||
with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
|
||||
patch.object(pde, "_dump_database", _writing_dump), \
|
||||
patch.object(pde, "_verify_archive", AsyncMock(return_value=2)), \
|
||||
patch.object(pde, "_dump_globals", fake_globals):
|
||||
with pytest.raises(Exception, match="FAILED globals: globals dump is empty"):
|
||||
await pde.execute(_config(tmp_path, globals=True), test_settings)
|
||||
|
||||
assert len(list(tmp_path.glob("*.dump"))) == 1 # alpha kept
|
||||
assert list(tmp_path.glob("globals-*.sql")) == []
|
||||
assert list(tmp_path.glob("*.partial")) == []
|
||||
|
||||
|
||||
@pytest.mark.executor
|
||||
@pytest.mark.unit
|
||||
class TestVerifyArchive:
|
||||
|
||||
def test_toc_entry_count_counts_non_comment_lines(self):
|
||||
output = (
|
||||
";\n"
|
||||
"; Archive created at 2026-10-19 09:45:07 UTC\n"
|
||||
"; TOC Entries: 2\n"
|
||||
";\n"
|
||||
"216; 1259 44268 TABLE public scheduled_tasks test_scheduler_user\n"
|
||||
"215; 1259 44267 SEQUENCE public scheduled_tasks_id_seq test_scheduler_user\n"
|
||||
)
|
||||
assert pde._toc_entry_count(output) == 2
|
||||
|
||||
def test_toc_entry_count_zero_for_header_only(self):
|
||||
output = (
|
||||
";\n"
|
||||
"; Archive created at 2026-10-19 09:45:07 UTC\n"
|
||||
"; TOC Entries: 0\n"
|
||||
";\n"
|
||||
)
|
||||
assert pde._toc_entry_count(output) == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verify_archive_raises_on_nonzero_exit(self, tmp_path):
|
||||
proc = MagicMock()
|
||||
proc.communicate = AsyncMock(return_value=(b"", b"not an archive"))
|
||||
proc.returncode = 1
|
||||
with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
|
||||
with pytest.raises(Exception, match="pg_restore --list failed"):
|
||||
await pde._verify_archive(tmp_path / "broken.dump")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verify_archive_raises_on_zero_entries(self, tmp_path):
|
||||
proc = MagicMock()
|
||||
proc.communicate = AsyncMock(return_value=(b";\n; TOC Entries: 0\n;\n", b""))
|
||||
proc.returncode = 0
|
||||
with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
|
||||
with pytest.raises(Exception, match="zero TOC entries"):
|
||||
await pde._verify_archive(tmp_path / "empty.dump")
|
||||
|
||||
|
||||
@pytest.mark.executor
|
||||
@pytest.mark.unit
|
||||
class TestTimeout:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dump_times_out_per_database(self, tmp_path):
|
||||
"""The actual asyncio.wait_for/kill mechanism, not a stand-in for it."""
|
||||
|
||||
class HangingStdout:
|
||||
async def read(self, n):
|
||||
await asyncio.sleep(10)
|
||||
return b""
|
||||
|
||||
class EmptyStderr:
|
||||
async def read(self):
|
||||
return b""
|
||||
|
||||
fake_proc = MagicMock()
|
||||
fake_proc.stdout = HangingStdout()
|
||||
fake_proc.stderr = EmptyStderr()
|
||||
fake_proc.kill = MagicMock()
|
||||
fake_proc.wait = AsyncMock(return_value=0)
|
||||
|
||||
with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=fake_proc)):
|
||||
with pytest.raises(Exception, match="timed out after"):
|
||||
await pde._dump_database(
|
||||
"postgres-shared", "alpha", tmp_path / "alpha.partial", timeout=0.05
|
||||
)
|
||||
|
||||
fake_proc.kill.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timeout_failure_handled_like_any_other_db_failure(self, test_settings: Settings, tmp_path):
|
||||
async def fake_dump(container, db, destination, timeout):
|
||||
if db == "slow":
|
||||
raise Exception(f"timed out after {timeout}s: pg_dump")
|
||||
destination.write_bytes(b"x")
|
||||
|
||||
with patch.object(pde, "_discover_databases", AsyncMock(return_value=["fast", "slow"])), \
|
||||
patch.object(pde, "_dump_database", fake_dump), \
|
||||
patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
|
||||
with pytest.raises(Exception, match="FAILED slow: timed out after"):
|
||||
await pde.execute(_config(tmp_path, timeout_per_db=1), test_settings)
|
||||
|
||||
files = list(tmp_path.glob("*.dump"))
|
||||
assert len(files) == 1 and files[0].name.startswith("fast-")
|
||||
|
||||
|
||||
@pytest.mark.executor
|
||||
@pytest.mark.unit
|
||||
class TestRetention:
|
||||
|
||||
def _age(self, path, days):
|
||||
import os
|
||||
old = (datetime.now(timezone.utc) - timedelta(days=days)).timestamp()
|
||||
os.utime(path, (old, old))
|
||||
|
||||
def test_prunes_only_past_the_window(self, tmp_path):
|
||||
fresh = tmp_path / "alpha-20261019T094500Z.dump"
|
||||
stale = tmp_path / "alpha-20260101T120000Z.dump"
|
||||
for f in (fresh, stale):
|
||||
f.write_bytes(b"x")
|
||||
self._age(stale, 45)
|
||||
|
||||
assert pde._prune(tmp_path, 14) == 1
|
||||
assert fresh.exists() and not stale.exists()
|
||||
|
||||
def test_leaves_unrelated_files_alone(self, tmp_path):
|
||||
"""Retention must not touch anything it did not write."""
|
||||
other = tmp_path / "important-database-dump.tar.gz"
|
||||
named_alike = tmp_path / "alpha-backup-manual.dump"
|
||||
for f in (other, named_alike):
|
||||
f.write_bytes(b"x")
|
||||
self._age(f, 400)
|
||||
|
||||
assert pde._prune(tmp_path, 14) == 0
|
||||
assert other.exists() and named_alike.exists()
|
||||
|
||||
def test_prune_leaves_partial_files_alone(self, tmp_path):
|
||||
"""A .partial is never this executor's finished output; retention
|
||||
must never be the thing that deletes one out from under a dump that
|
||||
is still writing it."""
|
||||
partial = tmp_path / "alpha-20260101T120000Z.dump.partial"
|
||||
partial.write_bytes(b"x")
|
||||
self._age(partial, 400)
|
||||
|
||||
assert pde._prune(tmp_path, 14) == 0
|
||||
assert partial.exists()
|
||||
|
||||
def test_prunes_globals_files_too(self, tmp_path):
|
||||
stale = tmp_path / "globals-20260101T120000Z.sql"
|
||||
stale.write_bytes(b"x")
|
||||
self._age(stale, 400)
|
||||
|
||||
assert pde._prune(tmp_path, 14) == 1
|
||||
assert not stale.exists()
|
||||
Reference in New Issue
Block a user