Flaky Test Management in CI/CD

# flaky_detector.py
import json
from collections import defaultdict
from datetime import datetime, timedelta

class FlakyDetector:
    def __init__(self, threshold=0.05, lookback_days=14, min_runs=20):
        self.threshold = threshold
        self.lookback_days = lookback_days
        self.min_runs = min_runs
        self.test_history = defaultdict(list)

    def record_test_result(self, test_name, passed, duration, commit_sha, environment):
        """Record a single test result with full context"""
        result = {
            'timestamp': datetime.now().isoformat(),
            'passed': passed,
            'duration': duration,
            'commit': commit_sha,
            'environment': environment
        }
        self.test_history[test_name].append(result)

    def calculate_flakiness_score(self, test_name):
        """Calculate flakiness score based on multiple factors"""
        results = self.get_recent_results(test_name)

        if len(results) < self.min_runs:
            return None  # Insufficient data

        # Factor 1: Pass/fail transitions
        transitions = sum(
            1 for i in range(1, len(results))
            if results[i]['passed'] != results[i-1]['passed']
        )
        transition_score = transitions / len(results)

        # Factor 2: Failure clustering (failures happening together)
        failure_clusters = self.detect_failure_clusters(results)
        cluster_score = len(failure_clusters) / (len(results) / 10)

        # Factor 3: Environment-specific failures
        env_failures = defaultdict(lambda: {'passed': 0, 'failed': 0})
        for r in results:
            if r['passed']:
                env_failures[r['environment']]['passed'] += 1
            else:
                env_failures[r['environment']]['failed'] += 1

        env_score = 0
        for env, counts in env_failures.items():
            total = counts['passed'] + counts['failed']
            if total > 5:  # Minimum samples per environment
                env_failure_rate = counts['failed'] / total
                if 0.1 < env_failure_rate < 0.9:  # Partially failing
                    env_score += 1

        # Combined score
        flakiness_score = (
            transition_score * 0.5 +
            min(cluster_score, 1.0) * 0.3 +
            min(env_score / len(env_failures), 1.0) * 0.2
        )

        return flakiness_score

    def get_recent_results(self, test_name):
        """Get results from the lookback period"""
        cutoff = datetime.now() - timedelta(days=self.lookback_days)
        return [
            r for r in self.test_history[test_name]
            if datetime.fromisoformat(r['timestamp']) > cutoff
        ]

    def detect_failure_clusters(self, results):
        """Identify groups of consecutive failures"""
        clusters = []
        current_cluster = []

        for r in results:
            if not r['passed']:
                current_cluster.append(r)
            elif current_cluster:
                if len(current_cluster) > 1:
                    clusters.append(current_cluster)
                current_cluster = []

        if current_cluster and len(current_cluster) > 1:
            clusters.append(current_cluster)

        return clusters

    def get_flaky_tests(self):
        """Return list of flaky tests with analysis"""
        flaky_tests = []

        for test_name in self.test_history:
            score = self.calculate_flakiness_score(test_name)
            if score and score > self.threshold:
                results = self.get_recent_results(test_name)
                pass_rate = sum(1 for r in results if r['passed']) / len(results)

                flaky_tests.append({
                    'name': test_name,
                    'flakiness_score': score,
                    'pass_rate': pass_rate,
                    'total_runs': len(results),
                    'status': self.categorize_flakiness(score)
                })

        return sorted(flaky_tests, key=lambda x: x['flakiness_score'], reverse=True)

    def categorize_flakiness(self, score):
        """Categorize flakiness severity"""
        if score > 0.3:
            return 'CRITICAL'
        elif score > 0.15:
            return 'HIGH'
        elif score > 0.05:
            return 'MEDIUM'
        return 'LOW'

name: Flaky Test Detection

on:
  schedule:
    - cron: '0 0 * * *'  # Daily analysis
  workflow_dispatch:

jobs:
  analyze-flakiness:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Fetch test history
        run: |
          # Download last 30 days of test results
          gh run list --json databaseId,conclusion,createdAt \
            --limit 1000 | jq -r '.[] | select(.conclusion != null) | .databaseId' \
            | xargs -I {} gh run download {} --name test-results

      - name: Analyze flakiness
        run: python3 scripts/analyze_flakiness.py --output flaky-report.json

      - name: Create issue for critical flaky tests
        run: |
          CRITICAL=$(jq -r '.[] | select(.status == "CRITICAL") | .name' flaky-report.json)
          if [ ! -z "$CRITICAL" ]; then
            gh issue create \
              --title "Critical Flaky Tests Detected" \
              --body "$(cat flaky-report.json)" \
              --label "flaky-test,priority:high"
          fi

      - name: Post comment to recent PRs
        run: python3 scripts/notify_flaky_tests.py

