coverage.py #3

#!/usr/bin/env python3
"""Report per-script behavioral evidence against an authoritative source export.

This is scenario coverage, not line coverage. A syntax check never establishes
that a script works. Missing service/topology tests remain explicit gaps.
"""
import argparse
import hashlib
import json
from pathlib import Path
import re
import subprocess


# Reviewed associations. A suite association describes focused cases, not every
# function/path in the script. Evidence is credited only when the corresponding
# tests actually passed for the exact exported source tree.
ISOLATED = {}


def associate(names, *suites):
    for name in names.split():
        ISOLATED.setdefault(name, []).extend(suites)


associate("backup_functions.sh", "test_runtime_review.BackupRuntimeReview", "test_runtime_improvements.RuntimeImprovements", "test_systemd_service_detection.ServiceDetection")
associate("backup_functions.sh", "test_sendq_cleanup.LocalClientDirectoryResolution",
          "test_sendq_cleanup.ReadonlyCopyResolution", "test_sendq_cleanup.SendqCleanup",
          "test_sendq_switch.SendqSwitch", "test_journal_gaps.JournalGaps")
associate("recreate_db_checkpoint.sh", "test_sendq_switch.SendqSwitch")
associate("sdp_checkpoint_lock.sh", "test_runtime_improvements.RuntimeImprovements")
associate("backup_functions.sh sdp_checkpoint_lock.sh", "test_recovery_completion.FailureLockRelease")
associate("backup_functions.sh", "test_recovery_completion.OfflineRestoreCompletion")
associate("load_checkpoint.sh", "test_recovery_completion.LoaderCompletionPhases")
associate("recover_edge.sh", "test_runtime_improvements.EdgeRecovery", "test_runtime_review.RecoveryReview")
associate("convert_standby_to_master.sh", "test_runtime_review.RecoveryReview")
associate("load_checkpoint.sh", "test_runtime_review.RecoveryReview")
associate("rsync_missing.sh", "test_runtime_improvements.MissingTransfers", "test_maintenance_logging.MaintenanceLogging")
associate("cacheclean.sh", "test_runtime_review.CacheCleanupReview", "test_core_improvements.BoundedCacheCleanup")
associate("sdp_cache_cleanup.sh proxyclean.sh", "test_core_improvements.BoundedCacheCleanup", "test_maintenance_logging.MaintenanceLogging")
associate("sdp_cache_cleanup.sh", "test_cache_cancellation.CacheCancellation")
associate("copy_missing_shelf_files.sh kill_idle.sh p4login_master p4sanity_check.sh update_limits.py", "test_maintenance_logging.MaintenanceLogging")
associate("mkedge.sh mkedge-standby.sh mkstandby.sh mkstandby_shared.sh", "test_maintenance_logging.MaintenanceLogging", "test_runtime_review.PasswordFileReview")
associate("sdp_service_user.sh configure_new_server.sh mkedge.sh mkedge-standby.sh mkstandby.sh mkstandby_shared.sh", "test_service_password.ServicePassword")
associate("sync_replica.sh recreate_db_sync_replica.sh", "test_maintenance_logging.MaintenanceLogging")
associate("daily_checkpoint.sh live_checkpoint.sh rotate_journal.sh recreate_db_checkpoint.sh recreate_offline_db.sh journal_watch.sh convert_old_master_to_standby.sh", "test_runtime_improvements.RuntimeImprovements")
associate("sdp_structured_log.sh sdp_structured_log.py", "test_core_review.CoreReview", "test_core_improvements.LogRetention")
associate("p4verify.py", "test_core_improvements.LogRetention")
associate("p4verify.py", "test_verification_depots.VerificationDepots")
associate("ccheck.sh", "test_core_review.CoreReview")
associate("ccheck.sh configure_new_server.sh", "test_ccheck_compatibility.Compatibility")
associate("get_p4_binaries.sh", "test_core_review.CoreReview")
associate("recommend_limits.py", "test_recommend_limits.MetricAvailabilityTests")
associate("Server/Unix/p4/common/cloud/aws/bin/snapshot.sh", "test_core_review.CoreReview")
associate("sdp_snapshot.py", "test_cloud_improvements.AWSTests", "test_cloud_improvements.AzureTests", "test_cloud_improvements.Commands")
associate("install_sdp.sh os_tweaks.sh mkdirs.sh", "test_setup_triggers_review.ShellReviewTests")
associate("install_sdp_python.sh reset.sh disposable_test_guard.sh sdp_test_harness.sh", "test_setup_improvements.SetupImprovements")
associate("pull.sh submit.sh", "test_setup_triggers_review.ShellReviewTests")
associate("CheckCaseTrigger.py SetLabelOptions.py keep_group_unset.py", "test_setup_triggers_review.PythonTriggerTests")
associate("rad_authcheck.py", "test_setup_triggers_review.PythonTriggerTests", "test_setup_improvements.TriggerImprovements")
associate("command_block.py", "test_setup_improvements.TriggerImprovements")
associate("sdputils.py", "test_maintenance_review.Helpers", "test_maintenance_improvements.Selection")
associate("sdputils.py", "test_logging_concurrency.MaintenanceLogConcurrency")
associate("unload_clients.py", "test_baseline_maintenance.BaselineMaintenance")
associate("accessdates.py unloadaccessdates.py unload_clients.py unload_clients_with_delete.py delete_unload_clients.py email_pending_client_deletes.py p4deleteuser.py totalusers.py", "test_maintenance_review.APIFailures")
associate("addusertogroup.py convert_label_to_autoreload.py isitalabel.py delusers.py remove_empty_pending_changes.py", "test_maintenance_review.Entrypoints")
associate("rollback.py lowertree.py", "test_maintenance_review.RollbackAndFilesystem")
associate("maintenance edge_maintenance email.sh create_p4_filelist.sh", "test_maintenance_review.Wrappers")
associate("maintenance edge_maintenance maintenance_workspace.sh", "test_maintenance_improvements.Workspaces")
associate("accessdates.py unload_clients.py", "test_maintenance_improvements.Selection")
associate("email_pending_client_deletes.py", "test_maintenance_improvements.Notifications")
associate("file_inventory.py", "test_maintenance_improvements.Inventory")
associate("clean_protect.py", "test_maintenance_improvements.Protections")
associate("createusers.py creategroups.py removeuserfromgroups.py maintain_user_from_groups.py convert_label_to_autoreload.py", "test_maintenance_improvements.ScratchFiles")
associate("countrevs.py", "test_remaining_utilities.LocalFilters.test_revision_report_counts_file_heads_and_usage_is_nonzero")
associate("proxysearch.py", "test_remaining_utilities.LocalFilters.test_proxy_search_deduplicates_and_reports_missing_input")
associate("lowercp.py", "test_remaining_utilities.LocalFilters.test_lowercase_copy_preserves_existing_destination_on_collision")
associate("protect_groups.py", "test_remaining_utilities.LocalFilters.test_protection_group_report_matches_case_and_preserves_query_failure")
associate("clean_log.sh", "test_remaining_utilities.LocalFilters.test_clean_log_removes_virtual_lock_blocks_and_keeps_database_metrics")
associate("check_pull_status.sh", "test_remaining_utilities.LocalFilters.test_progress_report_counts_completed_batches_and_handles_empty_input")
associate("edge_vars", "test_remaining_utilities.LocalFilters.test_modern_edge_table_lists_include_storage_tables")
associate("cancel_pull_transfers.py", "test_remaining_utilities.TransferCancellation")
associate("group_audit.py", "test_remaining_utilities.GroupAudit")
associate("p4broker_instance_init.template p4p_instance_init.template", "test_remaining_utilities.TemplateEntrypoints.test_broker_proxy_templates_preserve_instance_arguments_and_child_status")
associate("Server/Unix/p4/1/bin/p4d_1", "test_remaining_utilities.TemplateEntrypoints.test_sample_p4d_wrapper_adds_case_flag_and_preserves_child_status")
associate("p4cachepurge.sh", "test_remaining_utilities.RuntimeCommands.test_cachepurge_passes_scope_and_reports_failed_command")
associate("run_if_master.sh run_if_edge.sh run_if_replica.sh", "test_remaining_utilities.RoleWrappers.test_server_role_routing_preserves_arguments_and_child_exit")
associate("run_if_broker.sh run_if_proxy.sh", "test_remaining_utilities.RoleWrappers.test_broker_proxy_route_only_when_init_script_is_executable")
associate("ps_functions.sh", "test_remaining_utilities.RoleWrappers.test_process_helper_excludes_init_and_search_process")
associate("find_blocker.py log_analyzer2.py", "test_remaining_utilities.LogAnalysis.test_compression_magic_and_suffix_inputs_are_read_without_modification")
associate("log_analyzer2.py", "test_remaining_utilities.LogAnalysis.test_parser_extracts_command_and_waits_and_filters_users")
associate("check_dir_ownership.sh", "test_remaining_utilities.OwnershipDiagnostics")
associate("broker_rotate.sh proxy_rotate.sh", "test_remaining_utilities.RuntimeCommands.test_broker_proxy_login_failure_stops_cleanup_and_keeps_diagnostic")
associate("replica_cleanup.sh", "test_remaining_utilities.RuntimeCommands.test_replica_cleanup_reads_offline_floor_before_retention_and_releases_lock",
          "test_runtime_improvements.RuntimeImprovements.test_all_existing_checkpoint_callers_skip_before_shared_logs")
