bstein-dev-home/testing/ci/publish_metrics.py

from __future__ import annotations

"""Command-line entry point for publishing CI test metrics."""

import argparse
import json
import os
from pathlib import Path

from .summary import load_junit_cases, load_junit_summary, publish_quality_metrics


def _build_parser() -> argparse.ArgumentParser:
    """Build the CLI parser for the metrics publisher."""

    parser = argparse.ArgumentParser(description="Publish test-suite metrics to Pushgateway")
    parser.add_argument("--gateway", required=True, help="Pushgateway base URL")
    parser.add_argument("--suite", required=True, help="Logical suite name")
    parser.add_argument("--job", default="platform-quality-ci", help="Pushgateway job label")
    parser.add_argument("--status", choices=("ok", "failed"), required=True, help="Gate outcome")
    parser.add_argument("--junit", nargs="*", default=(), help="JUnit XML files to aggregate")
    parser.add_argument("--quality-report", default="build/quality-gate.json", help="Quality gate JSON report")
    branch_default = os.getenv("BRANCH_NAME") or os.getenv("GIT_BRANCH") or "unknown"
    if branch_default.startswith("origin/"):
        branch_default = branch_default[len("origin/") :]
    parser.add_argument("--branch", default=branch_default, help="SCM branch")
    parser.add_argument("--build-number", default=os.getenv("BUILD_NUMBER", ""), help="Jenkins build number")
    parser.add_argument("--jenkins-job", default=os.getenv("JOB_NAME", "bstein-dev-home"), help="Jenkins job name")
    return parser


def _load_quality_report(path: Path) -> tuple[float, int, dict[str, str]]:
    """Read workspace coverage/LOC summary from the quality gate JSON output."""

    if not path.exists():
        return 0.0, 0, {
            "tests": "not_applicable",
            "coverage": "not_applicable",
            "loc": "not_applicable",
            "docs_naming": "not_applicable",
            "gate_glue": "ok",
            "sonarqube": "not_applicable",
            "supply_chain": "not_applicable",
        }
    payload = json.loads(path.read_text(encoding="utf-8"))
    coverage = payload.get("workspace_line_coverage_percent")
    if not isinstance(coverage, (int, float)):
        coverage = 0.0
    source_lines = payload.get("source_lines_over_500")
    if not isinstance(source_lines, int):
        source_lines = 0
    issue_checks = [item.get("check") for item in payload.get("issues", []) if isinstance(item, dict)]
    docs_failed = any(str(check).lower() in {"docstring", "docs", "naming"} for check in issue_checks)
    coverage_failed = any(str(check).lower() == "coverage" for check in issue_checks)
    loc_failed = any(str(check).lower() in {"loc", "smell"} for check in issue_checks) or source_lines > 0
    checks = {
        "tests": "ok" if payload.get("issue_count", 0) == 0 else "failed",
        "coverage": "failed" if coverage_failed or float(coverage) < 95.0 else "ok",
        "loc": "failed" if loc_failed else "ok",
        "docs_naming": "failed" if docs_failed else "ok",
        "gate_glue": "ok",
        "sonarqube": "not_applicable",
        "supply_chain": "not_applicable",
    }
    sonarqube_report = Path(os.getenv("QUALITY_GATE_SONARQUBE_REPORT", "build/sonarqube-quality-gate.json"))
    if sonarqube_report.exists():
        try:
            sonarqube_payload = json.loads(sonarqube_report.read_text(encoding="utf-8"))
            status = (
                sonarqube_payload.get("status")
                or (sonarqube_payload.get("projectStatus") or {}).get("status")
                or (sonarqube_payload.get("qualityGate") or {}).get("status")
            )
            if isinstance(status, str):
                checks["sonarqube"] = "ok" if status.strip().lower() in {"ok", "pass", "passed", "success"} else "failed"
        except Exception:
            checks["sonarqube"] = "failed"
    ironbank_report = Path(os.getenv("QUALITY_GATE_IRONBANK_REPORT", "build/ironbank-compliance.json"))
    if ironbank_report.exists():
        try:
            ironbank_payload = json.loads(ironbank_report.read_text(encoding="utf-8"))
            compliant = ironbank_payload.get("compliant")
            if isinstance(compliant, bool):
                checks["supply_chain"] = "ok" if compliant else "failed"
            else:
                status = ironbank_payload.get("status") or ironbank_payload.get("result")
                if isinstance(status, str):
                    checks["supply_chain"] = (
                        "ok" if status.strip().lower() in {"ok", "pass", "passed", "success", "compliant"} else "failed"
                    )
        except Exception:
            checks["supply_chain"] = "failed"
    return float(coverage), int(source_lines), checks


def main(argv: list[str] | None = None) -> int:
    """Parse arguments, aggregate JUnit files, and publish metrics."""

    parser = _build_parser()
    args = parser.parse_args(argv)
    junit_paths = [Path(path) for path in args.junit]
    summary = load_junit_summary(junit_paths)
    test_cases = load_junit_cases(junit_paths)
    coverage_percent, source_lines_over_500, checks = _load_quality_report(Path(args.quality_report))
    publish_quality_metrics(
        gateway=args.gateway,
        suite=args.suite,
        job=args.job,
        status=args.status,
        summary=summary,
        workspace_line_coverage_percent=coverage_percent,
        source_lines_over_500=source_lines_over_500,
        branch=args.branch,
        build_number=args.build_number,
        jenkins_job=args.jenkins_job,
        checks=checks,
        test_cases=test_cases,
    )
    return 0


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