test_release_gate.py #3

"""Acceptance must reject green process statuses with missing/failed evidence."""
import hashlib
import io
import json
from pathlib import Path
import sys
import tempfile
import tarfile
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parent / "containers"))
import release_gate as gate
import export_baseline
import export_archive
import baseline_comparison


def passed_scenario(name):
    row = {"name": name, "status": "passed"}
    if name == "sendq_cleanup":
        row["evidence"] = {"checks": [{"name": check, "status": "passed", "real_partitioned_have_preserved": True,
                                     "server_was_stopped": check != "running_server_refused",
                                     "required_alias_and_queue_preserved": True}
                                     for check in sorted(gate.SENDQ_CHECKS)],
                           "regular_files_and_symlinks_removed": True, "directories_and_fifo_preserved": True,
                           "nested_symlink_targets_preserved": True, "fixture_records_and_bytes_preserved": True}
    elif name == "database_switch":
        row["evidence"] = {"both_physical_roots_verified": True, "sendq_physical_roots_cleaned": ["/db1", "/db2"], "real_partitioned_have_preserved": True,
                           "readonly_path_contains_space_and_equals": True}
    return row


class ReleaseGate(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="sdp-release-gate-")
        self.addCleanup(self.temp.cleanup)
        self.folder = Path(self.temp.name)
        self.write("source-manifest.json", [{"path": "Server/Unix/setup/install_sdp_python.sh", "sha256": "candidate"}])
        self.write("baseline-source-manifest.json", [{"path": "source", "sha256": "baseline"}])
        self.candidate = gate.manifest_hash(self.folder / "source-manifest.json")
        self.baseline = gate.manifest_hash(self.folder / "baseline-source-manifest.json")
        self.write("test-outcomes.json", {stage: 0 for stage in gate.PRIMARY_STAGES + gate.DISTRO_STAGES})
        self.write("container-inspect.json", {"Image": "sha256:final", "State": {"Running": True}})
        self.write("image-inspect.json", [{"Id": "sha256:final"}])
        self.write("journal-path.txt", "/p4/1/journals/journal\n")
        self.write("p4d-process-argv.json", {"argv": ["p4d", "-J", "/p4/1/journals/journal"]})
        self.write("python-runtimes.json", {"distro_supported": True, "baseline": {"version": "3.14.8"}})
        self.write("python-installer-evidence.json", {"source_sha256": "candidate", "exit_code": 0, "build_os_user": "perforce"})
        self.write("installer-command.txt", "/source/mkdirs.sh /fixture/mkdirs.cfg\n")
        self.write("service-final-state.txt", "ActiveState=active\nMainPID=123\n")
        self.write("coverage.json", {"source_matches_results": True, "failed_stages": [], "scripts": [
            {"script": "backup_functions.sh", "level": "live", "outcome": "passed-for-listed-scenarios"}]})
        for prefix in ("", "distro-"):
            offline_rows = [name.rsplit(".", 1)[1] + " (" + name + ") ... ok" for name in sorted(gate.OFFLINE_CASES)]
            self.write(prefix + "offline.log", "\n".join(offline_rows) + "\nRan %d tests in 1s\nOK\n" % len(offline_rows))
            self.write(prefix + "integration.log", "\n".join("[PASS] " + name for name in sorted(gate.HARNESS_CASES)) + "\ntotal=7 pass=7 fail=0 skip=0\n")
            self.write(prefix + "runtime.log", {"scenarios": [passed_scenario(name) for name in sorted(gate.RUNTIME_CASES)]})
            self.write(prefix + "maintenance.log", {"scenarios": [{"name": name, "status": "passed"} for name in sorted(gate.MAINTENANCE_CASES)]})
            self.write(prefix + "baseline-comparison.json", {"scenarios": [{"name": name, "status": "passed"} for name in sorted(gate.BASELINE_CASES)],
                       "baseline_manifest_sha256": self.baseline, "candidate_manifest_sha256": self.candidate})

    def write(self, name, value):
        (self.folder / name).write_text(value if isinstance(value, str) else json.dumps(value))

    def check(self):
        return gate.platform_errors(self.folder, self.candidate, self.baseline)

    def test_complete_consistent_evidence_is_accepted(self):
        self.assertEqual(self.check(), [])

    def test_zero_stage_status_does_not_hide_failed_assertion(self):
        p = self.folder / "integration.log"
        p.write_text(p.read_text() + "[FAIL] fixture bytes differ\n")
        self.assertTrue(any("failed assertion" in e for e in self.check()))

    def test_green_summary_does_not_replace_required_behavioral_checks(self):
        self.write("integration.log", "[PASS] command exited\ntotal=1 pass=1 fail=0 skip=0\n")
        self.assertTrue(any("missing state/content assertion" in e for e in self.check()))

    def test_missing_and_failed_stage_each_reject(self):
        for status in (None, 7):
            stages = {s: 0 for s in gate.PRIMARY_STAGES + gate.DISTRO_STAGES}
            if status is None:
                stages.pop("baseline-comparison")
            else:
                stages["baseline-comparison"] = status
            self.write("test-outcomes.json", stages)
            self.assertTrue(self.check())

    def test_skips_and_absent_concurrency_regression_reject(self):
        for log in ("Ran 3 tests in 1s\nOK (skipped=1)\n", "Ran 3 tests in 1s\nOK\n"):
            self.write("offline.log", log)
            self.assertTrue(self.check())

    def test_expected_application_error_logs_are_distinct_from_unittest_failure_headers(self):
        original = (self.folder / "offline.log").read_text()
        self.write("offline.log", "ERROR:root:example.py failed (exit 7): denied\n" + original)
        self.assertEqual(self.check(), [])
        for header in ("ERROR: test_failure (test_module.TestCase)\n", "FAIL: test_failure (test_module.TestCase)\n", "ERROR: setUpClass (test_module.TestCase)\n", "FAILED (failures=1)\n"):
            with self.subTest(header=header):
                self.write("offline.log", original + header)
                self.assertTrue(any("failed or skipped test" in error for error in self.check()))

    def test_module_mentions_and_expected_failures_cannot_replace_passed_tests(self):
        self.write("offline.log", "test_logging_concurrency test_baseline_runtime test_baseline_maintenance\nRan 3 tests in 1s\nOK\n")
        self.assertTrue(self.check())
        rows = [name.rsplit(".", 1)[1] + " (" + name + ") ... expected failure" for name in sorted(gate.OFFLINE_CASES)]
        self.write("offline.log", "\n".join(rows) + "\nRan 22 tests in 1s\nOK\n")
        self.assertTrue(self.check())

    def test_missing_recovery_completion_regression_rejects_each_runtime(self):
        required = {name for name in gate.OFFLINE_CASES if name.startswith("test_recovery_completion.")}
        self.assertGreaterEqual(len(required), 10)
        for filename in ("offline.log", "distro-offline.log"):
            original = (self.folder / filename).read_text()
            for name in sorted(required):
                with self.subTest(filename=filename, missing=name):
                    self.write(filename, "\n".join(line for line in original.splitlines() if name not in line) + "\n")
                    self.assertTrue(any(name in error for error in self.check()))
            self.write(filename, original)

    def test_arbitrary_passing_case_does_not_replace_live_or_baseline_contracts(self):
        for report in ("maintenance.log", "baseline-comparison.json"):
            original = (self.folder / report).read_text()
            self.write(report, {"scenarios": [{"name": "anything", "status": "passed"}],
                       "baseline_manifest_sha256": self.baseline, "candidate_manifest_sha256": self.candidate})
            self.assertTrue(any("missing required scenario" in e for e in self.check()))
            self.write(report, original)

    def test_rebuilt_image_or_wrong_journal_path_reject(self):
        self.write("image-inspect.json", [{"Id": "sha256:new-unexercised-image"}])
        self.assertTrue(any("final image" in e for e in self.check()))
        self.write("image-inspect.json", [{"Id": "sha256:final"}])
        self.write("journal-path.txt", "/p4/1/logs/journal\n")
        self.assertTrue(any("journal-path" in e for e in self.check()))

    def test_failed_or_omitted_recovery_scenario_rejects(self):
        for scenario in ({"name": "database_switch", "status": "failed"}, {"name": "other", "status": "passed"}):
            self.write("runtime.log", {"scenarios": [scenario]})
            self.assertTrue(self.check())

    def test_sendq_green_label_cannot_replace_required_scope_and_preservation_checks(self):
        for name in ("sendq_cleanup", "database_switch"):
            rows = [passed_scenario(case) for case in sorted(gate.RUNTIME_CASES)]
            next(row for row in rows if row["name"] == name).pop("evidence")
            self.write("runtime.log", {"scenarios": rows})
            self.assertTrue(self.check(), name)
        for mutation in ("missing", "failed", "client_lost", "running", "alias"):
            rows = [passed_scenario(case) for case in sorted(gate.RUNTIME_CASES)]
            checks = next(row for row in rows if row["name"] == "sendq_cleanup")["evidence"]["checks"]
            if mutation == "missing":
                checks.pop()
            elif mutation == "failed":
                checks[0]["status"] = "failed"
            elif mutation == "running":
                checks[0]["server_was_stopped"] = False
            elif mutation == "alias":
                next(check for check in checks if check["name"] == "protected_root_alias_refused")["required_alias_and_queue_preserved"] = False
            else:
                next(check for check in checks if check["name"].startswith("shared_readonly_"))["real_partitioned_have_preserved"] = False
            self.write("runtime.log", {"scenarios": rows})
            self.assertTrue(self.check(), mutation)

    def test_source_and_baseline_mismatches_reject(self):
        self.assertTrue(gate.platform_errors(self.folder, "wrong", self.baseline))
        self.assertTrue(gate.platform_errors(self.folder, self.candidate, "wrong"))

    def test_unexercised_interpreter_build_and_wrong_live_journal_reject(self):
        self.write("python-installer-evidence.json", {"source_sha256": "old", "exit_code": 0, "build_os_user": "root"})
        self.assertTrue(any("Python build" in e for e in self.check()))
        self.write("p4d-process-argv.json", {"argv": ["p4d", "-J", "/p4/1/logs/journal"]})
        self.assertTrue(any("running p4d" in e for e in self.check()))

    def test_coverage_failure_is_not_hidden_by_live_classification(self):
        self.write("coverage.json", {"source_matches_results": True, "scripts": [
            {"script": "backup_functions.sh", "level": "live", "outcome": "failed"}]})
        self.assertTrue(any("failed/incomplete" in e for e in self.check()))

    def test_topology_requires_every_named_failure_and_recovery_case(self):
        rows = [{"name": name, "status": "passed"} for name in gate.TOPOLOGY_CASES]
        self.assertEqual(gate.scenario_errors({"scenarios": rows}, "topology", gate.TOPOLOGY_CASES), [])
        rows[0]["status"] = "blocked"
        self.assertTrue(gate.scenario_errors({"scenarios": rows}, "topology", gate.TOPOLOGY_CASES))
        self.assertTrue(gate.scenario_errors({"scenarios": rows[1:]}, "topology", gate.TOPOLOGY_CASES))

    def test_topology_rejects_changed_driver_or_unverified_node_source_and_image(self):
        drivers = {name: "hash-" + name for name in ("topology_integration.py", "topology_bootstrap.sh")}
        manifest = [{"path": "test/containers/" + name, "sha256": digest} for name, digest in drivers.items()]
        document = {"scenarios": [{"name": name, "status": "passed"} for name in gate.TOPOLOGY_CASES],
                    "source_manifest_sha256": self.candidate, "all_required_passed": True,
                    "journal": "/p4/1/journals/journal", "shared_storage": False,
                    "driver_sha256": drivers, "image_evidence": {"node_id": "sha256:nodes"},
                    "nodes": {name: {"service": daemon, "main_pid": 123,
                                     "running_image_id": "sha256:nodes", "source_manifest_sha256": self.candidate,
                                     "journal": "/p4/1/journals/journal", "command": "p4d -J /p4/1/journals/journal -r /p4/1/root"}
                              for name, daemon in gate.TOPOLOGY_NODES.items()}}
        self.assertEqual(gate.topology_errors(document, self.candidate, manifest), [])
        drivers["topology_integration.py"] = "development-driver"
        self.assertTrue(any("driver differs" in error for error in gate.topology_errors(document, self.candidate, manifest)))
        drivers["topology_integration.py"] = "hash-topology_integration.py"
        for field, value in (("running_image_id", "sha256:unexercised"), ("source_manifest_sha256", "old-source"), ("main_pid", 0), ("command", "p4d -J /p4/1/logs/journal")):
            original = document["nodes"]["edge"][field]
            document["nodes"]["edge"][field] = value
            self.assertTrue(gate.topology_errors(document, self.candidate, manifest), field)
            document["nodes"]["edge"][field] = original
        document["nodes"].pop("proxy")
        self.assertTrue(gate.topology_errors(document, self.candidate, manifest))

    def test_versions_require_all_fourteen_cells_and_their_data_checks(self):
        cases = [{"platform": platform, "release": release, "status": "passed",
                  "source_manifest_sha256": self.candidate,
                  "journal": "/p4/1/journals/journal", "image_id": "sha256:fixture",
                  "architecture": "arm64", "host_architecture": "arm64", "emulated": False,
                  "binary_builds": {binary: "vendor/" + release + "/build" for binary in ("p4", "p4d", "p4broker", "p4p")},
                  "checksums": {binary: "a" * 64 for binary in ("p4", "p4d", "p4broker", "p4p")}, "python": "3.12",
                  "scenarios": [passed_scenario(name) for name in gate.VERSION_CASES]}
                 for platform in gate.VERSION_PLATFORMS for release in gate.VERSION_RELEASES]
        doc = {"status": "passed", "source_manifest_sha256": self.candidate, "cases": cases}
        self.assertEqual(gate.version_errors(doc, self.candidate), [])
        cases[0]["emulated"] = True
        self.assertTrue(any("architecture evidence" in error for error in gate.version_errors(doc, self.candidate)))
        cases[0]["emulated"] = False
        self.assertTrue(gate.version_errors(doc, "different-source"))
        cases[0]["source_manifest_sha256"] = "different-source"
        self.assertTrue(gate.version_errors(doc, self.candidate))
        cases[0]["source_manifest_sha256"] = self.candidate
        original = cases[0]["binary_builds"]["p4d"]
        cases[0]["binary_builds"]["p4d"] = "unmatched-version"
        self.assertTrue(gate.version_errors(doc, self.candidate))
        cases[0]["binary_builds"]["p4d"] = original
        cases[0]["scenarios"].pop()
        self.assertTrue(gate.version_errors(doc, self.candidate))
        cases.pop(0)
        self.assertTrue(gate.version_errors(doc, self.candidate))