associate("edge_dump.sh", "test_remaining_utilities.RuntimeCommands.test_edge_dump_keeps_checkpoint_filename_literal_and_releases_lock",
          "test_runtime_improvements.RuntimeImprovements.test_all_existing_checkpoint_callers_skip_before_shared_logs")
associate("request_replica_checkpoint.sh", "test_remaining_utilities.RuntimeCommands.test_replica_request_reports_failed_master_rotation")
associate("replica_status.sh", "test_remaining_utilities.RuntimeCommands.test_replica_status_archive_query_failures_are_nonzero",
          "test_remaining_utilities.RuntimeCommands.test_replica_status_rejects_missing_counters_and_keeps_failed_query_diagnostic")
associate("upgrade.sh", "test_remaining_utilities.UpgradeFailures")
associate("sizes.sh", "test_remaining_utilities.RuntimeCommands.test_sizes_uses_tagged_ids_and_preserves_query_failure")
associate("p4broker_base p4p_base", "test_service_lifecycle.ServiceLifecycle")
associate("p4cachepurge.sh", "test_service_lifecycle.ServiceLifecycle")
associate("submit_form_1_in.py", "test_setup_improvements.TriggerImprovements.test_incident_trigger_instance_does_not_consume_existing_user_arg")
associate("del_shelve.py", "test_maintenance_review.Entrypoints.test_shelf_cleanup_uses_current_query_and_preserves_unrelated_list")
associate("pymail.py", "test_additional_maintenance.AdditionalMaintenance.test_pymail_address_list_and_secure_transport",
          "test_additional_maintenance.AdditionalMaintenance.test_pymail_transport_failure_propagates")