// jest.config.js with quarantine support
module.exports = {
  testMatch: [
    '<rootDir>/tests/**/*.test.js',
    '!<rootDir>/tests/quarantine/**'  // Exclude quarantine
  ],

  // Separate runner for quarantine tests
  projects: [
    {
      displayName: 'stable',
      testMatch: ['<rootDir>/tests/**/*.test.js'],
      testPathIgnorePatterns: ['/quarantine/']
    },
    {
      displayName: 'quarantine',
      testMatch: ['<rootDir>/tests/quarantine/**/*.test.js'],
      // Run with retries
      jest-retries: 3
    }
  ]
};

# pytest quarantine implementation
# conftest.py
import pytest

def pytest_collection_modifyitems(config, items):
    """Mark quarantined tests"""
    for item in items:
        if 'quarantine' in item.nodeid:
            item.add_marker(pytest.mark.skip(reason="Quarantined flaky test"))

# Run quarantined tests separately
# pytest -m quarantine --reruns 5

# .github/workflows/test-with-retry.yml
name: Tests with Smart Retry

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Run tests with retry
        id: test_run
        uses: nick-invision/retry@v2
        with:
          timeout_minutes: 30
          max_attempts: 3
          retry_on: error
          command: npm test -- --json --outputFile=test-results.json

      - name: Analyze retry patterns
        if: always()
        run: |
          python3 << 'EOF'
          import json
          import sys

          with open('test-results.json') as f:
              results = json.load(f)

          attempt = int('${{ steps.test_run.outputs.attempt }}')

          if attempt > 1:
              # This test passed only after retry - mark as flaky
              failed_tests = [t for t in results['testResults']
                            if t['status'] == 'failed']

              if failed_tests:
                  print(f"⚠️ Test passed on attempt {attempt}")
                  print("Likely flaky tests:")
                  for test in failed_tests:
                      print(f"  - {test['fullName']}")

                  # Create incident report
                  with open('flaky-incident.json', 'w') as out:
                      json.dump({
                          'attempt': attempt,
                          'tests': failed_tests,
                          'timestamp': '${{ github.event.head_commit.timestamp }}'
                      }, out)
          EOF

      - name: Report flaky incident
        if: steps.test_run.outputs.attempt > 1
        run: |
          # Post to monitoring system
          curl -X POST $MONITORING_URL \
            -H "Content-Type: application/json" \
            -d @flaky-incident.json

# dashboard_generator.py
from flask import Flask, render_template
import json
from datetime import datetime, timedelta

app = Flask(__name__)

@app.route('/')
def dashboard():
    detector = FlakyDetector()
    detector.load_history('test_history.json')

    flaky_tests = detector.get_flaky_tests()

    # Calculate trends
    trends = calculate_trends(detector)

    # Get worst offenders
    worst_tests = flaky_tests[:10]

    # Calculate team impact
    impact = calculate_team_impact(flaky_tests)

    return render_template('dashboard.html',
                         flaky_tests=flaky_tests,
                         trends=trends,
                         worst_tests=worst_tests,
                         impact=impact)

def calculate_team_impact(flaky_tests):
    """Estimate engineering hours wasted"""
    # Average: 15 min per flaky failure investigation
    failures_per_day = sum(
        (1 - t['pass_rate']) * (t['total_runs'] / 14)  # Last 14 days
        for t in flaky_tests
    )

    hours_per_day = (failures_per_day * 15) / 60
    hours_per_week = hours_per_day * 5
    annual_cost = hours_per_week * 52 * 150  # $150/hour average

    return {
        'failures_per_day': round(failures_per_day, 1),
        'hours_per_week': round(hours_per_week, 1),
        'annual_cost_usd': round(annual_cost)
    }

if __name__ == '__main__':
    app.run(debug=True, port=5000)

# flaky_root_cause.py
import re
from collections import Counter

class RootCauseAnalyzer:
    def __init__(self):
        self.patterns = {
            'timing': [
                r'timeout',
                r'element not found',
                r'stale element',
                r'wait.*exceeded'
            ],
            'race_condition': [
                r'race',
                r'concurrent',
                r'thread.*blocked',
                r'deadlock'
            ],
            'network': [
                r'connection.*refused',
                r'network.*unreachable',
                r'DNS.*fail',
                r'ECONNRESET'
            ],
            'resource': [
                r'out of memory',
                r'port.*in use',
                r'file.*exists',
                r'disk.*full'
            ],
            'external_dependency': [
                r'503.*unavailable',
                r'502.*bad gateway',
                r'429.*too many requests',
                r'API.*down'
            ]
        }

    def analyze(self, test_name, failure_logs):
        """Analyze failure logs to determine root cause"""
        causes = Counter()

        for log in failure_logs:
            error_text = log.get('error', '').lower()
            stack_trace = log.get('stack_trace', '').lower()
            full_text = f"{error_text} {stack_trace}"

            for category, patterns in self.patterns.items():
                for pattern in patterns:
                    if re.search(pattern, full_text, re.IGNORECASE):
                        causes[category] += 1

        if not causes:
            return 'unknown'

        # Return most common cause
        return causes.most_common(1)[0][0]

    def suggest_fix(self, cause):
        """Provide remediation suggestions"""
        fixes = {
            'timing': [
                'Increase wait timeouts',
                'Use explicit waits instead of implicit',
                'Add retry logic with exponential backoff',
                'Check for element visibility, not just presence'
            ],
            'race_condition': [
                'Add proper synchronization primitives',
                'Use locks or semaphores',
                'Review async/await patterns',
                'Ensure proper test isolation'
            ],
            'network': [
                'Mock external network calls',
                'Use service virtualization',
                'Add network resilience patterns',
                'Implement circuit breakers'
            ],
            'resource': [
                'Improve test cleanup',
                'Check for resource leaks',
                'Use unique ports for parallel tests',
                'Implement proper teardown'
            ],
            'external_dependency': [
                'Mock external APIs',
                'Use contract testing',
                'Implement fallback behavior',
                'Add dependency health checks'
            ]
        }

        return fixes.get(cause, ['Manual investigation required'])

