DevOps Metrics Dashboard for QA: DORA Metrics, Test Stability, and Quality Insights

# metrics/dora_metrics_collector.py
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Dict
import requests

@dataclass
class DeploymentEvent:
    timestamp: datetime
    environment: str
    version: str
    status: str  # success, failed, rolled_back
    test_results: Dict

class DORAMetricsCollector:
    def __init__(self, jenkins_url: str, github_url: str):
        self.jenkins_url = jenkins_url
        self.github_url = github_url

    def calculate_deployment_frequency(self, days: int = 30) -> Dict:
        """Calculate deployment frequency and test correlation"""
        deployments = self._fetch_deployments(days=days)

        successful_deployments = [d for d in deployments if d.status == 'success']
        failed_deployments = [d for d in deployments if d.status == 'failed']

        total_days = days
        frequency_per_day = len(successful_deployments) / total_days

        # Analyze test coverage correlation
        test_coverage_correlation = self._correlate_test_coverage_with_success(deployments)

        return {
            'frequency_per_day': frequency_per_day,
            'total_deployments': len(deployments),
            'successful_deployments': len(successful_deployments),
            'failed_deployments': len(failed_deployments),
            'success_rate': len(successful_deployments) / len(deployments) * 100,
            'test_coverage_correlation': test_coverage_correlation,
            'period_days': days
        }

    def calculate_lead_time_for_changes(self, days: int = 30) -> Dict:
        """Calculate lead time with testing phase breakdown"""
        commits = self._fetch_commits(days=days)

        lead_times = []
        testing_durations = []

        for commit in commits:
            # Time from commit to production
            commit_time = commit['timestamp']
            deployment = self._find_deployment_for_commit(commit['sha'])

            if deployment:
                lead_time = (deployment.timestamp - commit_time).total_seconds() / 3600
                lead_times.append(lead_time)

                # Testing phase duration
                test_duration = self._calculate_test_duration(commit['sha'])
                testing_durations.append(test_duration)

        if not lead_times:
            return {'error': 'No data available'}

        avg_lead_time = sum(lead_times) / len(lead_times)
        avg_test_duration = sum(testing_durations) / len(testing_durations)
        testing_percentage = (avg_test_duration / avg_lead_time) * 100

        return {
            'average_lead_time_hours': avg_lead_time,
            'median_lead_time_hours': sorted(lead_times)[len(lead_times)//2],
            'p95_lead_time_hours': sorted(lead_times)[int(len(lead_times)*0.95)],
            'average_testing_duration_hours': avg_test_duration,
            'testing_percentage_of_lead_time': testing_percentage,
            'samples': len(lead_times)
        }

    def calculate_change_failure_rate(self, days: int = 30) -> Dict:
        """Calculate change failure rate with test quality correlation"""
        deployments = self._fetch_deployments(days=days)

        failed_changes = []
        successful_changes = []

        for deployment in deployments:
            if deployment.status in ['failed', 'rolled_back']:
                failed_changes.append(deployment)
            else:
                successful_changes.append(deployment)

        cfr = len(failed_changes) / len(deployments) * 100 if deployments else 0

        # Analyze failures by test coverage
        failures_by_coverage = self._analyze_failures_by_coverage(failed_changes)

        # Identify test gaps in failed deployments
        test_gaps = self._identify_test_gaps(failed_changes)

        return {
            'change_failure_rate_percent': cfr,
            'total_deployments': len(deployments),
            'failed_deployments': len(failed_changes),
            'failures_by_coverage': failures_by_coverage,
            'identified_test_gaps': test_gaps,
            'recommendation': self._generate_cfr_recommendation(cfr, test_gaps)
        }

    def calculate_mttr(self, days: int = 30) -> Dict:
        """Calculate Mean Time to Restore with testing impact analysis"""
        incidents = self._fetch_incidents(days=days)

        restoration_times = []
        test_suite_run_times = []

        for incident in incidents:
            resolution_time = (incident.resolved_at - incident.created_at).total_seconds() / 3600
            restoration_times.append(resolution_time)

            # Time spent running test suites during incident
            test_time = self._calculate_incident_test_time(incident)
            test_suite_run_times.append(test_time)

        if not restoration_times:
            return {'error': 'No incidents in period'}

        avg_mttr = sum(restoration_times) / len(restoration_times)
        avg_test_time = sum(test_suite_run_times) / len(test_suite_run_times)

        return {
            'mean_time_to_restore_hours': avg_mttr,
            'median_mttr_hours': sorted(restoration_times)[len(restoration_times)//2],
            'average_test_execution_time': avg_test_time,
            'test_time_percentage': (avg_test_time / avg_mttr) * 100,
            'incidents_analyzed': len(incidents),
            'recommendation': self._generate_mttr_recommendation(avg_mttr, avg_test_time)
        }

    def _correlate_test_coverage_with_success(self, deployments: List[DeploymentEvent]) -> Dict:
        """Analyze correlation between test coverage and deployment success"""
        coverage_ranges = {
            '0-50%': {'success': 0, 'failed': 0},
            '50-70%': {'success': 0, 'failed': 0},
            '70-85%': {'success': 0, 'failed': 0},
            '85-100%': {'success': 0, 'failed': 0}
        }

        for deployment in deployments:
            coverage = deployment.test_results.get('coverage', 0)

            if coverage < 50:
                range_key = '0-50%'
            elif coverage < 70:
                range_key = '50-70%'
            elif coverage < 85:
                range_key = '70-85%'
            else:
                range_key = '85-100%'

            if deployment.status == 'success':
                coverage_ranges[range_key]['success'] += 1
            else:
                coverage_ranges[range_key]['failed'] += 1

        # Calculate success rates by coverage range
        success_rates = {}
        for range_key, counts in coverage_ranges.items():
            total = counts['success'] + counts['failed']
            if total > 0:
                success_rates[range_key] = (counts['success'] / total) * 100

        return success_rates

# metrics/flaky_test_analyzer.py
from collections import defaultdict
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
import statistics

class FlakyTestAnalyzer:
    def __init__(self, test_results_db):
        self.db = test_results_db

    def identify_flaky_tests(self, days: int = 14, min_runs: int = 10) -> List[Dict]:
        """Identify tests with inconsistent pass/fail patterns"""
        test_runs = self._fetch_test_runs(days=days)

        test_results_map = defaultdict(list)

        for run in test_runs:
            for test in run.tests:
                test_results_map[test.name].append({
                    'status': test.status,
                    'duration': test.duration,
                    'timestamp': run.timestamp,
                    'commit_sha': run.commit_sha,
                    'environment': run.environment
                })

        flaky_tests = []

        for test_name, results in test_results_map.items():
            if len(results) < min_runs:
                continue

            # Calculate flakiness score
            pass_count = sum(1 for r in results if r['status'] == 'passed')
            fail_count = sum(1 for r in results if r['status'] == 'failed')
            total_runs = len(results)

            # A test is flaky if it both passes and fails in the same period
            if pass_count > 0 and fail_count > 0:
                flakiness_score = min(pass_count, fail_count) / total_runs * 100

                # Analyze patterns
                patterns = self._analyze_flaky_patterns(results)

                flaky_tests.append({
                    'test_name': test_name,
                    'flakiness_score': flakiness_score,
                    'pass_rate': (pass_count / total_runs) * 100,
                    'total_runs': total_runs,
                    'pass_count': pass_count,
                    'fail_count': fail_count,
                    'patterns': patterns,
                    'impact': self._calculate_flaky_test_impact(results),
                    'recommendation': self._generate_flaky_test_recommendation(patterns)
                })

        # Sort by flakiness score
        return sorted(flaky_tests, key=lambda x: x['flakiness_score'], reverse=True)

    def _analyze_flaky_patterns(self, results: List[Dict]) -> Dict:
        """Identify patterns in flaky test behavior"""
        patterns = {
            'time_dependent': self._check_time_dependency(results),
            'environment_specific': self._check_environment_specificity(results),
            'sequential_dependency': self._check_sequential_dependency(results),
            'resource_contention': self._check_resource_contention(results),
            'network_dependency': self._check_network_dependency(results)
        }

        return {k: v for k, v in patterns.items() if v['detected']}

    def _check_time_dependency(self, results: List[Dict]) -> Dict:
        """Check if failures correlate with time of day"""
        failures_by_hour = defaultdict(int)
        runs_by_hour = defaultdict(int)

        for result in results:
            hour = result['timestamp'].hour
            runs_by_hour[hour] += 1
            if result['status'] == 'failed':
                failures_by_hour[hour] += 1

        # Find hours with significantly higher failure rates
        avg_failure_rate = sum(failures_by_hour.values()) / sum(runs_by_hour.values())
        problematic_hours = []

        for hour in runs_by_hour:
            if runs_by_hour[hour] >= 3:  # Minimum sample size
                hour_failure_rate = failures_by_hour[hour] / runs_by_hour[hour]
                if hour_failure_rate > avg_failure_rate * 1.5:
                    problematic_hours.append({
                        'hour': hour,
                        'failure_rate': hour_failure_rate * 100
                    })

        return {
            'detected': len(problematic_hours) > 0,
            'problematic_hours': problematic_hours
        }

    def calculate_test_suite_stability(self, days: int = 30) -> Dict:
        """Calculate overall test suite stability metrics"""
        test_runs = self._fetch_test_runs(days=days)

        stability_data = {
            'total_runs': len(test_runs),
            'consistent_runs': 0,
            'flaky_runs': 0,
            'average_duration': 0,
            'duration_variance': 0
        }

        durations = []
        flaky_run_count = 0

        for run in test_runs:
            durations.append(run.total_duration)

            # Count runs with flaky tests
            if run.flaky_test_count > 0:
                flaky_run_count += 1
            else:
                stability_data['consistent_runs'] += 1

        stability_data['flaky_runs'] = flaky_run_count
        stability_data['stability_rate'] = (stability_data['consistent_runs'] / len(test_runs)) * 100

        if durations:
            stability_data['average_duration'] = statistics.mean(durations)
            stability_data['duration_variance'] = statistics.stdev(durations) if len(durations) > 1 else 0
            stability_data['duration_coefficient_of_variation'] = (
                stability_data['duration_variance'] / stability_data['average_duration'] * 100
            )

        return stability_data

# metrics/coverage_analyzer.py
from typing import Dict, List
import json

class CoverageQualityAnalyzer:
    def __init__(self, coverage_data_source, defect_tracking_system):
        self.coverage_source = coverage_data_source
        self.defect_system = defect_tracking_system

    def analyze_coverage_effectiveness(self, release_version: str) -> Dict:
        """Analyze relationship between test coverage and escaped defects"""
        coverage_data = self.coverage_source.get_coverage_for_release(release_version)
        escaped_defects = self.defect_system.get_production_defects(release_version)

        # Group defects by module
        defects_by_module = self._group_defects_by_module(escaped_defects)

        effectiveness_analysis = []

        for module, coverage_info in coverage_data.items():
            module_defects = defects_by_module.get(module, [])

            effectiveness_analysis.append({
                'module': module,
                'line_coverage': coverage_info['line_coverage'],
                'branch_coverage': coverage_info['branch_coverage'],
                'mutation_score': coverage_info.get('mutation_score', 0),
                'escaped_defects': len(module_defects),
                'defect_density': len(module_defects) / coverage_info['lines_of_code'],
                'coverage_effectiveness_score': self._calculate_effectiveness_score(
                    coverage_info,
                    len(module_defects)
                )
            })

        return {
            'release': release_version,
            'overall_line_coverage': self._calculate_overall_coverage(coverage_data, 'line_coverage'),
            'overall_branch_coverage': self._calculate_overall_coverage(coverage_data, 'branch_coverage'),
            'total_escaped_defects': len(escaped_defects),
            'module_analysis': effectiveness_analysis,
            'recommendations': self._generate_coverage_recommendations(effectiveness_analysis)
        }

    def calculate_test_roi(self, period_days: int = 90) -> Dict:
        """Calculate Return on Investment for testing efforts"""
        # Costs
        test_automation_costs = self._calculate_automation_costs(period_days)
        test_execution_costs = self._calculate_execution_costs(period_days)
        test_maintenance_costs = self._calculate_maintenance_costs(period_days)

        total_testing_cost = test_automation_costs + test_execution_costs + test_maintenance_costs

        # Benefits
        defects_prevented = self._estimate_defects_prevented(period_days)
        avg_defect_cost = self._get_average_defect_fix_cost()
        cost_avoided = defects_prevented * avg_defect_cost

        deployment_efficiency_gain = self._calculate_deployment_efficiency_gain(period_days)
        time_to_market_improvement = self._calculate_time_to_market_improvement(period_days)

        total_value = cost_avoided + deployment_efficiency_gain + time_to_market_improvement

        roi_percentage = ((total_value - total_testing_cost) / total_testing_cost) * 100

        return {
            'period_days': period_days,
            'costs': {
                'test_automation': test_automation_costs,
                'test_execution': test_execution_costs,
                'test_maintenance': test_maintenance_costs,
                'total': total_testing_cost
            },
            'benefits': {
                'defects_prevented': defects_prevented,
                'cost_per_defect': avg_defect_cost,
                'defect_cost_avoided': cost_avoided,
                'deployment_efficiency_gain': deployment_efficiency_gain,
                'time_to_market_improvement': time_to_market_improvement,
                'total_value': total_value
            },
            'roi_percentage': roi_percentage,
            'payback_period_months': self._calculate_payback_period(total_testing_cost, total_value)
        }

# grafana/qa-metrics-dashboard.json
{
  "dashboard": {
    "title": "QA DevOps Metrics Dashboard",
    "tags": ["qa", "devops", "metrics"],
    "timezone": "browser",
    "panels": [
      {
        "id": 1,
        "title": "Deployment Frequency & Success Rate",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(deployments_total[24h])",
            "legendFormat": "Deployment Frequency"
          },
          {
            "expr": "(sum(deployments_total{status=\"success\"}) / sum(deployments_total)) * 100",
            "legendFormat": "Success Rate %"
          }
        ],
        "yaxes": [
          {
            "label": "Deployments/day",
            "format": "short"
          },
          {
            "label": "Success Rate %",
            "format": "percent"
          }
        ]
      },
      {
        "id": 2,
        "title": "Test Execution Metrics",
        "type": "stat",
        "targets": [
          {
            "expr": "sum(test_runs_total)",
            "legendFormat": "Total Test Runs"
          },
          {
            "expr": "sum(test_passed) / sum(test_runs_total) * 100",
            "legendFormat": "Pass Rate %"
          },
          {
            "expr": "avg(test_duration_seconds)",
            "legendFormat": "Avg Duration (s)"
          }
        ]
      },
      {
        "id": 3,
        "title": "Flaky Tests Over Time",
        "type": "graph",
        "targets": [
          {
            "expr": "flaky_tests_count",
            "legendFormat": "Flaky Tests"
          },
          {
            "expr": "flaky_tests_impact_minutes",
            "legendFormat": "CI Time Impact (min)"
          }
        ]
      },
      {
        "id": 4,
        "title": "Test Coverage Trends",
        "type": "graph",
        "targets": [
          {
            "expr": "test_coverage_line_percent",
            "legendFormat": "Line Coverage"
          },
          {
            "expr": "test_coverage_branch_percent",
            "legendFormat": "Branch Coverage"
          },
          {
            "expr": "test_coverage_mutation_score",
            "legendFormat": "Mutation Score"
          }
        ],
        "thresholds": [
          {
            "value": 80,
            "color": "yellow"
          },
          {
            "value": 90,
            "color": "green"
          }
        ]
      },
      {
        "id": 5,
        "title": "Change Failure Rate by Test Coverage",
        "type": "bargauge",
        "targets": [
          {
            "expr": "change_failure_rate_by_coverage_range",
            "legendFormat": "{{coverage_range}}"
          }
        ]
      },
      {
        "id": 6,
        "title": "Test Execution Time Breakdown",
        "type": "piechart",
        "targets": [
          {
            "expr": "sum(test_duration_seconds) by (test_type)",
            "legendFormat": "{{test_type}}"
          }
        ]
      }
    ],
    "templating": {
      "list": [
        {
          "name": "environment",
          "type": "query",
          "query": "label_values(deployments_total, environment)"
        },
        {
          "name": "time_range",
          "type": "interval",
          "options": ["24h", "7d", "30d", "90d"]
        }
      ]
    }
  }
}