class SubmittedExport(unittest.TestCase):
    def test_baseline_paths_cannot_escape_selected_depot(self):
        for file in ("//other/source/file", "//depot/source/../outside", "//depot/source//absolute"):
            with self.subTest(file=file), self.assertRaises(ValueError):
                export_baseline.relative_path("//depot/source", file)

    def test_export_uses_submitted_revisions_not_workspace_bytes(self):
        with tempfile.TemporaryDirectory() as temp:
            target = Path(temp) / "baseline"
            rows = [{"depotFile": "//depot/sdp/script.sh", "rev": "5", "action": "edit", "type": "text+x"},
                    {"depotFile": "//depot/sdp/retired", "rev": "3", "action": "delete", "type": "text"}]
            from subprocess import CompletedProcess
            def print_revision(argv, **kwargs):
                self.assertEqual(argv, ["p4", "print", "-q", "//depot/sdp/script.sh#5"])
                return CompletedProcess(argv, 0, b"submitted content\n", b"")
            with patch.object(export_baseline, "records", return_value=rows), patch.object(export_baseline.subprocess, "run", side_effect=print_revision):
                manifest = export_baseline.export("//depot/sdp", 34003, target)
            self.assertEqual(len(manifest), 1)
            self.assertEqual((target / "script.sh").read_bytes(), b"submitted content\n")
            self.assertEqual(manifest[0]["sha256"], hashlib.sha256(b"submitted content\n").hexdigest())
            self.assertFalse((target / "retired").exists())
            self.assertTrue((target / "script.sh").stat().st_mode & 0o111)


