Files
scheduler/tests/test_pg_dump_executor.py
T
jpmschweitzerandClaude Sonnet 5 a14f134bfc 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>
2026-10-19 10:54:43 +02:00

365 lines
15 KiB
Python

"""
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()