tests/scripts/ci_trending_integration.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CI Trending Integration Script

Purpose: Orchestrates all trending steps in CI environment, handles missing
coverage files gracefully, and provides structured output for job summaries.

Usage:
    python ci_trending_integration.py [options]

Options:
    --coverage-dir PATH       Path to coverage artifacts (default: ../coverage/)
    --output-dir PATH         Path for output files (default: tests/coverage_reports/)
    --commit-sha STR          Current commit SHA (from GITHUB_SHA)
    --branch STR              Branch name (from GITHUB_REF_NAME)
    --format FORMAT           Output format: json|markdown (default: json)

Example:
    python ci_trending_integration.py --coverage-dir ./coverage/ --output-dir ./output/
    python ci_trending_integration.py --commit-sha abc123 --branch main --format markdown
"""

import argparse
import json
import logging
import sys
from datetime import datetime
from pathlib import Path
try:
    from typing import Dict
except ImportError:
    # Python 3.9+ supports built-in generic types
    Dict = dict

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(levelname)s: %(message)s'
)
logger = logging.getLogger(__name__)

# Default paths
DEFAULT_COVERAGE_DIR = Path("../coverage/")
DEFAULT_OUTPUT_DIR = Path("tests/coverage_reports/")


def check_coverage_files_exist(coverage_dir: Path) -> Dict[str, bool]:
    """
    Check if coverage files exist for each platform.

    Args:
        coverage_dir: Path to coverage artifacts directory

    Returns:
        Dict mapping platform names to file existence status
    """
    platforms = {
        "backend": coverage_dir / "backend" / "coverage.json",
        "frontend": coverage_dir / "frontend" / "coverage-final.json",
        "mobile": coverage_dir / "mobile" / "coverage-final.json",
        "desktop": coverage_dir / "desktop" / "coverage.json"
    }

    existence = {}
    for platform, file_path in platforms.items():
        exists = file_path.exists()
        existence[platform] = exists

        if not exists:
            logger.warning(f"Coverage file not found: {file_path}")

    return existence


def orchestrate_trending(
    coverage_dir: Path,
    output_dir: Path,
    commit_sha: str = "",
    branch: str = ""
) -> Dict:
    """
    Orchestrate all trending steps in CI environment.

    Steps:
    1. Check coverage files exist
    2. Run cross_platform_coverage_gate.py
    3. Run update_cross_platform_trending.py with commit tracking
    4. Run coverage_trend_analyzer.py for regression detection
    5. Run generate_coverage_dashboard.py for visual trends

    Args:
        coverage_dir: Path to coverage artifacts
        output_dir: Path for output files
        commit_sha: Current commit SHA for tracking
        branch: Branch name for tracking

    Returns:
        Dict with:
            - success: bool
            - summary_path: Path or None
            - dashboard_path: Path or None
            - regression_count: int
            - critical_regressions: List[str]
            - errors: List[str]
    """
    result = {
        "success": True,
        "summary_path": None,
        "dashboard_path": None,
        "regression_count": 0,
        "critical_regressions": [],
        "errors": []
    }

    # Create output directory
    output_dir.mkdir(parents=True, exist_ok=True)

    # Step 1: Check coverage files exist
    logger.info("Checking coverage files...")
    file_existence = check_coverage_files_exist(coverage_dir)

    available_platforms = [p for p, exists in file_existence.items() if exists]
    missing_platforms = [p for p, exists in file_existence.items() if not exists]

    if available_platforms:
        logger.info(f"Available platforms: {', '.join(available_platforms)}")
    else:
        error_msg = "No coverage files found for any platform"
        logger.error(error_msg)
        result["errors"].append(error_msg)
        result["success"] = False
        return result

    if missing_platforms:
        logger.warning(f"Missing platforms: {', '.join(missing_platforms)}")
        result["errors"].append(f"Missing coverage for: {', '.join(missing_platforms)}")

    # Step 2: Run cross-platform coverage gate
    logger.info("Running cross-platform coverage gate...")
    summary_path = output_dir / "metrics" / "cross_platform_summary.json"
    summary_path.parent.mkdir(parents=True, exist_ok=True)

    try:
        import subprocess

        cmd = [
            "python",
            "tests/scripts/cross_platform_coverage_gate.py",
            "--backend-coverage", str(coverage_dir / "backend" / "coverage.json"),
            "--frontend-coverage", str(coverage_dir / "frontend" / "coverage-final.json"),
            "--mobile-coverage", str(coverage_dir / "mobile" / "coverage-final.json"),
            "--desktop-coverage", str(coverage_dir / "desktop" / "coverage.json"),
            "--format", "json",
            "--output-json", str(summary_path)
        ]

        process_result = subprocess.run(cmd, capture_output=True, text=True, cwd=".")
        if process_result.returncode != 0:
            logger.warning(f"Coverage gate failed: {process_result.stderr}")
            result["errors"].append(f"Coverage gate failed: {process_result.stderr}")

        if summary_path.exists():
            result["summary_path"] = str(summary_path)
            logger.info(f"Summary generated: {summary_path}")
        else:
            logger.warning("Summary file not created")

    except Exception as e:
        logger.error(f"Error running coverage gate: {e}")
        result["errors"].append(f"Coverage gate error: {str(e)}")

    # Step 3: Update trending data with commit tracking
    if result["summary_path"]:
        logger.info("Updating trending data...")

        try:
            trend_file = output_dir / "metrics" / "cross_platform_trend.json"

            cmd = [
                "python",
                "tests/scripts/update_cross_platform_trending.py",
                "--summary", str(summary_path),
                "--trending-file", str(trend_file),
                "--commit-sha", commit_sha,
                "--branch", branch,
                "--prune"
            ]

            process_result = subprocess.run(cmd, capture_output=True, text=True, cwd=".")
            if process_result.returncode != 0:
                logger.warning(f"Trending update failed: {process_result.stderr}")
                result["errors"].append(f"Trending update failed: {process_result.stderr}")

            logger.info("Trending data updated")

        except Exception as e:
            logger.error(f"Error updating trending data: {e}")
            result["errors"].append(f"Trending update error: {str(e)}")

    # Step 4: Detect regressions
    if result["summary_path"]:
        logger.info("Detecting regressions...")

        try:
            regression_file = output_dir / "metrics" / "coverage_regressions.json"
            trend_file = output_dir / "metrics" / "cross_platform_trend.json"

            cmd = [
                "python",
                "tests/scripts/coverage_trend_analyzer.py",
                "--trending-file", str(trend_file),
                "--regression-threshold", "1.0",
                "--output", str(regression_file),
                "--format", "json"
            ]

            process_result = subprocess.run(cmd, capture_output=True, text=True, cwd=".")

            if regression_file.exists():
                with open(regression_file, 'r') as f:
                    regression_data = json.load(f)

                regressions = regression_data.get("regressions", [])
                result["regression_count"] = len(regressions)

                # Extract critical regressions
                critical = [r["platform"] for r in regressions if r.get("severity") == "critical"]
                result["critical_regressions"] = critical

                logger.info(f"Detected {len(regressions)} regressions ({len(critical)} critical)")
            else:
                logger.info("No regressions detected")

        except Exception as e:
            logger.error(f"Error detecting regressions: {e}")
            result["errors"].append(f"Regression detection error: {str(e)}")

    # Step 5: Generate trend dashboard
    logger.info("Generating trend dashboard...")

    try:
        dashboard_dir = output_dir / "dashboards"
        dashboard_dir.mkdir(parents=True, exist_ok=True)

        dashboard_path = dashboard_dir / "coverage_trend_30d.html"
        trend_file = output_dir / "metrics" / "cross_platform_trend.json"

        cmd = [
            "python",
            "tests/scripts/generate_coverage_dashboard.py",
            "--trending-file", str(trend_file),
            "--output", str(dashboard_path),
            "--days", "30"
        ]

        process_result = subprocess.run(cmd, capture_output=True, text=True, cwd=".")

        if dashboard_path.exists():
            result["dashboard_path"] = str(dashboard_path)
            logger.info(f"Dashboard generated: {dashboard_path}")
        else:
            logger.warning("Dashboard file not created")

    except Exception as e:
        logger.error(f"Error generating dashboard: {e}")
        result["errors"].append(f"Dashboard generation error: {str(e)}")

    # Determine overall success
    if result["critical_regressions"]:
        result["success"] = False

    return result


def generate_ci_summary(results: Dict) -> str:
    """
    Create markdown summary for job posting.

    Args:
        results: Result dict from orchestrate_trending()

    Returns:
        Formatted markdown string
    """
    lines = []
    lines.append("## Coverage Trending Summary")
    lines.append("")
    lines.append(f"**Generated:** {datetime.utcnow().isoformat() + 'Z'}")
    lines.append("")

    # Platform status
    lines.append("### Platform Status")
    lines.append("")

    if results.get("summary_path"):
        lines.append("✅ Coverage summary generated")
    else:
        lines.append("⚠️ Coverage summary not available")

    lines.append("")

    # Regression status
    lines.append("### Regression Status")
    lines.append("")

    regression_count = results.get("regression_count", 0)
    critical = results.get("critical_regressions", [])

    if regression_count > 0:
        lines.append(f"⚠️ **{regression_count} regression(s) detected**")
        lines.append("")

        if critical:
            lines.append(f"**Critical ({len(critical)}):** {', '.join(critical)}")
            lines.append("")

        lines.append("See detailed regression report in artifacts")
    else:
        lines.append("✅ **No regressions detected**")

    lines.append("")

    # Dashboard status
    lines.append("### Dashboard")
    lines.append("")

    if results.get("dashboard_path"):
        lines.append("✅ Trend dashboard generated")
        lines.append(f"Path: `{results['dashboard_path']}`")
    else:
        lines.append("⚠️ Dashboard not available")

    lines.append("")

    # Errors
    if results.get("errors"):
        lines.append("### Warnings/Errors")
        lines.append("")

        for error in results["errors"]:
            lines.append(f"- ⚠️ {error}")

        lines.append("")

    # Overall status
    lines.append("### Overall Status")
    lines.append("")

    if results.get("success"):
        lines.append("✅ **Trending analysis complete**")
    else:
        lines.append("❌ **Critical issues detected**")
        lines.append("")
        lines.append("Build may fail due to critical regressions")

    lines.append("")

    return "\n".join(lines)


def main():
    """Main execution function."""
    parser = argparse.ArgumentParser(
        description="CI trending integration script"
    )

    parser.add_argument(
        "--coverage-dir",
        type=Path,
        default=DEFAULT_COVERAGE_DIR,
        help="Path to coverage artifacts"
    )

    parser.add_argument(
        "--output-dir",
        type=Path,
        default=DEFAULT_OUTPUT_DIR,
        help="Path for output files"
    )

    parser.add_argument(
        "--commit-sha",
        type=str,
        default="",
        help="Current commit SHA (from GITHUB_SHA)"
    )

    parser.add_argument(
        "--branch",
        type=str,
        default="",
        help="Branch name (from GITHUB_REF_NAME)"
    )

    parser.add_argument(
        "--format",
        type=str,
        choices=["json", "markdown"],
        default="json",
        help="Output format"
    )

    args = parser.parse_args()

    # Run orchestration
    results = orchestrate_trending(
        args.coverage_dir,
        args.output_dir,
        args.commit_sha,
        args.branch
    )

    # Generate output
    if args.format == "json":
        # JSON output for machine parsing
        output = json.dumps(results, indent=2)
        print(output)

        # Exit with error code if critical regressions
        if not results["success"]:
            sys.exit(1)

    elif args.format == "markdown":
        # Markdown output for job summaries
        summary = generate_ci_summary(results)
        print(summary)

        # Exit with error code if critical regressions
        if not results["success"]:
            sys.exit(1)

    return 0


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