Add Phase 2 test defect diagnostics

README.md

@@ -228,6 +228,7 @@ juno-repo/
 - **Reasoning Engine**: Multi-factor confidence scoring with audit trails
 - **Risk Forecasting**: Predictive analytics with 89%+ accuracy
 - **Governance Framework**: Role-based approval workflows
+- **Test Defect Diagnostics**: Automatic analysis of failing tests
 
 **Performance Metrics**:
 - Decision latency: 127ms average

juno-agent/src/phase2/defect_diagnostics.py

@@ -0,0 +1,78 @@
+"""Test Defect Diagnostics Module for JUNO Phase 2.
+
+This module analyzes test results and categorizes failures to assist
+with root cause identification.
+"""
+
+from dataclasses import dataclass, field
+from typing import List, Dict
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class TestCaseResult:
+    """Represents a single test case result."""
+    name: str
+    status: str
+    error: str | None = None
+
+
+@dataclass
+class DiagnosticsReport:
+    """Aggregated diagnostics for a set of test results."""
+    total_tests: int
+    passed: int
+    failed: int
+    failure_rate: float
+    failure_categories: Dict[str, int] = field(default_factory=dict)
+    example_failures: List[Dict[str, str]] = field(default_factory=list)
+
+
+class TestDefectDiagnostics:
+    """Analyze test results and produce defect diagnostics."""
+
+    def analyze(self, results: List[TestCaseResult]) -> DiagnosticsReport:
+        total = len(results)
+        passed = len([r for r in results if r.status.upper() == "PASS"])
+        failed_cases = [r for r in results if r.status.upper() == "FAIL"]
+        failed = len(failed_cases)
+        failure_rate = failed / total * 100 if total else 0.0
+
+        categories: Dict[str, int] = {}
+        examples: List[Dict[str, str]] = []
+
+        for case in failed_cases:
+            if case.error:
+                category = case.error.split(":", 1)[0]
+            else:
+                category = "UnknownError"
+            categories[category] = categories.get(category, 0) + 1
+            if len(examples) < 5:
+                examples.append({"test": case.name, "error": case.error or ""})
+
+        report = DiagnosticsReport(
+            total_tests=total,
+            passed=passed,
+            failed=failed,
+            failure_rate=round(failure_rate, 2),
+            failure_categories=categories,
+            example_failures=examples,
+        )
+
+        logger.info(
+            "Generated test defect diagnostics: %s failed/%s total",
+            failed,
+            total,
+        )
+        return report
+
+
+if __name__ == "__main__":
+    sample = [
+        TestCaseResult(name="test_example", status="FAIL", error="AssertionError: expected 1"),
+        TestCaseResult(name="test_ok", status="PASS"),
+    ]
+    diag = TestDefectDiagnostics().analyze(sample)
+    print(diag)

juno-agent/src/phase2/test_suite.py

@@ -22,6 +22,7 @@
 from velocity_analysis import VelocityAnalyzer, TrendDirection
 from stale_triage_resolution import StaleTriageEngine, TriageAction, StalenessLevel
 from governance_framework import GovernanceRoleManager, ApprovalWorkflowEngine, GovernanceRole
+from defect_diagnostics import TestDefectDiagnostics, TestCaseResult
 
 logger = logging.getLogger(__name__)
 
@@ -35,6 +36,7 @@ def __init__(self):
         self.test_results = {}
         self.performance_metrics = {}
         self.validation_results = {}
+        self.defect_diagnostics = {}
 
     def run_all_tests(self) -> Dict[str, Any]:
         """Run complete test suite and return results."""
@@ -57,7 +59,10 @@ def run_all_tests(self) -> Dict[str, Any]:
 
         # Validation tests
         self.validation_results = self._run_validation_tests()
-
+
+        # Defect diagnostics
+        self.defect_diagnostics = self._run_defect_diagnostics()
+
         # Generate summary
         summary = self._generate_test_summary()
 
@@ -613,8 +618,30 @@ def _run_validation_tests(self) -> Dict[str, Any]:
             }
         else:
             validation["triage_recommendation_accuracy"] = {"accuracy": "FAIL", "reason": "No recommendations generated"}
-        
+
         return validation
+
+    def _run_defect_diagnostics(self) -> Dict[str, Any]:
+        """Analyze failed tests and categorize defects."""
+        diagnostics_engine = TestDefectDiagnostics()
+        cases: List[TestCaseResult] = []
+
+        for component, results in self.test_results.items():
+            for entry in results.get("tests", []):
+                cases.append(
+                    TestCaseResult(
+                        name=f"{component}:{entry.get('name')}",
+                        status=entry.get('status', 'UNKNOWN'),
+                        error=entry.get('error'),
+                    )
+                )
+
+        report = diagnostics_engine.analyze(cases)
+        return {
+            "failure_rate": report.failure_rate,
+            "failure_categories": report.failure_categories,
+            "example_failures": report.example_failures,
+        }
 
     def _generate_test_summary(self) -> Dict[str, Any]:
         """Generate comprehensive test summary."""
@@ -649,6 +676,7 @@ def _generate_test_summary(self) -> Dict[str, Any]:
             "component_results": self.test_results,
             "performance_metrics": self.performance_metrics,
             "validation_results": self.validation_results,
+            "defect_diagnostics": self.defect_diagnostics,
             "recommendations": self._generate_recommendations()
         }
 
@@ -712,6 +740,15 @@ def _generate_recommendations(self) -> List[str]:
     for test_name, result in results['validation_results'].items():
         status = "✅" if result.get('accuracy') == 'PASS' else "❌"
         print(f"  {status} {test_name}: {result.get('accuracy', 'N/A')}")
+
+    print("\n🛠 Defect Diagnostics:")
+    print(f"  Failure Rate: {results['defect_diagnostics']['failure_rate']}%")
+    for cat, count in results['defect_diagnostics']['failure_categories'].items():
+        print(f"  - {cat}: {count}")
+    if results['defect_diagnostics']['example_failures']:
+        print("  Examples:")
+        for ex in results['defect_diagnostics']['example_failures']:
+            print(f"    * {ex['test']}: {ex['error']}")
 
     print("\n💡 Recommendations:")
     for rec in results['recommendations']:

tests/unit/test_defect_diagnostics.py

@@ -0,0 +1,27 @@
+"""Unit tests for TestDefectDiagnostics."""
+import unittest
+import sys
+
+sys.path.append('juno-agent/src/phase2')
+
+from defect_diagnostics import TestDefectDiagnostics, TestCaseResult
+
+
+class TestDiagnosticsModule(unittest.TestCase):
+    def test_basic_analysis(self):
+        diagnostics = TestDefectDiagnostics()
+        sample = [
+            TestCaseResult(name="a", status="PASS"),
+            TestCaseResult(name="b", status="FAIL", error="AssertionError: boom"),
+            TestCaseResult(name="c", status="FAIL", error="TimeoutError: slow"),
+        ]
+        report = diagnostics.analyze(sample)
+        self.assertEqual(report.passed, 1)
+        self.assertEqual(report.failed, 2)
+        self.assertAlmostEqual(report.failure_rate, 66.67, places=1)
+        self.assertIn("AssertionError", report.failure_categories)
+        self.assertIn("TimeoutError", report.failure_categories)
+
+
+if __name__ == "__main__":
+    unittest.main()