release_gate.py #3

#!/usr/bin/env python3
"""Fail closed on missing, stale or failed assurance evidence.

A passing gate covers the declared fixture matrix. It is not certification of
an unidentified production revision, host kernel or external integration.
"""
import argparse
import hashlib
import json
from pathlib import Path
import re

from baseline_comparison import BASELINE_CASES, verify_manifest

PRIMARY_STAGES = ("offline", "integration", "runtime", "maintenance", "baseline-comparison", "python-installer")
DISTRO_STAGES = tuple("distro-" + stage for stage in PRIMARY_STAGES if stage != "python-installer")
RUNTIME_CASES = {"checkpoint_restore", "journal_rotation", "database_switch", "ssh_host_trust", "sendq_cleanup"}
MAINTENANCE_CASES = {"maintenance_fixture", "maintenance_groups", "maintenance_user_reports", "maintenance_inventory",
                     "maintenance_labels", "maintenance_label_failure_recovery", "maintenance_client_unload",
                     "maintenance_label_unload", "maintenance_shelf_cleanup", "maintenance_protections", "maintenance_jobs"}
OFFLINE_CASES = BASELINE_CASES | {
    "test_logging_concurrency.MaintenanceLogConcurrency.test_" + name for name in (
        "overlapping_runs_keep_active_inode_and_all_diagnostics_at_one_and_default_retention",
        "simultaneous_preparation_opens_one_shared_inode",
        "failure_and_handled_or_abrupt_term_release_leases_and_preserve_evidence",
    )
}
OFFLINE_CASES |= {
    "test_recovery_completion.FailureLockRelease.test_die_releases_owned_marker_and_keeps_original_failure_status",
    "test_recovery_completion.FailureLockRelease.test_die_preserves_replaced_marker",
    "test_recovery_completion.FailureLockRelease.test_die_preserves_unowned_legacy_marker",
    "test_recovery_completion.FailureLockRelease.test_child_die_does_not_release_parent_marker",
    "test_recovery_completion.FailureLockRelease.test_child_completion_does_not_release_parent_marker",
    "test_recovery_completion.FailureLockRelease.test_failed_marker_removal_does_not_recurse_or_replace_original_error",
    "test_recovery_completion.OfflineRestoreCompletion.test_serial_restore_uses_completion_sidecar_without_verification_prepass",
    "test_recovery_completion.OfflineRestoreCompletion.test_parallel_restore_does_not_require_jvp_capability",
    "test_recovery_completion.OfflineRestoreCompletion.test_directory_or_fifo_serial_source_does_not_remove_database",
    "test_recovery_completion.OfflineRestoreCompletion.test_restore_failure_is_not_reported_as_usable_or_successful",
    "test_recovery_completion.LoaderCompletionPhases.test_serial_and_parallel_load_execute_only_the_normal_restore",
    "test_recovery_completion.LoaderCompletionPhases.test_missing_sidecar_plain_gzip_and_parallel_are_rejected_before_restore",
    "test_recovery_completion.LoaderCompletionPhases.test_directory_named_md5_is_not_a_completion_file",
    "test_recovery_completion.LoaderCompletionPhases.test_gzip_stem_sidecar_works_when_parent_directory_contains_dot_gz",
    "test_recovery_completion.LoaderCompletionPhases.test_actual_restore_failure_remains_a_failure",
    "test_journal_gaps.JournalGaps.test_missing_unarchived_number_does_not_stop_later_selected_journal",
    "test_journal_gaps.JournalGaps.test_gzip_command_error_is_not_hidden_by_existing_archive",
    "test_journal_gaps.JournalGaps.test_real_move_failure_preserves_compressed_source_and_stops",
    "test_journal_gaps.JournalGaps.test_set_vars_defaults_parallel_checkpoint_threads_to_sixteen",
    "test_sendq_cleanup.LocalClientDirectoryResolution.test_legacy_global_p4name_redirect_is_overridden_by_named_or_In",
    "test_sendq_cleanup.ReadonlyCopyResolution.test_copy_uses_legacy_scope_and_preserves_equals_in_readonly_path",
    "test_sendq_cleanup.SendqCleanup.test_protected_roots_and_ancestors_are_not_cleaned",
    "test_sendq_cleanup.SendqCleanup.test_config_failure_prevents_any_deletion",
    "test_sendq_cleanup.SendqCleanup.test_partial_deletion_failure_is_propagated",
    "test_sendq_cleanup.SendqCleanup.test_shared_or_nested_readonly_directory_is_warned_and_left_untouched",
    "test_sendq_cleanup.SendqCleanup.test_outside_readonly_alias_chained_through_queue_preserves_all_shared_entries",
    "test_sendq_cleanup.SendqCleanup.test_protected_configured_paths_depending_on_queue_alias_are_rejected",
    "test_sendq_cleanup.SendqCleanup.test_sendq_path_depending_on_its_own_nested_alias_is_rejected",
    "test_sendq_cleanup.SendqCleanup.test_configured_journal_alias_chain_is_rejected_but_external_file_is_safe",
    "test_sendq_switch.SendqSwitch.test_first_cleanup_failure_preserves_database_links_and_stays_down",
    "test_sendq_switch.SendqSwitch.test_second_cleanup_failure_rolls_back_and_restarts_original",
    "test_sendq_switch.SendqSwitch.test_callers_without_opt_in_do_not_clean_either_root",
}
HARNESS_CASES = {
    "create and verify exact submitted fixture",
    "p4d init: start brings server up", "p4d init: stop brings server down",
    "p4d init: restart brings server up", "verify_shelves checks the real fixture shelf",
    "verification detects missing archives and accepts restored exact file/shelf bytes",
}
ALLOWED_HARNESS_SKIPS = {"kill_idle.sh", "replica/edge/standby/broker tooling"}
ALLOWED_STATIC_ONLY = {"Server/Unix/p4/common/etc/init.d/p4dtg_instance_init.template"}
TOPOLOGY_CASES = {"six_node_install", "exact_fixture", "replication_and_local_shelves", "broker_proxy_routes",
                  "checkpoint_database_switch", "retained_generation_replay", "missing_archive_repair",
                  "restart_failure_preserves_data", "interrupted_recovery_preserves_data", "standby_promotion"}