associate("email_pending_user_deletes.py", "test_additional_maintenance.AdditionalMaintenance.test_inactive_user_notifications_filter_protected_recent_and_invalid_accounts",
          "test_additional_maintenance.AdditionalMaintenance.test_inactive_user_query_failure_prevents_notifications")
associate("setpass.py", "test_additional_maintenance.AdditionalMaintenance.test_password_reset_uses_stdin_and_propagates_rejection")
associate("mirroraccess.py", "test_additional_maintenance.AdditionalMaintenance.test_mirror_groups_preserves_members_and_adds_requested_users",
          "test_additional_maintenance.AdditionalMaintenance.test_mirror_query_failure_prevents_group_writes")
associate("update-changes.py", "test_additional_maintenance.AdditionalMaintenance.test_update_changes_preserves_specs_and_stops_after_failed_read")
associate("gen_default_broker_cfg.sh", "test_operational_edges.OperationalEdges.test_broker_generation_substitutes_endpoint_paths_and_autotune",
          "test_operational_edges.OperationalEdges.test_broker_failure_never_becomes_successful_empty_configuration",
          "test_operational_edges.OperationalEdges.test_server_query_failure_cleans_temporary_database_and_stops_generation")
associate("Server/Unix/p4/common/cloud/azure/bin/snapshot.sh", "test_operational_edges.OperationalEdges.test_azure_wrapper_selects_provider_and_preserves_instance_argument")
associate("ec2id", "test_operational_edges.OperationalEdges.test_ec2id_reads_current_cloud_init_link_and_reports_missing_link")
for name, methods in {
    "JobsCmdFilter.py": ["jobs_filter_rejects_expensive_patterns_and_allows_specific_search"],
    "PreventWsNonAscii.py": ["workspace_ascii_and_name_checks"],
    "SetWsOptions.py": ["workspace_defaults_change_only_new_forms"],
    "SetWsOptionsAndView.py": ["workspace_template_and_stream_views_are_preserved"],
    "TFSJobCheck.py": ["tfs_job_must_be_real_job_not_description_text"],
    "submit_form_1.py": ["description_template_applies_only_to_group_members"],
    "command_block.py": ["command_block_honors_exempt_user_and_read_only_option"],
    "block_checkpoint.sh": ["checkpoint_trigger_blocks_checkpoint_and_allows_journal"],
    "P4Triggers.py": ["p4change_preserves_files_revisions_and_jobs", "p4trigger_unexpected_exception_returns_failure_and_logs"],
    "radtest.py": ["radius_diagnostic_reports_denial_as_failure", "radius_diagnostic_tries_next_server_after_failure"],
}.items():
    associate(name, *("test_trigger_utilities.TriggerUtilityTests.test_" + method for method in methods))

