test_baseline_runtime.py #1

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