TOPOLOGY_NODES = {"commit": "p4d", "edge": "p4d", "standby": "p4d", "edge-standby": "p4d",
                  "broker": "p4broker", "proxy": "p4p"}
VERSION_RELEASES = {"2023.1", "2023.2", "2024.1", "2024.2", "2025.1", "2025.2", "2026.1"}
VERSION_PLATFORMS = {"rhel8", "rhel9"}
VERSION_CASES = {"clean_install", "exact_fixture", "restart_preserves_data", "checkpoint_restore",
                 "journal_rotation_restore", "maintenance_queries", "service_user_authentication", "sendq_cleanup"}

SENDQ_CHECKS = {"unset_noop", "any_relative", "any_absolute", "local_relative_overrides_any",
                "local_absolute_overrides_any", "other_server_ignored", "named_empty_disables_any",
                "p4name_selects_scope", "startup_In_selects_scope", "running_server_refused",
                "configured_root_symlink", "missing_directory_noop", "configured_file_refused",
                "database_p4name_one_hop", "startup_In_bypasses_database_redirect",
                "global_database_p4name_redirect", "named_p4name_overrides_global_redirect",
                "shared_readonly_equal_preserved", "shared_readonly_child_preserved", "shared_readonly_parent_preserved",
                "shared_readonly_nested_alias_preserved", "shared_readonly_alias_chain_preserved",
                "protected_root_alias_refused", "configured_sendq_alias_refused"}


