baseline_comparison.py #1

#!/usr/bin/env python3
"""Require paired baseline regressions; skipped or missing evidence fails."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import sys
import unittest

# Reviewed contracts, deliberately independent of unittest discovery. Deleting
# or disabling a regression must not silently reduce the acceptance criteria.
BASELINE_CASES = {
    "test_baseline_maintenance.BaselineMaintenance.test_" + name for name in (
        "client_age_cutoff_order_swarm_and_invalid_access_match_baseline",
        "client_200_cap_mutations_status_and_disconnect_match_baseline",
        "snapshot_recheck_preserves_baseline_fresh_activity_decision",
        "snapshot_recheck_failure_explicitly_stops_candidate_before_mutation",
        "pending_cleanup_retains_skip_and_failure_policies",
        "group_mutation_preserves_members_owners_subgroups_and_limits",
        "approved_private_scratch_change_preserves_unrelated_group_file",
        "failed_group_write_preserves_same_recovery_spec_and_failure_status",
        "membership_user_creation_deletion_policy_matches_baseline",
        "classic_protection_selection_matches_baseline",
        "approved_host_restricted_protection_removal_changes_only_selected_group",
        "approved_protection_format_preservation_retains_crlf_and_spaces",
        "approved_inventory_completeness_adds_root_file_without_losing_nested_file",
        "approved_inventory_spaces_are_complete_instead_of_partial_failure",
        "approved_atomic_inventory_preserves_previous_output_on_late_query_failure",
    )
} | {
    "test_baseline_runtime.BaselineRuntime.test_" + name for name in (
        "gzip_threshold_and_payloads_match_submitted_behavior",
        "failed_compression_preserves_same_inputs_and_failure_status",
        "disabled_retention_preserves_identical_bytes_and_configuration",
        "approved_retention_difference_keeps_oldest_retained_checkpoint_chain",
    )
}


class Results(unittest.TextTestResult):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.scenarios = []

    def addSuccess(self, test):
        super().addSuccess(test)
        self.scenarios.append({"name": test.id(), "status": "passed"})

    def addFailure(self, test, error):
        super().addFailure(test, error)
        self.scenarios.append({"name": test.id(), "status": "failed", "detail": self._exc_info_to_string(error, test)})

    def addError(self, test, error):
        super().addError(test, error)
        self.scenarios.append({"name": test.id(), "status": "failed", "detail": self._exc_info_to_string(error, test)})

    def addSkip(self, test, reason):
        super().addSkip(test, reason)
        self.scenarios.append({"name": test.id(), "status": "skipped", "detail": reason})

    def addExpectedFailure(self, test, error):
        super().addExpectedFailure(test, error)
        self.scenarios.append({"name": test.id(), "status": "failed", "detail": "Expected failures are not acceptance evidence"})

    def addUnexpectedSuccess(self, test):
        super().addUnexpectedSuccess(test)
        self.scenarios.append({"name": test.id(), "status": "failed", "detail": "Unexpected success requires review"})

    def addSubTest(self, test, subtest, error):
        super().addSubTest(test, subtest, error)
        if error is not None:
            self.scenarios.append({"name": subtest.id(), "status": "failed", "detail": self._exc_info_to_string(error, test)})


def verify_manifest(root):
    path = root / ".export-manifest.json"
    rows = json.loads(path.read_text())
    if not rows:
        raise ValueError("Empty source manifest")
    seen = set()
    for row in rows:
        relative = Path(row["path"])
        if relative.is_absolute() or ".." in relative.parts or row["path"] in seen:
            raise ValueError("Unsafe or duplicate manifest path")
        seen.add(row["path"])
        target = root / relative
        data = os.readlink(target).encode() if target.is_symlink() else target.read_bytes()
        if hashlib.sha256(data).hexdigest() != row["sha256"]:
            raise ValueError("Source changed: " + row["path"])
    return hashlib.sha256(path.read_bytes()).hexdigest()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--baseline", type=Path, required=True)
    parser.add_argument("--candidate", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    evidence = {"schema_version": 1, "scope": "Paired isolated behavior, not existing-install migration or full production equivalence",
                "production_baseline_confirmed": False, "scenarios": []}
    try:
        evidence["baseline_manifest_sha256"] = verify_manifest(args.baseline)
        evidence["candidate_manifest_sha256"] = verify_manifest(args.candidate)
        evidence["baseline"] = json.loads((args.baseline / ".baseline-info.json").read_text())
        evidence["production_baseline_confirmed"] = evidence["baseline"].get("production_baseline_confirmed") is True
        os.environ["SDP_BASELINE_SOURCE"] = str(args.baseline.resolve())
        os.environ["SDP_CANDIDATE_SOURCE"] = str(args.candidate.resolve())
        sys.path.insert(0, str(args.candidate / "test"))
        suite = unittest.defaultTestLoader.loadTestsFromNames(["test_baseline_maintenance", "test_baseline_runtime"])
        result = unittest.TextTestRunner(stream=sys.stderr, verbosity=2, resultclass=Results).run(suite)
        passed = {row["name"] for row in result.scenarios if row["status"] == "passed"}
        missing = sorted(BASELINE_CASES - passed)
        evidence.update(tests_run=result.testsRun, scenarios=result.scenarios, missing_required=missing,
                        status="passed" if result.wasSuccessful() and not result.skipped and
                        not result.expectedFailures and not missing and result.testsRun else "failed")
    except (OSError, ValueError, KeyError) as error:
        evidence.update(status="failed", error=str(error))
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(evidence, indent=2) + "\n")
    print(json.dumps({"status": evidence["status"], "tests_run": evidence.get("tests_run", 0), "output": str(args.output)}))
    return int(evidence["status"] != "passed")


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