# Explicit labels from the guarded single-commit-server harness. A role wrapper
# skip on a commit is a tested branch; it is not a live replica/edge topology.
HARNESS = {
    "p4_vars": ["p4_vars sets P4PORT", "p4_vars sets P4ROOT"],
    "ccheck.sh": ["ccheck.sh PASS (required configs)"],
    "p4login": ["p4login"], "p4login_master": ["p4login_master"],
    "run_if_master.sh": ["run_if_master runs on master"],
    "run_if_replica.sh": ["run_if_replica skips on master"],
    "run_if_edge.sh": ["run_if_edge skips on master"],
    "p4d_base": ["p4d init: start brings server up", "p4d init: stop brings server down", "p4d init: restart brings server up"],
    "p4sanity_check.sh": ["p4sanity_check.sh"],
    "live_checkpoint.sh": ["live_checkpoint.sh"],
    "daily_checkpoint.sh": ["daily_checkpoint.sh"],
    "rotate_journal.sh": ["rotate_journal.sh"],
    "recreate_offline_db.sh": ["recreate_offline_db.sh"],
    "p4verify.py": ["p4verify.py"], "verify_shelves.sh": ["verify_shelves.sh"],
    "p4master_run": ["p4_vars / p4master_run wrapper"],
    "sizes.sh": ["sizes.sh"], "clean_log.sh": ["clean_log.sh"],
    "journal_watch.sh": ["journal_watch.sh"], "update_limits.py": ["update_limits.py"],
    "p4dstate.sh": ["p4dstate.sh"],
}


def script_kind(path, data):
    if path.parts[0] == "test" and (path.name.startswith("test_") or "containers" in path.parts):
        return None  # Test implementation is evidence, not an operational script.
    if path.suffix == ".py":
        return "python"
    if path.suffix == ".sh" or path.name in {"p4_vars", "edge_vars"}:
        return "shell"
    if path.suffix in {".ps1", ".pl"}:
        return {".ps1": "powershell", ".pl": "perl"}[path.suffix]
    first_line = data.split(b"\n", 1)[0]
    if first_line.startswith(b"#!"):
        if b"python" in first_line:
            return "python"
        if b"sh" in first_line:
            return "shell"
        return "other"
    return None


def static_check(path, kind):
    try:
        if kind == "python":
            compile(path.read_bytes(), str(path), "exec")
            return {"status": "passed", "scope": "Python parser only; no imports or execution"}
        if kind == "shell":
            result = subprocess.run(["bash", "-n", str(path)], capture_output=True, text=True, timeout=30)
            return {"status": "passed" if result.returncode == 0 else "failed", "scope": "Bash syntax only", "diagnostic": result.stderr[-2000:]}
        return {"status": "not_run", "scope": "No non-executing validator configured"}
    except (OSError, SyntaxError, UnicodeError, subprocess.SubprocessError) as error:
        return {"status": "failed", "scope": "Syntax/parser check", "diagnostic": str(error)}


