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:
2026-10-19 10:54:43 +02:00
co-authored by Claude Sonnet 5
parent f74e6b2b98
commit a14f134bfc
3 changed files with 683 additions and 0 deletions
+8
View File
@@ -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
+311
View File
@@ -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
+364
View File
@@ -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()