#!/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""" Coverage Trend Report - Atom

📊 Coverage Trend Report

Last updated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
Current Coverage
{analysis.get("current_coverage", 0):.1f}%
{trend_emoji} {analysis.get("trend_direction", "unknown").title()}
30-Day Average
{analysis.get("thirty_day_avg", 0):.1f}%
Moving average
Week-over-Week
{analysis.get("week_over_week_change", 0):+.1f}%
Change from last week
Target Prediction
{analysis.get("target_prediction", {}).get("estimated_date", "N/A")}
Confidence: {analysis.get("target_prediction", {}).get("confidence", "N/A").title()}
Coverage History (Last 30 Data Points)
{chart_svg}
{f'''
Regression Alerts
{''.join(f'
{a.get("message", "")}
' for a in alerts)} {'
No regression alerts detected
' if not alerts else ''}
''' if alerts else ''}
""" # Write HTML file output_file = Path(output_path) output_file.parent.mkdir(parents=True, exist_ok=True) with open(output_file, 'w') as f: f.write(html_content) print(f"HTML report generated: {output_path}") def generate_svg_chart(dates: List[str], values: List[float], target: float) -> str: """Generate SVG line chart for coverage trends.""" if not values: return '
No data available
' width = 800 height = 400 padding = 40 # Calculate scales min_val = min(values) max_val = max(max(values), target) val_range = max_val - min_val or 1 x_step = (width - 2 * padding) / max(len(values) - 1, 1) # Generate points points = [] for i, (date, value) in enumerate(zip(dates, values)): x = padding + i * x_step y = height - padding - ((value - min_val) / val_range) * (height - 2 * padding) points.append((x, y)) # Generate SVG svg_lines = [] # Grid lines for i in range(5): y = padding + i * (height - 2 * padding) / 4 val = max_val - i * val_range / 4 svg_lines.append(f'') svg_lines.append(f'{val:.1f}%') # Target line target_y = height - padding - ((target - min_val) / val_range) * (height - 2 * padding) svg_lines.append(f'') svg_lines.append(f'Target ({target}%)') # Data line if len(points) > 1: path_data = "M" + " L".join(f"{x:.1f},{y:.1f}" for x, y in points) svg_lines.append(f'') # Area fill area_path = path_data + f" L{points[-1][0]:.1f},{height-padding} L{points[0][0]:.1f},{height-padding} Z" svg_lines.append(f'') # Data points for i, (x, y) in enumerate(points): color = "#28a745" if values[i] >= target else "#dc3545" if values[i] < 70 else "#ffc107" svg_lines.append(f'') # Show date for some points if i % max(len(points) // 5, 1) == 0: svg_lines.append(f'{dates[i]}') svg = f''' {''.join(svg_lines)} ''' return svg def main(): """Main entry point for trend generation.""" parser = argparse.ArgumentParser( description="Generate coverage trend reports with analysis and predictions" ) parser.add_argument( "--coverage-json", default=DEFAULT_COVERAGE_JSON, help="Path to coverage.json file" ) parser.add_argument( "--trending-json", default=DEFAULT_TRENDING_JSON, help="Path to trending.json file" ) parser.add_argument( "--html-output", default=DEFAULT_HTML_OUTPUT, help="Path for HTML report output" ) parser.add_argument( "--target", type=float, default=80.0, help="Coverage target percentage (default: 80)" ) parser.add_argument( "--phase", default=os.getenv("GSD_PHASE", "090"), help="Current phase number" ) parser.add_argument( "--plan", default=os.getenv("GSD_PLAN", "03"), help="Current plan number" ) args = parser.parse_args() # Load current coverage print(f"Loading coverage from: {args.coverage_json}") coverage_data = load_current_coverage(args.coverage_json) if not coverage_data: sys.exit(1) # Extract coverage metrics totals = coverage_data["totals"] current_coverage = totals["percent_covered"] files_covered = totals.get("num_statements", 0) # Using statements as proxy for files files_total = totals.get("covered_lines", 0) + totals.get("missing_lines", 0) branches_covered = totals.get("covered_branches", 0) branches_total = totals.get("num_branches", 0) print(f"Current coverage: {current_coverage:.2f}%") # Load trending data print(f"Loading trending data from: {args.trending_json}") trending = load_trending_data(args.trending_json) # Get git metrics git_metrics = get_git_metrics() # Create new history entry new_entry = { "date": datetime.now().isoformat() + "Z", "phase": args.phase, "plan": args.plan, "coverage_percent": round(current_coverage, 2), "files_covered": files_covered, "files_total": files_total, "branches_covered": branches_covered, "branches_total": branches_total, "new_files_added": git_metrics["new_files_added"], "modified_files": git_metrics["modified_files"], "trend": "stable" } # Append to history trending["coverage_history"].append(new_entry) # Calculate trend metrics print("Calculating trend metrics...") trend_metrics = calculate_trend_metrics(trending["coverage_history"]) # Detect regression if len(trending["coverage_history"]) >= 2: baseline = trending["coverage_history"][-2]["coverage_percent"] regression = detect_regression(current_coverage, baseline) else: regression = {"regression_detected": False, "severity": "none", "message": "Insufficient data"} # Add alert if regression detected if regression["regression_detected"]: trending["regression_alerts"].append({ "date": datetime.now().isoformat() + "Z", "severity": regression["severity"], "message": regression["message"], "from": baseline, "to": current_coverage }) # Update trending.json structure to maintain compatibility with existing data # Keep old structure but add new fields if "history" not in trending: trending["history"] = [] for entry in trending.get("coverage_history", []): trending["history"].append({ "date": entry["date"], "phase": entry.get("phase", ""), "plan": entry.get("plan", ""), "coverage_pct": entry["coverage_percent"], "lines_covered": entry.get("files_covered", 0), "lines_total": entry.get("files_total", 0), "trend": entry.get("trend", "stable") }) trending["latest"] = trending["history"][-1] if trending["history"] else {} # Predict target date print(f"Predicting target date ({args.target}%)...") prediction = predict_target_date(trending["coverage_history"], args.target) # Update trend_analysis section trending["trend_analysis"] = { "current_coverage": round(current_coverage, 2), "seven_day_avg": trend_metrics["seven_day_avg"], "thirty_day_avg": trend_metrics["thirty_day_avg"], "week_over_week_change": trend_metrics["week_over_week_change"], "trend_direction": trend_metrics["trend_direction"], "regression_detected": regression["regression_detected"], "target_prediction": prediction, "last_updated": datetime.now().isoformat() + "Z" } # Save updated trending.json trending_path = Path(args.trending_json) trending_path.parent.mkdir(parents=True, exist_ok=True) with open(trending_path, 'w') as f: json.dump(trending, f, indent=2) print(f"Trending data saved to: {args.trending_json}") # Generate HTML report print("Generating HTML report...") generate_html_report(trending, args.html_output) # Add alert if regression detected if regression["regression_detected"]: trending["regression_alerts"].append({ "date": datetime.now().isoformat() + "Z", "severity": regression["severity"], "message": regression["message"], "from": baseline, "to": current_coverage }) # Predict target date print(f"Predicting target date ({args.target}%)...") prediction = predict_target_date(trending["coverage_history"], args.target) # Update trend_analysis section trending["trend_analysis"] = { "current_coverage": round(current_coverage, 2), "seven_day_avg": trend_metrics["seven_day_avg"], "thirty_day_avg": trend_metrics["thirty_day_avg"], "week_over_week_change": trend_metrics["week_over_week_change"], "trend_direction": trend_metrics["trend_direction"], "regression_detected": regression["regression_detected"], "target_prediction": prediction, "last_updated": datetime.now().isoformat() + "Z" } # Save updated trending.json trending_path = Path(args.trending_json) trending_path.parent.mkdir(parents=True, exist_ok=True) with open(trending_path, 'w') as f: json.dump(trending, f, indent=2) print(f"Trending data saved to: {args.trending_json}") # Generate HTML report print("Generating HTML report...") generate_html_report(trending, args.html_output) # Print summary print("\n" + "="*60) print("COVERAGE TREND SUMMARY") print("="*60) print(f"Current Coverage: {current_coverage:.2f}%") print(f"30-Day Average: {trend_metrics['thirty_day_avg']:.2f}%") print(f"Week-over-Week: {trend_metrics['week_over_week_change']:+.2f}%") print(f"Trend Direction: {trend_metrics['trend_direction'].title()}") print(f"Regression Detected: {regression['regression_detected']}") print(f"Target Prediction: {prediction.get('estimated_date', 'N/A')} ({prediction.get('confidence', 'N/A')} confidence)") print("="*60) return 0 if __name__ == "__main__": sys.exit(main())