#!/usr/bin/env python3 """ Generate Phase 101 Coverage Summary Generates comprehensive coverage summary for Phase 101 target services: - agent_governance_service.py - episode_segmentation_service.py - episode_retrieval_service.py - episode_lifecycle_service.py - canvas_tool.py - agent_guidance_canvas_tool.py Output: backend/tests/coverage_reports/metrics/phase_101_coverage_summary.json """ import json import subprocess import sys from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Tuple # Target services for Phase 101 TARGET_SERVICES = [ "backend/core/agent_governance_service.py", "backend/core/episode_segmentation_service.py", "backend/core/episode_retrieval_service.py", "backend/core/episode_lifecycle_service.py", "backend/tools/canvas_tool.py", "backend/tools/agent_guidance_canvas_tool.py", ] # Baseline coverage from coverage_baseline.json BASELINE_COVERAGE = { "backend/core/agent_governance_service.py": 20.0, "backend/core/episode_segmentation_service.py": 8.25, "backend/core/episode_retrieval_service.py": 9.03, "backend/core/episode_lifecycle_service.py": 0.0, # Not in baseline, assuming 0% "backend/tools/canvas_tool.py": 3.8, "backend/tools/agent_guidance_canvas_tool.py": 0.0, # Not in baseline, assuming 0% } def run_coverage() -> Dict[str, Any]: """Run pytest with coverage and return JSON output.""" print("Running coverage analysis for Phase 101 target services...") cmd = [ sys.executable, "-m", "pytest", "backend/tests/unit/governance/", "backend/tests/unit/episodes/", "backend/tests/unit/canvas/", "backend/tests/property_tests/governance/", "backend/tests/property_tests/episodes/", "-v", "--cov=backend/core/agent_governance_service", "--cov=backend/core/episode_segmentation_service", "--cov=backend/core/episode_retrieval_service", "--cov=backend/core/episode_lifecycle_service", "--cov=backend/tools/canvas_tool", "--cov=backend/tools/agent_guidance_canvas_tool", "--cov-report=term-missing", "--cov-report=json:", "--cov-report=term-missing:skip-covered", ] print(f"Running: {' '.join(cmd)}") result = subprocess.run( cmd, capture_output=True, text=True, cwd=Path(__file__).parent.parent.parent, ) print(result.stdout) if result.stderr: print("STDERR:", result.stderr, file=sys.stderr) # Load coverage.json coverage_file = Path(__file__).parent.parent.parent / "coverage.json" if not coverage_file.exists(): print(f"ERROR: coverage.json not found at {coverage_file}", file=sys.stderr) return {} with open(coverage_file) as f: coverage_data = json.load(f) return coverage_data def extract_service_metrics(coverage_data: Dict[str, Any]) -> List[Dict[str, Any]]: """Extract coverage metrics for each target service.""" services = [] for service_path in TARGET_SERVICES: # Normalize path for coverage.json lookup normalized_path = service_path.replace("backend/", "") # Find coverage data files = coverage_data.get("files", {}) coverage_info = None # Try exact match first if normalized_path in files: coverage_info = files[normalized_path] else: # Try finding by partial match for file_path, file_data in files.items(): if service_path.split("/")[-1] in file_path: coverage_info = file_data normalized_path = file_path break if not coverage_info: print(f"WARNING: No coverage data found for {service_path}", file=sys.stderr) continue summary = coverage_info.get("summary", {}) total_lines = summary.get("num_statements", 0) covered_lines = summary.get("covered_lines", 0) coverage_pct = summary.get("percent_covered", 0.0) # Get baseline coverage baseline_pct = BASELINE_COVERAGE.get(service_path, 0.0) # Calculate baseline covered lines estimate baseline_covered = int(total_lines * baseline_pct / 100) # Estimate tests added (rough estimate based on coverage improvement) # Assume each test covers ~20 lines on average new_lines_covered = covered_lines - baseline_covered tests_added = max(0, round(new_lines_covered / 20)) services.append({ "file": service_path, "coverage_pct_before": round(baseline_pct, 2), "coverage_pct_after": round(coverage_pct, 2), "covered_lines_before": baseline_covered, "covered_lines_after": covered_lines, "total_lines": total_lines, "threshold_met": coverage_pct >= 60.0, "tests_added": tests_added, "delta_pct": round(coverage_pct - baseline_pct, 2), }) return services def generate_summary(services: List[Dict[str, Any]]) -> Dict[str, Any]: """Generate overall summary metrics.""" total_services = len(services) services_meeting_threshold = sum(1 for s in services if s["threshold_met"]) # Calculate overall coverage (weighted by lines of code) total_lines = sum(s["total_lines"] for s in services) total_covered = sum(s["covered_lines_after"] for s in services) overall_coverage = round((total_covered / total_lines * 100) if total_lines > 0 else 0, 2) # Calculate baseline overall coverage total_covered_before = sum(s["covered_lines_before"] for s in services) baseline_coverage = round((total_covered_before / total_lines * 100) if total_lines > 0 else 0, 2) # Total tests added (sum of estimates) total_tests_added = sum(s["tests_added"] for s in services) return { "total_services": total_services, "services_meeting_threshold": services_meeting_threshold, "overall_coverage_pct_before": baseline_coverage, "overall_coverage_pct_after": overall_coverage, "threshold_met": services_meeting_threshold == total_services, "tests_added": total_tests_added, "delta_pct": round(overall_coverage - baseline_coverage, 2), } def main(): """Main execution.""" print("=" * 80) print("Phase 101 Coverage Summary Generator") print("=" * 80) # Run coverage analysis coverage_data = run_coverage() if not coverage_data: print("ERROR: Failed to generate coverage data", file=sys.stderr) sys.exit(1) # Extract metrics for each service services = extract_service_metrics(coverage_data) if not services: print("ERROR: No service metrics extracted", file=sys.stderr) sys.exit(1) # Generate summary summary = generate_summary(services) # Build final JSON structure output = { "phase": "101", "generated_at": datetime.utcnow().isoformat() + "+00:00Z", "baseline_version": "v5.0", "target_threshold": 60.0, "summary": summary, "services": services, } # Write output file output_dir = Path(__file__).parent.parent / "coverage_reports" / "metrics" output_dir.mkdir(parents=True, exist_ok=True) output_file = output_dir / "phase_101_coverage_summary.json" with open(output_file, "w") as f: json.dump(output, f, indent=2) print("\n" + "=" * 80) print("Phase 101 Coverage Summary") print("=" * 80) print(f"Total Services: {summary['total_services']}") print(f"Services Meeting 60% Threshold: {summary['services_meeting_threshold']}") print(f"Overall Coverage (Before): {summary['overall_coverage_pct_before']}%") print(f"Overall Coverage (After): {summary['overall_coverage_pct_after']}%") print(f"Delta: +{summary['delta_pct']}%") print(f"Threshold Met: {summary['threshold_met']}") print(f"Estimated Tests Added: {summary['tests_added']}") print(f"\nOutput written to: {output_file}") print("=" * 80) # Print individual service breakdown print("\nIndividual Service Breakdown:") print("-" * 80) for service in services: status = "✓ PASS" if service["threshold_met"] else "✗ FAIL" print(f"{status} | {service['file']}") print(f" Before: {service['coverage_pct_before']}% → After: {service['coverage_pct_after']}% (Delta: +{service['delta_pct']}%)") print(f" Lines: {service['covered_lines_after']}/{service['total_lines']} ({service['coverage_pct_after']}%)") print(f" Tests Added: ~{service['tests_added']}") # Exit with error if threshold not met if not summary["threshold_met"]: print("\nERROR: Not all services meet 60% threshold", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()