# prometheus/qa-alerts.yml
groups:
  - name: qa_metrics_alerts
    interval: 5m
    rules:
      - alert: HighTestFailureRate
        expr: (sum(test_failed) / sum(test_runs_total)) * 100 > 10
        for: 15m
        labels:
          severity: warning
          team: qa
        annotations:
          summary: "Test failure rate above 10%"
          description: "Current failure rate: {{ $value }}%"

      - alert: FlakyTestsIncreasing
        expr: increase(flaky_tests_count[24h]) > 5
        for: 1h
        labels:
          severity: warning
          team: qa
        annotations:
          summary: "Flaky tests increased by {{ $value }} in last 24h"

      - alert: TestCoverageDropped
        expr: test_coverage_line_percent < 80
        for: 30m
        labels:
          severity: critical
          team: qa
        annotations:
          summary: "Test coverage dropped below 80%"
          description: "Current coverage: {{ $value }}%"

      - alert: DeploymentFailureRateHigh
        expr: (sum(deployments_total{status="failed"}) / sum(deployments_total)) * 100 > 15
        for: 2h
        labels:
          severity: critical
          team: qa
        annotations:
          summary: "Deployment failure rate above 15%"

      - alert: SlowTestExecution
        expr: avg(test_duration_seconds) > 1800
        for: 30m
        labels:
          severity: warning
          team: qa
        annotations:
          summary: "Average test execution time exceeds 30 minutes"