#!/usr/bin/env python3 """ Coverage Trend Generation and Analysis Script Tracks coverage trends over time, detects regressions, and predicts completion dates. Generates HTML reports with visual charts and trend analysis. Usage: python tests/scripts/generate_coverage_trend.py --help python tests/scripts/generate_coverage_trend.py --html-output /tmp/report.html python tests/scripts/generate_coverage_trend.py --coverage-json custom/coverage.json """ import argparse import json import os import subprocess import sys from datetime import datetime, timedelta from pathlib import Path from typing import Dict, Any, List, Tuple, Optional # Default paths (relative to backend directory) DEFAULT_COVERAGE_JSON = "tests/coverage_reports/metrics/coverage.json" DEFAULT_TRENDING_JSON = "tests/coverage_reports/metrics/trending.json" DEFAULT_HTML_OUTPUT = "tests/coverage_reports/metrics/coverage_trend_report.html" def load_current_coverage(coverage_json_path: str) -> Optional[Dict[str, Any]]: """Load current coverage from coverage.json.""" coverage_path = Path(coverage_json_path) if not coverage_path.exists(): print(f"ERROR: Coverage file not found: {coverage_path}") print("Run pytest with coverage first:") print(" pytest --cov=core --cov=api --cov=tools --cov-report=json") return None with open(coverage_path) as f: data = json.load(f) return data def load_trending_data(trending_json_path: str) -> Dict[str, Any]: """Load trending.json or create new structure if doesn't exist.""" trending_path = Path(trending_json_path) if not trending_path.exists(): # Initialize new trending structure return { "coverage_history": [], "trend_analysis": {}, "regression_alerts": [], "baselines": {}, "metadata": { "created": datetime.now().isoformat(), "version": "2.0" } } with open(trending_path) as f: data = json.load(f) # If old format, migrate to new format if "history" in data and "coverage_history" not in data: # Migrate old "history" to new "coverage_history" data["coverage_history"] = [] for entry in data["history"]: data["coverage_history"].append({ "date": entry["date"], "phase": entry.get("phase", ""), "plan": entry.get("plan", ""), "coverage_percent": entry.get("coverage_pct", 0), "files_covered": entry.get("lines_covered", 0), "files_total": entry.get("lines_total", 0), "branches_covered": entry.get("branches_covered", 0), "branches_total": entry.get("branches_total", 0), "new_files_added": 0, "modified_files": 0, "trend": entry.get("trend", "stable") }) data["trend_analysis"] = {} data["regression_alerts"] = [] # Ensure all required keys exist if "coverage_history" not in data: data["coverage_history"] = [] if "trend_analysis" not in data: data["trend_analysis"] = {} if "regression_alerts" not in data: data["regression_alerts"] = [] if "baselines" not in data: data["baselines"] = {} if "metadata" not in data: data["metadata"] = {"version": "2.0"} return data def get_git_metrics() -> Dict[str, int]: """Get file metrics from git diff (new files, modified files).""" try: # Get list of modified/added Python files result = subprocess.run( ["git", "diff", "--name-only", "HEAD~1", "HEAD"], capture_output=True, text=True, timeout=5 ) files = result.stdout.strip().split('\n') if result.stdout.strip() else [] python_files = [f for f in files if f.endswith('.py') and 'core/' in f or 'api/' in f or 'tools/' in f] return { "new_files_added": len([f for f in python_files if 'new file' in result.stdout]), "modified_files": len(python_files) } except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError): # Git not available or error return { "new_files_added": 0, "modified_files": 0 } def calculate_trend_metrics(history: List[Dict[str, Any]]) -> Dict[str, Any]: """Calculate trend metrics from coverage history.""" if len(history) < 2: return { "seven_day_avg": history[0]["coverage_percent"] if history else 0, "thirty_day_avg": history[0]["coverage_percent"] if history else 0, "week_over_week_change": 0, "trend_direction": "stable" } # Get recent history recent_entries = history[-30:] # Last 30 entries # Calculate averages seven_day_entries = recent_entries[-7:] if len(recent_entries) >= 7 else recent_entries thirty_day_entries = recent_entries seven_day_avg = sum(e["coverage_percent"] for e in seven_day_entries) / len(seven_day_entries) thirty_day_avg = sum(e["coverage_percent"] for e in thirty_day_entries) / len(thirty_day_entries) # Calculate week-over-week change if len(history) >= 7: wow_change = history[-1]["coverage_percent"] - history[-7]["coverage_percent"] elif len(history) >= 2: wow_change = history[-1]["coverage_percent"] - history[-2]["coverage_percent"] else: wow_change = 0 # Determine trend direction if wow_change > 0.5: trend_direction = "increasing" elif wow_change < -0.5: trend_direction = "decreasing" else: trend_direction = "stable" return { "seven_day_avg": round(seven_day_avg, 2), "thirty_day_avg": round(thirty_day_avg, 2), "week_over_week_change": round(wow_change, 2), "trend_direction": trend_direction } def detect_regression(current: float, baseline: float, threshold: float = 5.0) -> Dict[str, Any]: """Detect if coverage has regressed beyond threshold.""" diff = current - baseline if diff < -threshold: return { "regression_detected": True, "severity": "high" if diff < -10 else "medium", "change": round(diff, 2), "message": f"Coverage dropped by {abs(diff):.2f}% (threshold: {threshold}%)" } return { "regression_detected": False, "severity": "none", "change": round(diff, 2), "message": "No regression detected" } def predict_target_date(history: List[Dict[str, Any]], target: float = 80) -> Dict[str, Any]: """ Predict when coverage will reach target using linear regression. Returns estimated date and confidence level. """ if len(history) < 3: return { "target_percent": target, "estimated_date": None, "confidence": "low", "message": "Insufficient data for prediction (need 3+ data points)" } # Get last 30 data points recent = history[-30:] # Extract dates and coverage values dates = [] for e in recent: date_str = e["date"] # Handle both 'Z' suffix and timezone-aware formats if date_str.endswith('Z'): date_str = date_str.replace('Z', '+00:00') try: dates.append(datetime.fromisoformat(date_str)) except ValueError: # Fallback for various datetime formats dates.append(datetime.fromisoformat(date_str.replace('+00:00', ''))) coverages = [e["coverage_percent"] for e in recent] # Strip timezone info for calculations first_date = dates[0].replace(tzinfo=None) x_values = [(d.replace(tzinfo=None) - first_date).days for d in dates] # Calculate linear regression: y = mx + b n = len(x_values) sum_x = sum(x_values) sum_y = sum(coverages) sum_xy = sum(x * y for x, y in zip(x_values, coverages)) sum_x2 = sum(x ** 2 for x in x_values) # Calculate slope (m) and intercept (b) denominator = n * sum_x2 - sum_x ** 2 if denominator == 0: return { "target_percent": target, "estimated_date": None, "confidence": "low", "message": "Cannot calculate trend (insufficient variation)" } slope = (n * sum_xy - sum_x * sum_y) / denominator intercept = (sum_y - slope * sum_x) / n # Calculate R-squared for confidence y_mean = sum_y / n ss_tot = sum((y - y_mean) ** 2 for y in coverages) ss_res = sum((y - (slope * x + intercept)) ** 2 for x, y in zip(x_values, coverages)) r_squared = 1 - (ss_res / ss_tot) if ss_tot > 0 else 0 # Determine confidence based on R-squared and data points if r_squared > 0.7 and len(history) >= 10: confidence = "high" elif r_squared > 0.5 and len(history) >= 5: confidence = "medium" else: confidence = "low" # Predict days to reach target if slope <= 0.001: return { "target_percent": target, "estimated_date": None, "confidence": confidence, "message": "Coverage not trending upward (slope: {:.4f})".format(slope) } current_coverage = coverages[-1] if current_coverage >= target: return { "target_percent": target, "estimated_date": dates[-1].strftime("%Y-%m-%d"), "confidence": confidence, "message": "Target already achieved!" } days_to_target = (target - intercept) / slope estimated_date = first_date + timedelta(days=days_to_target) return { "target_percent": target, "estimated_date": estimated_date.strftime("%Y-%m-%d"), "confidence": confidence, "slope": round(slope, 4), "r_squared": round(r_squared, 2), "days_to_target": int(days_to_target), "message": f"Estimated {int(days_to_target)} days to reach {target}% target" } return { "target_percent": target, "estimated_date": estimated_date.strftime("%Y-%m-%d"), "confidence": confidence, "slope": round(slope, 4), "r_squared": round(r_squared, 2), "days_to_target": int(days_to_target), "message": f"Estimated {int(days_to_target)} days to reach {target}% target" } def generate_html_report(trending: Dict[str, Any], output_path: str) -> None: """Generate HTML trend report with charts and visualizations.""" history = trending.get("coverage_history", []) analysis = trending.get("trend_analysis", {}) alerts = trending.get("regression_alerts", []) # Prepare chart data dates = [e["date"][:10] for e in history[-30:]] # Last 30 entries coverage_values = [e["coverage_percent"] for e in history[-30:]] # Generate SVG chart chart_svg = generate_svg_chart(dates, coverage_values, analysis.get("target_prediction", {}).get("target_percent", 80)) # Create trend indicators trend_emoji = "📈" if analysis.get("trend_direction") == "increasing" else "📉" if analysis.get("trend_direction") == "decreasing" else "➡️" trend_color = "green" if analysis.get("trend_direction") == "increasing" else "red" if analysis.get("trend_direction") == "decreasing" else "gray" html_content = f"""