"""Paired submitted/candidate filesystem behavior; real gzip, no live server. Real restoreability is checked separately by the container topology scenarios. These tests compare observable byte inventories and explicitly name intentional differences rather than accepting arbitrary differences as improvements. """ import gzip import hashlib import os from pathlib import Path import shutil import subprocess import tempfile import unittest class BaselineRuntime(unittest.TestCase): @classmethod def setUpClass(cls): baseline = os.environ.get("SDP_BASELINE_SOURCE") if not baseline: raise unittest.SkipTest("Set SDP_BASELINE_SOURCE for the required paired comparison stage") cls.baseline = Path(baseline) cls.candidate = Path(os.environ.get("SDP_CANDIDATE_SOURCE", Path(__file__).resolve().parents[1])) for root in (cls.baseline, cls.candidate): if not (root / "Server/Unix/p4/common/bin/backup_functions.sh").is_file(): raise RuntimeError("Missing submitted/candidate backup implementation: " + str(root)) cls.bash = os.environ.get("SDP_TEST_BASH", shutil.which("bash")) def run_side(self, source, fixture, operation, variables=None, fail_gzip=False): with tempfile.TemporaryDirectory(prefix="sdp-paired-runtime-") as temp: root = Path(temp) for name in ("journals", "journals.rotated", "checkpoints", "logs", "offline_db", "mockbin"): (root / name).mkdir() (root / "journals/journal").write_bytes(b"live journal must survive\n") (root / "site-config").write_bytes(b"custom-instance-value\n") fixture(root) environment = dict(os.environ, TEST_ROOT=str(root), P4CBIN=str(source / "Server/Unix/p4/common/bin"), LOGS=str(root / "logs"), LOGFILE=str(root / "logs/run.log"), JOURNALS=str(root / "journals.rotated"), CHECKPOINTS=str(root / "checkpoints"), P4JOURNAL=str(root / "journals/journal"), P4CKP_PREFIX="site.prefix", KEEPCKPS="2", KEEPJNLS="2", KEEPLOGS="2", OFFLINEJNLNUM="4", AWK="awk") environment.update(variables or {}) if fail_gzip: fake = root / "mockbin/gzip" fake.write_text("#!/bin/sh\nexit 13\n") fake.chmod(0o755) environment["PATH"] = str(fake.parent) + ":" + environment["PATH"] code = '''source "$P4CBIN/backup_functions.sh" OFFLINE_DB="$TEST_ROOT/offline_db" log () { printf '%s\\n' "$*" >> "$LOGFILE"; } die () { log "$*"; exit 97; } ''' + operation result = subprocess.run([self.bash, "-c", code], env=environment, cwd=root, capture_output=True, text=True, timeout=30) inventory = {} for file in root.rglob("*"): if not file.is_file() or file.is_relative_to(root / "logs") or file.is_relative_to(root / "mockbin"): continue data = gzip.decompress(file.read_bytes()) if file.suffix == ".gz" else file.read_bytes() inventory[file.relative_to(root).as_posix()] = hashlib.sha256(data).hexdigest() return result.returncode, inventory, result.stderr def paired(self, fixture, operation, variables=None, fail_gzip=False): return tuple(self.run_side(source, fixture, operation, variables, fail_gzip) for source in (self.baseline, self.candidate)) @staticmethod def journal_fixture(count): def populate(root): for number in range(1, count + 1): (root / ("journals.rotated/site.prefix.jnl.%d" % number)).write_text("journal payload %d\n" % number) (root / "journals.rotated/site.prefix-other.jnl.1").write_text("another instance\n") return populate def test_gzip_threshold_and_payloads_match_submitted_behavior(self): for count in (0, 1, 2, 3, 4, 5, 10): with self.subTest(raw_journals=count): old, new = self.paired(self.journal_fixture(count), "gzip_mv_journals") self.assertEqual(old[:2], new[:2]) self.assertEqual(new[0], 0, new[2]) moved = 0 if count <= 2 else 1 if count < 5 else count - 4 archived = [p for p in new[1] if p.startswith("checkpoints/")] self.assertEqual(len(archived), moved) self.assertIn("journals/journal", new[1]) self.assertIn("site-config", new[1]) self.assertIn("journals.rotated/site.prefix-other.jnl.1", new[1]) def test_failed_compression_preserves_same_inputs_and_failure_status(self): old, new = self.paired(self.journal_fixture(5), "gzip_mv_journals", fail_gzip=True) self.assertEqual(old[:2], new[:2]) self.assertEqual(new[0], 97) self.assertEqual(sum(p.startswith("journals.rotated/") for p in new[1]), 6) @staticmethod def retention_fixture(root): for number in (1, 2, 3): checkpoint = root / ("checkpoints/site.prefix.ckp.%d.gz" % number) checkpoint.write_bytes(gzip.compress(("checkpoint fixture %d" % number).encode())) Path(str(checkpoint) + ".md5").write_text("fixture sidecar\n") for number in (1, 2, 3, 4): (root / ("checkpoints/site.prefix.jnl.%d.gz" % number)).write_bytes(gzip.compress(("journal %d" % number).encode())) (root / "checkpoints/unrelated.txt").write_text("preserve unrelated content\n") def test_disabled_retention_preserves_identical_bytes_and_configuration(self): old, new = self.paired(self.retention_fixture, "remove_old_checkpoints_and_journals", {"KEEPCKPS": "0"}) self.assertEqual(old[:2], new[:2]) self.assertEqual(new[0], 0, new[2]) self.assertEqual(len(new[1]), 13) def test_approved_retention_difference_keeps_oldest_retained_checkpoint_chain(self): old, new = self.paired(self.retention_fixture, "remove_old_checkpoints_and_journals") self.assertEqual((old[0], new[0]), (0, 0), (old[2], new[2])) # Approved change: KEEPJNLS alone formerly removed journal 2 although # checkpoint 2 survives. The candidate must retain exactly that extra # chain member for this fixture; no other unexplained difference passes. extra = "checkpoints/site.prefix.jnl.2.gz" self.assertEqual(set(new[1]) - set(old[1]), {extra}) self.assertEqual(set(old[1]) - set(new[1]), set()) self.assertEqual(old[1], {k: v for k, v in new[1].items() if k != extra}) self.assertNotIn("checkpoints/site.prefix.ckp.1.gz", new[1]) if __name__ == "__main__": unittest.main()