gochat/backend/scripts/ci_quality_gate.py

#!/usr/bin/env python3
"""Validate Go test coverage, execution evidence, and elapsed time."""

import argparse
import json
import sys
from collections import defaultdict
from pathlib import Path


def coverage(path):
    files = defaultdict(lambda: [0, 0])
    total = [0, 0]
    for line in Path(path).read_text(encoding="utf-8").splitlines():
        if not line or line.startswith("mode:"):
            continue
        fields = line.split()
        if len(fields) != 3:
            raise ValueError(f"invalid coverage line: {line}")
        file_name = fields[0].split(":", 1)[0]
        statements, count = int(fields[1]), int(fields[2])
        total[0] += statements
        total[1] += statements if count else 0
        files[file_name][0] += statements
        files[file_name][1] += statements if count else 0
    if total[0] == 0:
        raise ValueError("coverage profile has no statements")
    return total, files


def test_evidence(path):
    packages = defaultdict(lambda: {"pass": 0, "fail": 0, "skip": 0})
    totals = {"pass": 0, "fail": 0, "skip": 0}
    for line in Path(path).read_text(encoding="utf-8").splitlines():
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        action = event.get("Action")
        package = event.get("Package", "")
        if not event.get("Test") or action not in totals:
            continue
        packages[package][action] += 1
        totals[action] += 1
    return totals, packages


def package_matches(name, spec):
    recursive = spec.endswith("/...")
    package = spec.removesuffix("/...")
    return name == package or recursive and name.startswith(package + "/")


def run(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument("--coverage", required=True)
    parser.add_argument("--test-json", required=True)
    parser.add_argument("--duration", required=True)
    parser.add_argument("--report", required=True)
    parser.add_argument("--min-coverage", type=float, default=65.0)
    parser.add_argument("--max-duration", type=float, required=True)
    parser.add_argument("--test-exit-code", type=int, default=0)
    parser.add_argument("--critical-package", action="append", default=[])
    args = parser.parse_args(argv)

    failures = []
    try:
        total, files = coverage(args.coverage)
        total_tests, packages = test_evidence(args.test_json)
        elapsed = float(Path(args.duration).read_text(encoding="utf-8").strip())
    except (OSError, ValueError) as error:
        failures.append(str(error))
        total, files = [0, 0], {}
        total_tests, packages, elapsed = {"pass": 0, "fail": 0, "skip": 0}, {}, None

    percent = total[1] / total[0] * 100 if total[0] else 0.0
    if percent < args.min_coverage:
        failures.append(f"overall coverage {percent:.2f}% is below {args.min_coverage:.2f}%")
    if args.test_exit_code:
        failures.append(f"go test exited with {args.test_exit_code}")
    if total_tests["fail"]:
        failures.append(f"test evidence contains {total_tests['fail']} failed test events")
    if elapsed is not None and elapsed > args.max_duration:
        failures.append(f"test duration {elapsed:.2f}s exceeds {args.max_duration:.2f}s")
    if sum(total_tests.values()) == 0:
        failures.append("test evidence contains no test results")

    critical, critical_coverage = [], []
    for spec in args.critical_package:
        matched = {
            name: values for name, values in packages.items() if package_matches(name, spec)
        }
        item = {
            "package": spec,
            "matched": sorted(matched),
            "passed": sum(values["pass"] for values in matched.values()),
            "failed": sum(values["fail"] for values in matched.values()),
            "skipped": sum(values["skip"] for values in matched.values()),
        }
        critical.append(item)
        if not matched:
            failures.append(f"critical suite {spec} produced no package evidence")
        elif item["passed"] == 0:
            failures.append(f"critical suite {spec} was entirely skipped")

        matching = [
            values
            for name, values in files.items()
            if package_matches(name.rsplit("/", 1)[0], spec)
        ]
        statements = sum(values[0] for values in matching)
        covered = sum(values[1] for values in matching)
        coverage_item = {
            "package": spec,
            "covered_statements": covered,
            "total_statements": statements,
            "percent": round(covered / statements * 100, 2) if statements else 0.0,
        }
        critical_coverage.append(coverage_item)
        if not statements:
            failures.append(f"critical coverage {spec} has no statements")
        elif coverage_item["percent"] < args.min_coverage:
            failures.append(
                f"critical coverage {spec} {coverage_item['percent']:.2f}% "
                f"is below {args.min_coverage:.2f}%"
            )

    report = {
        "schema_version": 1,
        "status": "failed" if failures else "passed",
        "thresholds": {
            "min_coverage_percent": args.min_coverage,
            "max_duration_seconds": args.max_duration,
        },
        "coverage": {
            "covered_statements": total[1],
            "total_statements": total[0],
            "percent": round(percent, 2),
            "critical": critical_coverage,
            "files": {
                name: {
                    "covered_statements": values[1],
                    "total_statements": values[0],
                    "percent": round(values[1] / values[0] * 100, 2) if values[0] else 0.0,
                }
                for name, values in sorted(files.items())
            },
        },
        "tests": {**total_tests, "critical": critical},
        "duration_seconds": elapsed,
        "test_exit_code": args.test_exit_code,
        "failures": failures,
    }
    report_path = Path(args.report)
    report_path.parent.mkdir(parents=True, exist_ok=True)
    report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(
        json.dumps(
            {
                "status": report["status"],
                "coverage_percent": report["coverage"]["percent"],
                "tests": report["tests"],
                "duration_seconds": elapsed,
                "failures": failures,
            },
            sort_keys=True,
        )
    )
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(run())