QuantEngineByItz/tools/verify_ci_reproducibility_v1.py

#!/usr/bin/env python3
"""
CI Reproducibility Validator v1.0

Verify that CI produces identical results when run multiple times on the same commit.
This ensures our build and tests are deterministic (no flaky tests, no hidden state).

Usage:
  python3 tools/verify_ci_reproducibility_v1.py --commit <sha> --runs 3
  python3 tools/verify_ci_reproducibility_v1.py --last-commit  # Uses HEAD
"""

import argparse
import json
import os
import subprocess
import sys
import time
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
from typing import List, Optional


@dataclass
class CIRun:
    """Represents a single CI run result."""
    run_number: int
    commit_sha: str
    timestamp: str
    duration_seconds: float
    status: str  # PASS, FAIL, TIMEOUT
    all_jobs_passed: bool
    failed_jobs: List[str]
    job_durations: dict  # job_name -> seconds
    build_outputs_hash: Optional[str]  # Hash of build artifacts


class CIReproducibilityValidator:
    """Validates CI reproducibility and determinism."""

    def __init__(self, repo_path: str = "."):
        self.repo_path = Path(repo_path)
        self.results: List[CIRun] = []

    def get_current_commit(self) -> str:
        """Get current commit SHA (short form)."""
        result = subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            cwd=self.repo_path,
            capture_output=True,
            text=True,
            check=True,
        )
        return result.stdout.strip()

    def trigger_ci(self, commit_sha: str) -> CIRun:
        """
        Trigger CI for a specific commit and wait for completion.

        Note: This is a mock implementation for demonstration.
        In real usage, you would:
        1. Use Gitea API to trigger workflow
        2. Poll for completion
        3. Fetch job results
        """
        run_number = len(self.results) + 1
        print(f"\n[Run {run_number}] Triggering CI for {commit_sha}...")

        # In production, call Gitea API:
        # POST /api/v1/repos/{owner}/{repo}/actions/workflows/{workflow}/dispatches
        # Then poll GET /api/v1/repos/{owner}/{repo}/actions/runs

        # For now, this is a placeholder
        print(f"  ⏳ CI in progress (this is a demo run)...")
        time.sleep(2)

        # Mock CI run result
        run = CIRun(
            run_number=run_number,
            commit_sha=commit_sha,
            timestamp=datetime.utcnow().isoformat(),
            duration_seconds=900.0,  # Mock: 15 minutes
            status="PASS",
            all_jobs_passed=True,
            failed_jobs=[],
            job_durations={
                "core": 450,
                "wbs-audit": 180,
                "dotnet-contracts": 200,
                "ui-storage": 120,
                "database-schema": 90,
                "calibration-pipeline": 150,
                "operational-reporting": 280,
                "security-validation": 60,
                "workflow-lint": 30,
                "notify-results": 20,
            },
            build_outputs_hash=None,
        )

        self.results.append(run)
        return run

    def compare_runs(self) -> dict:
        """Compare multiple CI runs for consistency."""
        if len(self.results) < 2:
            return {"error": "Need at least 2 runs to compare"}

        comparison = {
            "total_runs": len(self.results),
            "commit": self.results[0].commit_sha,
            "consistency": {
                "all_passed": all(r.all_jobs_passed for r in self.results),
                "same_status": len(set(r.status for r in self.results)) == 1,
                "duration_variance": self._calculate_variance(),
                "failed_jobs_consistent": self._check_failed_jobs_consistent(),
            },
            "issues": [],
            "status": "PASS",
        }

        # Check for inconsistencies
        if not comparison["consistency"]["all_passed"]:
            comparison["issues"].append("Not all runs passed")
            comparison["status"] = "FAIL"

        if not comparison["consistency"]["same_status"]:
            comparison["issues"].append("CI status differs between runs")
            comparison["status"] = "FAIL"

        if comparison["consistency"]["duration_variance"] > 0.20:  # 20% variance
            comparison["issues"].append(
                f"High duration variance: {comparison['consistency']['duration_variance']:.1%}"
            )
            # Not a hard fail, but a warning

        if not comparison["consistency"]["failed_jobs_consistent"]:
            comparison["issues"].append("Failed jobs inconsistent between runs")
            comparison["status"] = "FAIL"

        return comparison

    def _calculate_variance(self) -> float:
        """Calculate coefficient of variation in CI durations."""
        if not self.results:
            return 0.0

        durations = [r.duration_seconds for r in self.results]
        mean = sum(durations) / len(durations)
        variance = sum((x - mean) ** 2 for x in durations) / len(durations)
        std_dev = variance ** 0.5

        return std_dev / mean if mean > 0 else 0.0

    def _check_failed_jobs_consistent(self) -> bool:
        """Check if failed jobs are the same across all runs."""
        if not self.results:
            return True

        failed_jobs_sets = [set(r.failed_jobs) for r in self.results]
        return all(s == failed_jobs_sets[0] for s in failed_jobs_sets)

    def test_reproducibility(self, num_runs: int = 3, commit_sha: Optional[str] = None) -> bool:
        """
        Run CI multiple times and verify reproducibility.

        Args:
            num_runs: Number of times to run CI (default 3)
            commit_sha: Specific commit to test (default current)

        Returns:
            True if all runs are reproducible, False otherwise
        """
        if not commit_sha:
            commit_sha = self.get_current_commit()

        print(f"\n{'='*70}")
        print(f"CI Reproducibility Test: {num_runs} runs on commit {commit_sha}")
        print(f"{'='*70}")

        for i in range(num_runs):
            run = self.trigger_ci(commit_sha)
            print(f"  ✓ Run {i+1}: {run.status} ({run.duration_seconds:.0f}s)")

        comparison = self.compare_runs()

        print(f"\n{'─'*70}")
        print("Comparison Results:")
        print(f"{'─'*70}")

        for key, value in comparison["consistency"].items():
            status = "✓" if value else "✗"
            print(f"  {status} {key}: {value}")

        if comparison["issues"]:
            print("\nIssues Found:")
            for issue in comparison["issues"]:
                print(f"  ⚠ {issue}")

        print(f"\nOverall: {comparison['status']}")
        print(f"{'='*70}\n")

        return comparison["status"] == "PASS"

    def generate_report(self, output_file: str = "Temp/ci_reproducibility_report.json"):
        """Generate detailed report of reproducibility test."""
        Path("Temp").mkdir(exist_ok=True)

        report = {
            "timestamp": datetime.utcnow().isoformat(),
            "total_runs": len(self.results),
            "runs": [asdict(r) for r in self.results],
            "comparison": self.compare_runs(),
            "version": "1.0",
        }

        with open(output_file, "w") as f:
            json.dump(report, f, indent=2)

        print(f"\n✓ Report saved to {output_file}")
        return output_file


def main():
    parser = argparse.ArgumentParser(
        description="Verify CI reproducibility and determinism"
    )
    parser.add_argument(
        "--commit",
        help="Specific commit SHA to test (default: current)",
    )
    parser.add_argument(
        "--runs",
        type=int,
        default=3,
        help="Number of CI runs (default: 3)",
    )
    parser.add_argument(
        "--last-commit",
        action="store_true",
        help="Test current HEAD commit",
    )
    parser.add_argument(
        "--report",
        default="Temp/ci_reproducibility_report.json",
        help="Output report file (default: Temp/ci_reproducibility_report.json)",
    )

    args = parser.parse_args()

    validator = CIReproducibilityValidator()

    commit_sha = None
    if args.last_commit or not args.commit:
        commit_sha = validator.get_current_commit()
        print(f"Testing current commit: {commit_sha}")
    else:
        commit_sha = args.commit

    # Run reproducibility test
    success = validator.test_reproducibility(
        num_runs=args.runs,
        commit_sha=commit_sha
    )

    # Generate report
    validator.generate_report(args.report)

    # Exit with appropriate code
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()