from copy import deepcopy import hashlib import json import os from pathlib import Path import tempfile import subprocess import unittest from unittest.mock import patch from backup_core import BackupJob, WRITERS from transport import ComposeStack, DockerRepository, OperationsError SQL = b"CREATE TABLE member (id INTEGER PRIMARY KEY, username TEXT); INSERT INTO member VALUES (7, 'synthetic-author');\n" RDB = b"synthetic-redis-snapshot-fixture" class FakeStack: def __init__(self, root): self.project = "synthetic-post76" self.env_file = Path(root) / "synthetic.env" self.env_file.write_text("APP_ENV=testing\n") self.environment = {"POST76_BACKUP_CAPTURE_DIR": str(Path(root) / "capture")} self.calls = [] self.failures = set() self.states = {name: {"State": "running", "Health": "healthy"} for name in [*WRITERS, "mysql", "redis", "meilisearch"]} self.images = {name: {"image": "post76-" + name + ":ci"} for name in [*WRITERS, "mysql"]} self.images["mysql"]["environment"] = {"MYSQL_DATABASE": "post76"} def model(self): return {"services": deepcopy(self.images)} def state(self): return deepcopy(self.states) def stop(self, services): self.calls.append(("stop", list(services))) if "stop" in self.failures: raise OperationsError("synthetic stop failure") def resume(self, services): self.calls.append(("resume", list(services))) if "resume" in self.failures: raise OperationsError("synthetic resume failure") def dump(self, destination): self.calls.append(("dump",)) Path(destination).write_bytes(SQL) Path(destination).chmod(0o600) if "dump" in self.failures: raise OperationsError("synthetic dump failure") def mounts(self, capture, restore): self.environment.update({"POST76_BACKUP_CAPTURE_DIR": str(capture), "POST76_BACKUP_RESTORE_DIR": str(restore)}) def snapshot_redis(self): self.calls.append(("snapshot-redis",)) if "redis" in self.failures: raise OperationsError("synthetic Redis snapshot failure") class FakeRepository: def __init__(self): self.roots = DockerRepository(None).roots self.failures = set() self.rows = [] self.calls = [] self.node_values = [{"struct_type": "node", "type": "file", "path": "/capture/database.sql"}] def snapshots(self, tags): self.calls.append(("snapshots", tags)) if "repository" in self.failures: raise OperationsError("synthetic repository failure") return [row for row in self.rows if set(tags).issubset(row["tags"])] def capture(self, capture, tags): self.metadata = json.loads((Path(capture) / "manifest.json").read_text()) self.sql = (Path(capture) / "database.sql").read_bytes() self.calls.append(("capture",)) if "capture" in self.failures: raise OperationsError("synthetic partial backup") self.rows.append({"id": "a" * 64, "tags": list(tags)}) return "a" * 64 def verify_tag(self, identity): self.calls.append(("verify",)) if "verify" in self.failures: raise OperationsError("synthetic tag failure") self.rows[0]["tags"].append("verified") self.rows[0]["id"] = "b" * 64 return "b" * 64 def manifest(self, identity): return deepcopy(self.metadata) def nodes(self, identity): return deepcopy(self.node_values) def restore(self, identity, destination): self.calls.append(("restore",)) root = Path(destination) / "capture" root.mkdir() (root / "database.sql").write_bytes(self.sql) redis = Path(destination) / "redis" redis.mkdir() (redis / "dump.rdb").write_bytes(RDB) class BackupTest(unittest.TestCase): def setUp(self): parent = Path(__file__).resolve().parent.parent / "storage/app/private/backup-check" parent.mkdir(parents=True, exist_ok=True) self.temporary = tempfile.TemporaryDirectory(prefix="unit-", dir=parent) self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.capture = self.root / "capture" self.capture.mkdir(mode=0o700) self.stack = FakeStack(self.root) self.repository = FakeRepository() self.job = BackupJob(self.stack, self.repository, self.root / "workspace") def run_backup(self): return self.job.backup(self.capture, quiesce=True) def test_missing_explicit_maintenance_option_never_pauses_services(self): with self.assertRaises(OperationsError): self.job.backup(self.capture) self.assertEqual([], self.stack.calls) def test_unavailable_repository_never_pauses_services(self): self.repository.failures.add("repository") with self.assertRaises(OperationsError): self.run_backup() self.assertEqual([], self.stack.calls) self.assertFalse(json.loads(self.job.status.read_text())["success"]) def test_unhealthy_database_never_pauses_services(self): self.stack.states["mysql"]["Health"] = "unhealthy" with self.assertRaises(OperationsError): self.run_backup() self.assertEqual([], self.stack.calls) def test_success_resumes_only_originally_running_services_and_cleans_plaintext(self): self.stack.states["scheduler"]["State"] = "exited" self.assertEqual("b" * 64, self.run_backup()) expected = [name for name in WRITERS if name != "scheduler"] self.assertEqual(("stop", expected), self.stack.calls[0]) self.assertEqual(("resume", expected), self.stack.calls[-1]) self.assertFalse(self.job.journal.exists()) self.assertEqual([], list(self.capture.iterdir())) self.assertEqual(hashlib.sha256(SQL).hexdigest(), self.repository.metadata["database_sha256"]) self.assertTrue(json.loads(self.job.status.read_text())["success"]) def test_partial_pause_failure_still_resumes_original_services(self): self.stack.failures.add("stop") with self.assertRaises(OperationsError): self.run_backup() self.assertEqual(("resume", WRITERS), self.stack.calls[-1]) def test_partial_dump_failure_cannot_commit_a_snapshot(self): self.stack.failures.add("dump") with self.assertRaises(OperationsError): self.run_backup() self.assertNotIn(("capture",), self.repository.calls) self.assertEqual([], list(self.capture.iterdir())) self.assertEqual(("resume", WRITERS), self.stack.calls[-1]) def test_partial_file_capture_is_never_marked_verified(self): self.repository.failures.add("capture") with self.assertRaises(OperationsError): self.run_backup() self.assertNotIn(("verify",), self.repository.calls) self.assertEqual(("resume", WRITERS), self.stack.calls[-1]) def test_failed_queue_snapshot_never_commits_and_resumes_services(self): self.stack.failures.add("redis") with self.assertRaises(OperationsError): self.run_backup() self.assertNotIn(("capture",), self.repository.calls) self.assertEqual(("resume", WRITERS), self.stack.calls[-1]) self.assertEqual([], list(self.capture.iterdir())) def test_redis_error_reply_is_rejected_before_repository_capture(self): stack = ComposeStack(self.root, "synthetic-post76", self.stack.env_file, self.job.workspace) with patch.object(stack, "command", return_value=b"ERR synthetic failed SAVE\n"): with self.assertRaises(OperationsError): stack.snapshot_redis() def test_unverified_snapshot_is_refused_for_restore(self): self.repository.failures.add("verify") with self.assertRaises(OperationsError): self.run_backup() with self.assertRaises(OperationsError): self.job.restore("a" * 64, self.root / "recovery") self.assertNotIn(("restore",), self.repository.calls) def test_failed_resume_retains_journal_and_blocks_a_new_capture(self): self.stack.failures.add("resume") with self.assertRaises(OperationsError): self.run_backup() self.assertTrue(self.job.journal.exists()) self.assertTrue(json.loads(self.job.status.read_text())["success"]) with self.assertRaises(OperationsError): self.run_backup() self.stack.failures.clear() previous = json.loads(self.job.status.read_text())["last_success"] self.job.recover_services() self.assertFalse(self.job.journal.exists()) self.assertEqual(previous, json.loads(self.job.status.read_text())["last_success"]) def test_recovery_refuses_changed_deployment_images(self): self.stack.failures.add("resume") with self.assertRaises(OperationsError): self.run_backup() self.stack.images["app"]["image"] = "different:revision" with self.assertRaises(OperationsError): self.job.recover_services() self.assertTrue(self.job.journal.exists()) def test_cleanup_failure_still_resumes_services(self): original = Path.unlink def unlink(path, *args, **kwargs): if path.name == "database.sql": raise PermissionError("synthetic cleanup error") return original(path, *args, **kwargs) with patch.object(Path, "unlink", unlink): with self.assertRaises(OperationsError): self.run_backup() self.assertEqual(("resume", WRITERS), self.stack.calls[-1]) def test_recovery_refuses_a_changed_database_volume(self): self.stack.failures.add("resume") with self.assertRaises(OperationsError): self.run_backup() self.stack.images["mysql"]["volumes"] = [{"type": "volume", "source": "different-database", "target": "/var/lib/mysql"}] with self.assertRaises(OperationsError): self.job.recover_services() self.assertTrue(self.job.journal.exists()) def test_restore_checks_checksum_and_never_modifies_running_database(self): identity = self.run_backup() before = list(self.stack.calls) data = self.job.restore(identity, self.root / "recovery") self.assertEqual(SQL, (data / "capture/database.sql").read_bytes()) self.assertEqual(RDB, (data / "redis/dump.rdb").read_bytes()) self.assertEqual(before, self.stack.calls) self.assertFalse(json.loads((self.root / "recovery/restore-result.json").read_text())["production_database_modified"]) def test_restore_rejects_existing_destination_and_bad_integrity_metadata(self): identity = self.run_backup() with self.assertRaises(OperationsError): self.job.restore(identity, self.root) self.repository.metadata["database_sha256"] = "invalid" with self.assertRaises(OperationsError): self.job.restore(identity, self.root / "recovery") self.assertNotIn(("restore",), self.repository.calls) def test_restore_rejects_symlink_nodes_without_creating_links(self): identity = self.run_backup() self.repository.node_values = [{"struct_type": "node", "type": "symlink", "path": "/uploads/escape"}] with self.assertRaises(OperationsError): self.job.restore(identity, self.root / "recovery") self.assertNotIn(("restore",), self.repository.calls) def test_metrics_preserve_last_good_backup_after_failure(self): self.run_backup() previous = json.loads(self.job.status.read_text())["last_success"] self.repository.failures.add("repository") with self.assertRaises(OperationsError): self.run_backup() current = json.loads(self.job.status.read_text()) self.assertEqual(previous, current["last_success"]) self.assertFalse(current["success"]) self.assertEqual(0o644, (self.job.workspace / "metrics/backup.prom").stat().st_mode & 0o777) self.assertNotIn("synthetic-author", (self.job.workspace / "metrics/backup.prom").read_text()) def test_workspace_lock_refuses_overlapping_operations(self): with self.job.lock(): with self.assertRaises(OperationsError): with self.job.lock(): self.fail("Overlapping operation was accepted") def test_s3_wrapper_rejects_missing_default_credentials_before_any_backend_access(self): credentials = self.root / "synthetic-credentials" credentials.write_text("[other-profile]\naws_access_key_id=fake\naws_secret_access_key=fake\n") script = Path(__file__).resolve().parent.parent / "docker/backup/entrypoint.sh" result = subprocess.run(["sh", str(script), "snapshots"], env={"PATH": os.environ["PATH"], "RESTIC_REPOSITORY": "s3:https://example.invalid/test", "AWS_SHARED_CREDENTIALS_FILE": str(credentials)}, capture_output=True, text=True) self.assertEqual(1, result.returncode) self.assertIn("explicit default-profile", result.stderr) def test_legacy_capture_records_its_scope_and_sources(self): repository = DockerRepository(self.stack, legacy=True) self.assertEqual("/legacy/attachment", repository.roots["legacy_attachments"]) self.assertEqual("/legacy/avatar", repository.roots["legacy_avatars"]) self.repository.roots = repository.roots self.run_backup() self.assertEqual("captured", self.repository.metadata["legacy_files"]) def test_docker_transport_uses_local_socket_and_hides_failure_output(self): with patch.dict(os.environ, {"PATH": "/usr/bin", "DOCKER_HOST": "tcp://forbidden.invalid", "APP_KEY": "fake"}, clear=True): stack = ComposeStack(self.root, "synthetic-post76", self.stack.env_file, self.job.workspace) self.assertEqual("unix:///var/run/docker.sock", stack.environment["DOCKER_HOST"]) self.assertNotIn("APP_KEY", stack.environment) with patch("transport.subprocess.run", return_value=unittest.mock.Mock(returncode=12, stderr=b"synthetic sensitive output")): with self.assertRaises(OperationsError) as result: stack.command(["ps"], "Synthetic check") self.assertNotIn("sensitive", str(result.exception)) if __name__ == "__main__": unittest.main()