// ❌ Bad: Arbitrary sleep
await page.click('#submit');
await sleep(2000);  // Hope it's enough
const result = await page.textContent('#result');

// ✅ Good: Explicit condition waiting
await page.click('#submit');
await page.waitForSelector('#result', {
  state: 'visible',
  timeout: 10000
});
const result = await page.textContent('#result');

// ✅ Even better: Wait for specific state
await page.click('#submit');
await page.waitForFunction(
  () => document.querySelector('#result')?.textContent !== 'Loading...',
  { timeout: 10000 }
);
const result = await page.textContent('#result');

# ❌ Bad: Shared mutable state
class TestUserService:
    user_db = {}  # Shared across tests!

    def test_create_user(self):
        self.user_db['test'] = User('test')
        assert 'test' in self.user_db

    def test_delete_user(self):
        # Depends on previous test running first
        del self.user_db['test']

# ✅ Good: Proper isolation
class TestUserService:
    def setup_method(self):
        self.user_db = {}  # Fresh for each test

    def test_create_user(self):
        self.user_db['test'] = User('test')
        assert 'test' in self.user_db

    def test_delete_user(self):
        # Setup its own state
        self.user_db['test'] = User('test')
        del self.user_db['test']
        assert 'test' not in self.user_db

# ❌ Bad: Non-idempotent test
def test_increment_counter():
    current = get_counter()
    increment_counter()
    assert get_counter() == current + 1  # Fails on retry

# ✅ Good: Idempotent test
def test_increment_counter():
    reset_counter(0)  # Known state
    increment_counter()
    assert get_counter() == 1  # Always true

# chaos_test_decorator.py
import random
from functools import wraps

def chaos_monkey(failure_rate=0.1):
    """Randomly inject failures to test resilience"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if random.random() < failure_rate:
                # Inject random delay
                time.sleep(random.uniform(0.5, 2.0))

            # Occasionally fail test
            if random.random() < failure_rate / 2:
                raise Exception("Chaos Monkey struck!")

            return func(*args, **kwargs)
        return wrapper
    return decorator

# Use to test retry logic
@chaos_monkey(failure_rate=0.2)
def test_with_resilience():
    response = api_call_with_retry(max_attempts=3)
    assert response.status_code == 200

# Features for ML model
features = [
    'test_duration_variance',  # High variance = more flaky
    'dependency_count',        # More deps = more flaky
    'async_operations_count',  # More async = more flaky
    'external_calls_count',    # More external = more flaky
    'recent_modification',     # Recently changed = more flaky
    'historical_flakiness'     # Was flaky before = likely flaky again
]

# Model predicts flakiness probability
# Tests with >50% probability get quarantined preemptively

# team_policy.yml
flaky_test_policy:
  detection_threshold: 0.05  # 5% flakiness
  max_lifetime: 7_days       # Fix or remove within 1 week
  ownership: test_author     # Author responsible for fix
  escalation:
    - day_3: team_lead_notification
    - day_5: test_auto_disabled
    - day_7: test_removed

  consequences:
    - blocks_merge: true     # PR can't merge with flaky tests
    - sprint_priority: high  # Must be fixed in current sprint
    - metrics_tracked: true  # Counts against team KPIs

## Test Review Checklist

- [ ] Uses explicit waits (not arbitrary sleeps)
- [ ] Properly isolated (no shared state)
- [ ] Idempotent (can run multiple times)
- [ ] No hardcoded timeouts < 5 seconds
- [ ] External dependencies mocked
- [ ] Cleanup in teardown, not setup
- [ ] No reliance on test execution order
- [ ] Deterministic assertions (no ranges)

metrics_to_track = {
    'flaky_test_count': 'Number of flaky tests',
    'flakiness_rate': 'Percentage of flaky tests',
    'mean_time_to_fix': 'Average days to fix flaky test',
    'rerun_rate': 'Build rerun rate due to flakiness',
    'engineering_hours_lost': 'Hours spent on flaky tests'
}

# Set targets
targets = {
    'flaky_test_count': {'max': 5, 'target': 0},
    'flakiness_rate': {'max': 0.02, 'target': 0},
    'mean_time_to_fix': {'max': 7, 'target': 3}
}