def unittest_results(text):
    rows = []
    pending = None
    for line in text.splitlines():
        match = re.match(r"^(test\w+) \(([\w.]+)\)(?: \(.+\))? \.\.\.\s*(.*)$", line.strip())
        if match:
            method, suite, state = match.groups()
            if suite.endswith("." + method):
                suite = suite[:-(len(method) + 1)]
            pending = {"suite": suite, "test": suite + "." + method}
        else:
            state = line.strip()
        # Test diagnostics can appear after the verbose test label and before
        # unittest emits its result on a later line. Never infer success from
        # the mere presence of a test name.
        if pending and re.fullmatch(r"ok|FAIL|ERROR|skipped.*", state):
            rows.append(dict(pending, status={"ok": "passed", "FAIL": "failed", "ERROR": "failed"}.get(state, "skipped")))
            pending = None
    return rows


def read_optional(path):
    return path.read_text(errors="replace") if path.is_file() else ""


def aggregate_outcome(static, evidence):
    """Keep failure independent of the highest successfully exercised level."""
    if static["status"] == "failed" or any(row["status"] == "failed" for row in evidence):
        return "failed"
    if any(row["status"] == "passed" for row in evidence):
        return "partial" if any(row["status"] != "passed" for row in evidence) else "passed-for-listed-scenarios"
    return "no-behavioral-evidence"