def sendq_evidence_errors(row, label):
    evidence = row.get("evidence", {})
    checks = evidence.get("checks", [])
    names = [check.get("name") for check in checks]
    errors = []
    if len(set(names)) != len(names) or not SENDQ_CHECKS.issubset(names):
        errors.append(label + ": sendq cleanup lacks required scope/path/safety assertions")
    for assertion in ("regular_files_and_symlinks_removed", "directories_and_fifo_preserved",
                      "nested_symlink_targets_preserved", "fixture_records_and_bytes_preserved"):
        if evidence.get(assertion) is not True:
            errors.append(label + ": missing sendq preservation assertion " + assertion)
    for check in checks:
        if check.get("server_was_stopped") is not (check.get("name") != "running_server_refused"):
            errors.append(label + ": missing sendq service-state assertion " + str(check.get("name")))
        if check.get("status") != "passed":
            errors.append(label + ": sendq cleanup subcase did not pass: " + str(check.get("name")))
        if check.get("name") in {"protected_root_alias_refused", "configured_sendq_alias_refused"} and check.get("required_alias_and_queue_preserved") is not True:
            errors.append(label + ": missing required alias refusal preservation assertion")
        if check.get("name", "").startswith("shared_readonly_") and check.get("real_partitioned_have_preserved") is not True:
            errors.append(label + ": missing real partitioned-client preservation assertion")
    return errors


def load_json(path):
    return json.loads(path.read_text())