class DeploymentArchive(unittest.TestCase):
    def archive(self, path, members):
        with tarfile.open(path, "w:gz") as stream:
            for name, data, kind in members:
                info = tarfile.TarInfo(name)
                info.mode, info.type, info.size = 0o755, kind, len(data)
                stream.addfile(info, io.BytesIO(data))

    def test_archive_bytes_and_unix_content_preserved_windows_excluded(self):
        with tempfile.TemporaryDirectory() as temp:
            archive, target = Path(temp) / "source.tgz", Path(temp) / "baseline"
            self.archive(archive, [("sdp/Server/Unix/script", b"deployed\n", tarfile.REGTYPE),
                                   ("sdp/Server/Windows/old.ps1", b"unused", tarfile.REGTYPE)])
            before = archive.read_bytes()
            rows = export_archive.export(archive, target, confirmed=True)
            self.assertEqual(archive.read_bytes(), before)
            self.assertEqual(len(rows), 1)
            self.assertEqual((target / "Server/Unix/script").read_bytes(), b"deployed\n")
            self.assertFalse((target / "Server/Windows").exists())
            info = json.loads((target / ".baseline-info.json").read_text())
            self.assertTrue(info["production_baseline_confirmed"])
            self.assertEqual(info["archive_sha256"], hashlib.sha256(before).hexdigest())
            self.assertEqual(baseline_comparison.verify_manifest(target), gate.manifest_hash(target / ".export-manifest.json"))

    def test_archive_rejects_unsafe_paths_links_duplicates_before_writing(self):
        for members in ([("sdp/../../escape", b"x", tarfile.REGTYPE)],
                        [("/sdp/absolute", b"x", tarfile.REGTYPE)],
                        [("sdp/link", b"", tarfile.SYMTYPE)],
                        [("sdp/x", b"x", tarfile.REGTYPE)] * 2):
            with self.subTest(members=members), tempfile.TemporaryDirectory() as temp:
                archive, target = Path(temp) / "source.tgz", Path(temp) / "baseline"
                self.archive(archive, members)
                with self.assertRaises(ValueError):
                    export_archive.export(archive, target)
                self.assertFalse(target.exists())

    def test_expected_failure_record_is_not_passing_baseline_evidence(self):
        class KnownFailure(unittest.TestCase):
            @unittest.expectedFailure
            def test_failure(self):
                self.fail("known bug")
        result = unittest.TextTestRunner(stream=io.StringIO(), resultclass=baseline_comparison.Results).run(
            unittest.defaultTestLoader.loadTestsFromTestCase(KnownFailure))
        self.assertTrue(result.expectedFailures)
        self.assertEqual(result.scenarios[0]["status"], "failed")


if __name__ == "__main__":
    unittest.main()