def report(source, results):
    manifest_path = source / ".export-manifest.json"
    if not manifest_path.is_file():
        raise ValueError("An authoritative export manifest is required; local stale/deleted files must not be inventoried")
    manifest = json.loads(manifest_path.read_text())
    result_manifest = json.loads(read_optional(results / "source-manifest.json") or "[]")
    expected = {row["path"]: row["sha256"] for row in manifest}
    tested = {row["path"]: row["sha256"] for row in result_manifest}
    exact_source = expected == tested and bool(expected)
    actual_matches = all((source / row["path"]).is_file() and
                         hashlib.sha256((source / row["path"]).read_bytes()).hexdigest() == row["sha256"]
                         for row in manifest if row.get("symlink") is None)
    evidence_usable = exact_source and actual_matches
    tests = unittest_results(read_optional(results / "offline.log")) if evidence_usable else []
    harness = read_optional(results / "integration.log") if evidence_usable else ""
    # The console can contain ANSI colour resets between the marker and label.
    harness = re.sub(r"\x1b\[[0-9;]*m", "", harness)
    harness_cases = [{"scenario": label.split(" -- ", 1)[0],
                      "status": {"PASS": "passed", "FAIL": "failed", "SKIP": "skipped"}[state],
                      "scope": "Guarded single-commit-server harness branch", "diagnostic": label}
                     for state, label in re.findall(r"\[(PASS|FAIL|SKIP)\] (.+)", harness)]
    failures = [row["diagnostic"] for row in harness_cases if row["status"] == "failed"]
    outcomes = json.loads(read_optional(results / "test-outcomes.json") or "{}") if evidence_usable else {}
    provision = read_optional(results / "provision.log") if evidence_usable else ""
    bootstrap = read_optional(results / "bootstrap-console.log") if evidence_usable else ""
    runtime = []
    if evidence_usable:
        for log in ["runtime.log", "maintenance.log"]:
            try:
                runtime.extend(json.loads(read_optional(results / log) or "{}").get("scenarios", []))
            except json.JSONDecodeError:
                pass
    rows = []
    for entry in sorted(manifest, key=lambda row: row["path"]):
        relative = Path(entry["path"])
        path = source / relative
        if not path.is_file():
            continue
        kind = script_kind(relative, path.read_bytes())
        if kind is None:
            continue
        key = entry["path"] if entry["path"] in ISOLATED else path.name
        suites = ISOLATED.get(key, [])
        isolated = [row for row in tests if row["suite"] in suites or row["test"] in suites]
        live = [row for row in harness_cases if any(row["scenario"] == match or row["scenario"].startswith(match + " (")
                                                   for match in HARNESS.get(path.name, []))]
        live += [{"scenario": row["name"], "status": row["status"], "scope": row["detail"], "error": row.get("error")}
                 for row in runtime if path.name in row.get("scripts", [])]
        if path.name in {"reset.sh", "configure_new_server.sh", "mkdirs.sh"} and provision and bootstrap:
            live.append({"scenario": "container-provisioning",
                         "status": "passed" if "Provisioning complete; systemd owns the disposable instance." in bootstrap else "failed",
                         "scope": "Fresh marked single-instance deployment, configuration and systemd startup in this container",
                         "log": "provision.log"})
        if path.name == "p4d_instance_init.template":
            live += [dict(row, scope="Installed init template: " + row["scope"]) for row in harness_cases
                     if row["scenario"].startswith("p4d init:")]
            live += [{"scenario": row["name"], "status": row["status"], "scope": "Installed init template: " + row["detail"], "error": row.get("error")}
                     for row in runtime if row["name"] == "database_switch"]
        if path.name == "gen_sudoers.sh" and evidence_usable:
            sudoers_log = read_optional(results / "sudoers-preview.log")
            if sudoers_log:
                live.append({"scenario": "sudoers-preview", "status": "passed" if "SDP sudoers preview validation passed" in sudoers_log else "failed",
                             "scope": "Generate a limited sudoers preview against the disposable installation and validate with visudo; no policy installation",
                             "log": "sudoers-preview.log"})
        if path.name == "install_sdp_python.sh" and "python-installer" in outcomes:
            installer_log = read_optional(results / "python-installer.log")
            live.append({"scenario": "python-installer", "status": "passed" if outcomes["python-installer"] == 0 and installer_log else "failed",
                         "scope": "Actual isolated Python source build and package installation in the container",
                         "exit_code": outcomes["python-installer"], "log": "python-installer.log",
                         "diagnostic": installer_log[-2000:] if outcomes["python-installer"] != 0 or not installer_log else ""})
        static = static_check(path, kind)
        if any(row["status"] == "passed" for row in live):
            level = "live"
        elif any(row["status"] == "passed" for row in isolated):
            level = "isolated"
        elif static["status"] == "passed":
            level = "static_only"
        else:
            level = "untested"
        gap = "Only listed scenarios/branches were exercised; other inputs and failure modes remain unvalidated."
        if level in ("static_only", "untested"):
            gap = "No passing behavioral test recorded. Syntax success does not establish that this script works."
        elif "replica" in path.name or "edge" in path.name or "standby" in path.name:
            gap = "No live commit/replica/edge/standby topology is established by these tests; isolated branch/transfer tests only unless explicitly listed."
        elif "/cloud/" in entry["path"]:
            gap = "No live cloud-provider API or snapshot lifecycle was exercised."
        rows.append({"script": entry["path"], "language": kind, "level": level,
                     "outcome": aggregate_outcome(static, live + isolated),
                     "static": static, "isolated": isolated, "associated_suites": suites,
                     "live": live, "gap": gap})
    return {"schema_version": 1, "source_matches_results": evidence_usable,
            "source_manifest_matches": exact_source, "export_hashes_valid": actual_matches,
            "scope": "Per-script scenario evidence, not statement/branch coverage or certification of all behavior",
            "stage_outcomes": outcomes,
            "failed_stages": sorted(name for name, status in outcomes.items() if status != 0),
            "counts": {level: sum(row["level"] == level for row in rows) for level in ["live", "isolated", "static_only", "untested"]},
            "outcomes": {outcome: sum(row["outcome"] == outcome for row in rows) for outcome in
                         ["failed", "partial", "passed-for-listed-scenarios", "no-behavioral-evidence"]},
            "harness_failures": failures, "scripts": rows}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source", type=Path, required=True)
    parser.add_argument("--results", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    try:
        result = report(args.source, args.results)
        args.output.write_text(json.dumps(result, indent=2) + "\n")
    except (OSError, ValueError, KeyError) as error:
        parser.exit(1, str(error) + "\n")
    print(json.dumps({"coverage_report": str(args.output), "counts": result["counts"],
                      "source_matches_results": result["source_matches_results"]}))
    # Reporting failed evidence must not turn collect/CI into a successful run.
    # The separate release gate additionally requires named scenarios, complete
    # platforms, baseline comparisons and the multi-server topology evidence.
    return int(not result["source_matches_results"] or bool(result["failed_stages"]) or
               result["outcomes"].get("failed", 0) > 0)


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