def manifest_hash(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def scenario_errors(document, label, required=()):
    errors = []
    rows = document.get("scenarios", [])
    if not rows:
        return [label + ": no behavioral scenarios"]
    names = [row.get("name") for row in rows]
    if len(set(names)) != len(names):
        errors.append(label + ": duplicate scenario names")
    for name in sorted(set(required) - set(names)):
        errors.append(label + ": missing required scenario " + name)
    for row in rows:
        if row.get("name") == "sendq_cleanup":
            errors.extend(sendq_evidence_errors(row, label))
        if row.get("name") == "database_switch":
            roots = row.get("evidence", {}).get("sendq_physical_roots_cleaned", [])
            if len(roots) != 2 or len(set(roots)) != 2 or not all(isinstance(root, str) and root.startswith("/") for root in roots):
                errors.append(label + ": missing two distinct physical send-queue roots")
            for assertion in ("both_physical_roots_verified", "real_partitioned_have_preserved", "readonly_path_contains_space_and_equals"):
                if row.get("evidence", {}).get(assertion) is not True:
                    errors.append(label + ": database switch lacks required assertion " + assertion)
        if row.get("status") != "passed":
            errors.append(label + ": scenario did not pass: " + str(row.get("name")))
    if document.get("status", "passed") != "passed":
        errors.append(label + ": report status is not passed")
    return errors


def offline_errors(text, label):
    errors = []
    count = re.search(r"^Ran (\d+) tests? in ", text, re.M)
    if not count or int(count[1]) == 0 or not re.search(r"^OK\s*$", text, re.M):
        errors.append(label + ": missing successful, nonempty, zero-skip unittest completion")
    # Failure-path regressions intentionally emit application diagnostics such
    # as ERROR:root:permission denied. Only unittest's failure headers/summary
    # establish a failed test; requiring each named "ok" remains independent.
    if re.search(r"^FAILED(?:\s|$)|^(?:ERROR|FAIL):\s+\S.*\([^)]+\)", text, re.M) or re.search(r"\.\.\. skipped", text):
        errors.append(label + ": failed or skipped test")
    passed, pending = set(), None
    for line in text.splitlines():
        match = re.match(r"^(test\w+) \(([\w.]+)\)(?: \(.+\))? \.\.\.\s*(.*)$", line.strip())
        if match:
            method, suite, state = match.groups()
            pending = suite if suite.endswith("." + method) else suite + "." + method
        else:
            state = line.strip()
        if pending and state == "ok":
            passed.add(pending)
            pending = None
        elif pending and re.fullmatch(r"FAIL|ERROR|skipped.*|expected failure|unexpected success", state):
            errors.append(label + ": unsuccessful required evidence: " + pending)
            pending = None
    for name in sorted(OFFLINE_CASES - passed):
        errors.append(label + ": required regression did not pass: " + name)
    return errors


def harness_errors(text, label):
    text = re.sub(r"\x1b\[[0-9;]*m", "", text)
    errors = []
    summary = re.search(r"total=(\d+)\s+pass=(\d+)\s+fail=(\d+)\s+skip=(\d+)", text)
    if not summary or int(summary[2]) == 0 or int(summary[3]) != 0:
        errors.append(label + ": missing successful nonempty harness completion")
    passed = set(re.findall(r"\[PASS\] ([^\n]+)", text))
    for name in sorted(HARNESS_CASES - passed):
        errors.append(label + ": missing state/content assertion: " + name)
    if "[FAIL]" in text:
        errors.append(label + ": failed assertion despite process status")
    for skip in re.findall(r"\[SKIP\] ([^\n]+)", text):
        if skip.split(" (", 1)[0] not in ALLOWED_HARNESS_SKIPS:
            errors.append(label + ": unexpected skip " + skip)
    return errors


def version_errors(document, candidate_hash):
    errors, seen = [], set()
    if document.get("source_manifest_sha256") != candidate_hash:
        errors.append("version compatibility: candidate manifest mismatch")
    if document.get("status") != "passed":
        errors.append("version compatibility: report did not pass")
    for row in document.get("cases", []):
        key = (row.get("platform"), row.get("release"))
        label = "version compatibility " + str(key)
        if key in seen:
            errors.append(label + ": duplicate matrix cell")
        seen.add(key)
        errors.extend(scenario_errors(row, label, VERSION_CASES))
        if row.get("source_manifest_sha256") != candidate_hash:
            errors.append(label + ": cell used different candidate source")
        if row.get("journal") != "/p4/1/journals/journal":
            errors.append(label + ": wrong journal path")
        for field in ("image_id", "architecture", "binary_builds", "checksums", "python"):
            if not row.get(field):
                errors.append(label + ": missing " + field + " provenance")
        if not row.get("host_architecture") or type(row.get("emulated")) is not bool or row.get("emulated") != (row.get("architecture") != row.get("host_architecture")):
            errors.append(label + ": missing or inconsistent native/emulated architecture evidence")
        builds, checksums = row.get("binary_builds", {}), row.get("checksums", {})
        if set(builds) != {"p4", "p4d", "p4broker", "p4p"} or any("/" + str(row.get("release")) + "/" not in build for build in builds.values()):
            errors.append(label + ": binary release does not match the required cell")
        if len(checksums) != 4 or any(not re.fullmatch(r"[0-9a-fA-F]{64}", digest) for digest in checksums.values()):
            errors.append(label + ": incomplete vendor checksum evidence")
    for key in sorted({(platform, release) for platform in VERSION_PLATFORMS for release in VERSION_RELEASES} - seen):
        errors.append("version compatibility: missing required matrix cell " + str(key))
    return errors


def topology_errors(document, candidate_hash, manifest_rows):
    errors = scenario_errors(document, "topology", TOPOLOGY_CASES)
    if document.get("source_manifest_sha256") != candidate_hash:
        errors.append("topology: candidate manifest mismatch")
    if document.get("all_required_passed") is not True:
        errors.append("topology: full required run was not completed")
    if document.get("journal") != "/p4/1/journals/journal" or document.get("shared_storage") is not False:
        errors.append("topology: journal/storage layout differs from declared production fixture")
    source_hashes = {row["path"]: row["sha256"] for row in manifest_rows}
    for name in ("topology_integration.py", "topology_bootstrap.sh"):
        expected = source_hashes.get("test/containers/" + name)
        if not expected or document.get("driver_sha256", {}).get(name) != expected:
            errors.append("topology: driver differs from candidate export: " + name)
    node_image = document.get("image_evidence", {}).get("node_id")
    if not node_image:
        errors.append("topology: missing node image identity")
    nodes = document.get("nodes", {})
    if set(nodes) != set(TOPOLOGY_NODES):
        errors.append("topology: missing or unexpected node identities")
    for name, daemon in TOPOLOGY_NODES.items():
        row = nodes.get(name, {})
        if row.get("running_image_id") != node_image or not node_image or row.get("source_manifest_sha256") != candidate_hash:
            errors.append("topology " + name + ": installed source or running image differs")
        if row.get("service") != daemon or type(row.get("main_pid")) is not int or row.get("main_pid", 0) <= 0:
            errors.append("topology " + name + ": missing actual service process evidence")
        if daemon == "p4d" and (row.get("journal") != "/p4/1/journals/journal" or not re.search(r"(?:^|\s)-J /p4/1/journals/journal(?:\s|$)", row.get("command", ""))):
            errors.append("topology " + name + ": incorrect live journal path")
    return errors


def platform_errors(folder, candidate_hash, baseline_hash):
    errors = []
    try:
        outcomes = load_json(folder / "test-outcomes.json")
        for stage in PRIMARY_STAGES + DISTRO_STAGES:
            if outcomes.get(stage) != 0 or isinstance(outcomes.get(stage), bool):
                errors.append("missing/failed required stage: " + stage)
        for stage, status in outcomes.items():
            if status != 0:
                errors.append("recorded failure: " + stage)
        if manifest_hash(folder / "source-manifest.json") != candidate_hash:
            errors.append("candidate source manifest differs from requested export")
        if manifest_hash(folder / "baseline-source-manifest.json") != baseline_hash:
            errors.append("baseline manifest differs from requested comparison baseline")
        container = load_json(folder / "container-inspect.json")
        container = container[0] if isinstance(container, list) else container
        image = load_json(folder / "image-inspect.json")
        image = image[0] if isinstance(image, list) else image
        if not container.get("Image") or container["Image"] != image.get("Id"):
            errors.append("running container does not use recorded final image")
        if not container.get("State", {}).get("Running"):
            errors.append("test container is not running")
        if (folder / "journal-path.txt").read_text().strip() != "/p4/1/journals/journal":
            errors.append("production journal-path fixture was not used")
        argv = load_json(folder / "p4d-process-argv.json").get("argv", [])
        if "-J" not in argv or argv.index("-J") + 1 >= len(argv) or argv[argv.index("-J") + 1] != "/p4/1/journals/journal":
            errors.append("running p4d was not verified with the requested journal path")
        runtimes = load_json(folder / "python-runtimes.json")
        receipt = load_json(folder / "python-installer-evidence.json")
        manifest_rows = load_json(folder / "source-manifest.json")
        installer_hash = next((row["sha256"] for row in manifest_rows
                               if row["path"] == "Server/Unix/setup/install_sdp_python.sh"), None)
        if not installer_hash or receipt.get("source_sha256") != installer_hash or receipt.get("exit_code") != 0 or receipt.get("build_os_user") != "perforce":
            errors.append("Python build is missing or does not match the tested installer/account")
        if runtimes.get("distro_supported") is not True or not runtimes.get("baseline", {}).get("version", "").startswith("3.14.8"):
            errors.append("declared primary/distro Python runtimes were not verified")
        command = (folder / "installer-command.txt").read_text()
        if "mkdirs.sh" not in command or "mkdirs.cfg" not in command:
            errors.append("missing real mkdirs.sh/mkdirs.cfg installation evidence")
        service = (folder / "service-final-state.txt").read_text()
        if "ActiveState=active" not in service or not re.search(r"^MainPID=[1-9][0-9]*$", service, re.M):
            errors.append("final p4d service is not healthy")
        coverage = load_json(folder / "coverage.json")
        if not coverage.get("source_matches_results") or coverage.get("failed_stages"):
            errors.append("coverage evidence has failed stages or mismatched source")
        if not coverage.get("scripts"):
            errors.append("coverage contains no operational script evidence")
        for row in coverage.get("scripts", []):
            if row.get("outcome") in ("failed", "partial"):
                errors.append("failed/incomplete scenario evidence: " + row["script"])
            if row.get("level") in ("untested", "static_only") and row["script"] not in ALLOWED_STATIC_ONLY:
                errors.append("unapproved absence of behavioral evidence: " + row["script"])
        for prefix in ("", "distro-"):
            errors.extend(offline_errors((folder / (prefix + "offline.log")).read_text(), prefix + "offline"))
            errors.extend(harness_errors((folder / (prefix + "integration.log")).read_text(), prefix + "integration"))
            for kind in ("runtime", "maintenance"):
                errors.extend(scenario_errors(load_json(folder / (prefix + kind + ".log")), prefix + kind,
                                              RUNTIME_CASES if kind == "runtime" else MAINTENANCE_CASES))
            comparison = load_json(folder / (prefix + "baseline-comparison.json"))
            errors.extend(scenario_errors(comparison, prefix + "baseline-comparison", BASELINE_CASES))
            if comparison.get("candidate_manifest_sha256") != candidate_hash or comparison.get("baseline_manifest_sha256") != baseline_hash:
                errors.append(prefix + "baseline-comparison: source hashes mismatch")
    except (OSError, ValueError, KeyError, TypeError) as error:
        errors.append("missing/invalid required evidence: " + str(error))
    return errors


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source", type=Path, required=True)
    parser.add_argument("--baseline", type=Path, required=True)
    parser.add_argument("--results", type=Path, required=True)
    parser.add_argument("--topology-results", type=Path, action="append", required=True,
                        help="Full topology result; repeat for additional OS images. Every supplied run must pass.")
    parser.add_argument("--version-results", type=Path, required=True)
    parser.add_argument("--platform", action="append", choices=("ubuntu", "rhel", "rhel8", "rhel9"))
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    report = {"schema_version": 1, "status": "failed", "errors": [], "platforms": {},
              "scope": "Declared container fixture acceptance, not certification of all production behavior",
              "production_baseline_confirmed": False}
    try:
        candidate_hash, baseline_hash = verify_manifest(args.source), verify_manifest(args.baseline)
        report.update(candidate_manifest_sha256=candidate_hash, baseline_manifest_sha256=baseline_hash)
        baseline_info = load_json(args.baseline / ".baseline-info.json")
        report.update(baseline=baseline_info, production_baseline_confirmed=baseline_info.get("production_baseline_confirmed") is True)
        for platform in args.platform or ("ubuntu", "rhel", "rhel8", "rhel9"):
            errors = platform_errors(args.results / platform, candidate_hash, baseline_hash)
            report["platforms"][platform] = {"status": "failed" if errors else "passed", "errors": errors}
            report["errors"].extend(platform + ": " + error for error in errors)
        report["topologies"] = []
        manifest_rows = load_json(args.source / ".export-manifest.json")
        for result_path in args.topology_results:
            topology = load_json(result_path)
            errors = topology_errors(topology, candidate_hash, manifest_rows)
            report["topologies"].append({"result": str(result_path), "image": topology.get("image"),
                                        "status": "failed" if errors else "passed", "errors": errors})
            report["errors"].extend(str(result_path) + ": " + error for error in errors)
        versions = load_json(args.version_results)
        report["errors"].extend(version_errors(versions, candidate_hash))
        report["status"] = "failed" if report["errors"] else "passed"
    except (OSError, ValueError, KeyError, TypeError) as error:
        report["errors"].append(str(error))
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(report, indent=2) + "\n")
    print(json.dumps(report, indent=2))
    return int(report["status"] != "passed")


if __name__ == "__main__":
    raise SystemExit(main())