6. Flaky Test Management

Description

A flaky test passes sometimes and fails other times without changing the code. In AI apps, LLM variance is the most common cause. This capsule covers the four strategies for handling flakiness: retry logic, tolerance thresholds, quarantine, and relaxing assertions. Most importantly: how to differentiate between flakiness from legitimate LLM variance (which you handle) and flakiness from a poorly written test or a real bug (which you fix). Normalizing flakiness destroys confidence in the test suite.

The problem with normalized flaky testing

Week 1: Test fails → "It's the LLM, it's normal" → ignored
Week 2: Test fails → "The LLM again" → ignored
Week 3: Real bug → the test fails → "Surely it's the LLM" → ignored
Week 4: The bug reaches production

Normalizing flakiness kills the value of the test suite. When developers learn to ignore failing tests, all failures get ignored — including the ones that indicate real bugs.

The four causes of flakiness in AI apps

Cause 1: LLM variance (legitimate)

# The LLM can respond in slightly different ways:
result_1 = analyze_sentiment("Positive text")
# → "positive" with score 0.92

result_2 = analyze_sentiment("Positive text")
# → "positive" with score 0.88

# If the test does: assert result["score"] == 0.92 → fails 50% of the time
# Solution: assertion over a range, not an exact value

Cause 2: Poorly written test (fix, don't tolerate)

# Assertion too strict for non-deterministic output:
def test_bad():
    result = analyze_sentiment("I love it")
    assert result["explanation"] == "The text uses positive language."
    # Fails because the LLM might say "The text expresses positivity" or
    # "The text has a positive emotional charge" — all are correct
    # → FIX: use a semantic or property-based assertion

Cause 3: Infrastructure (fix)

# API timeout, rate limit, connectivity:
def test_api_timeout():
    result = analyze_sentiment("text")  # May fail due to a timeout
    assert result["sentiment"] == "positive"
    # → FIX: add an explicit timeout, retry with backoff, conditional skip

Cause 4: Shared state between tests (fix)

# Global cache that affects results:
_cache = {}

def analyze_with_cache(text):
    if text not in _cache:
        _cache[text] = analyze_sentiment(text)
    return _cache[text]

def test_a():
    result = analyze_with_cache("text x")
    # Modifies _cache

def test_b():
    # _cache["text x"] may have the result from test_a
    result = analyze_with_cache("text x")
    # → FIX: clear shared state between tests, or use fresh state

Strategy 1: Retry logic

For flakiness from LLM variance or transient API problems:

pip install pytest-rerunfailures

# Option A: per individual test
import pytest

@pytest.mark.integration
@pytest.mark.flaky(reruns=3, reruns_delay=2)
def test_sentiment_quality_with_retry():
    """
    Semantic quality test — may fail due to LLM variance.
    Retries up to 3 times with a 2s delay.
    """
    result = analyze_sentiment("I love this product")
    
    # This assertion can vary — that's why it has a retry
    assert result["score"] >= 0.7, \
        f"For clearly positive text, score must be >= 0.7. Got: {result['score']}"

# Option B: global configuration in pytest.ini
# [pytest]
# addopts = --reruns 2 --reruns-delay 1
# Only affects tests marked with @pytest.mark.flaky

# Option C: only for integration tests
# pytest.ini:
# [pytest]
# addopts = -m integration --reruns 2 --reruns-delay 1

Retry with exponential backoff

For APIs with rate limits, a fixed delay may not be enough:

import time
import openai
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=1.0):
    """Decorator for retry with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except openai.RateLimitError:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt)
                    time.sleep(delay)
                except openai.APITimeoutError:
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(base_delay)
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=1.0)
def analyze_sentiment_with_retry(text, client):
    return analyze_sentiment(text, client=client)

# Test that uses the function with retry:
@pytest.mark.integration
def test_with_api_retry(integration_client):
    result = analyze_sentiment_with_retry("text", integration_client)
    assert result["sentiment"] in ["positive", "negative", "neutral"]

Strategy 2: Tolerance thresholds

For tests that validate statistical properties — they expect to pass most of the time, not necessarily every time:

def run_with_tolerance(func, n_runs: int = 5, min_pass: int = 4):
    """
    Runs func n_runs times and verifies that at least min_pass pass.
    
    Args:
        func: The function to run (returns True if it passes, False if it fails)
        n_runs: Total number of runs
        min_pass: Minimum number of runs that must pass
    """
    results = []
    for i in range(n_runs):
        try:
            passed = func()
            results.append(passed if passed is not None else True)
        except AssertionError:
            results.append(False)
    
    n_passed = sum(results)
    assert n_passed >= min_pass, (
        f"Only {n_passed}/{n_runs} runs passed (minimum required: {min_pass})\n"
        f"Results: {results}"
    )
    return n_passed

# Usage example:
@pytest.mark.integration
def test_sentiment_quality_with_tolerance(integration_client):
    """
    IMPORTANT: This test uses tolerance because the score can vary.
    It accepts that 4 out of 5 runs produce a score >= 0.7 for clearly positive text.
    """
    def run_once():
        result = analyze_sentiment(
            "I absolutely love everything, it's incredible",
            client=integration_client
        )
        return result["score"] >= 0.7
    
    # Accepts up to 1 failure in 5 runs
    run_with_tolerance(run_once, n_runs=5, min_pass=4)

# Version with a direct assertion (simpler):
@pytest.mark.integration
def test_sentiment_passes_majority():
    """4 out of 5 analyses of the same text must return 'positive'."""
    text = "Perfect, I love it, it's wonderful"
    scores = []
    
    for _ in range(5):
        result = analyze_sentiment(text)
        scores.append(result["sentiment"] == "positive")
    
    n_positive = sum(scores)
    assert n_positive >= 4, f"Only {n_positive}/5 classified as positive"

Strategy 3: Quarantine

For unstable tests that you can't fix immediately:

# pytest.ini — register the marker
[pytest]
markers =
    unit: Unit tests, deterministic, no API
    integration: Tests with the real LLM
    flaky: Tests that may fail due to variance — with retry enabled
    quarantine: Unstable tests in quarantine — they don't run in regular CI
    e2e: End-to-end tests of the complete flow
    smoke: Basic smoke tests

# In the test:
@pytest.mark.quarantine(reason="Fails 20% of the time due to score variance")
@pytest.mark.integration
def test_exact_score_value():
    """
    QUARANTINE: This test verifies an exact score that varies.
    It must be fixed to use a range assertion instead of an exact value.
    See issue: #123
    """
    result = analyze_sentiment("positive text")
    assert result["score"] == 0.92  # ← Too strict

# .github/workflows/ci.yml
# Regular CI: exclude quarantine
- name: Unit tests
  run: pytest -m "not integration and not quarantine" -v

# Nightly: include quarantine to monitor
- name: Full test suite including quarantine
  run: pytest -v  # No filter
  if: github.event_name == 'schedule'  # Only in nightly

Quarantine workflow

Test fails repeatedly
       ↓
Analysis: Why does it fail?
  ├── Assertion too strict → Relax the assertion (fix)
  ├── Timing/infrastructure → Add retry or timeout (fix)
  ├── Shared state → Isolate the tests (fix)
  └── Legitimate LLM variance that can't be avoided
       ↓
Add @pytest.mark.quarantine(reason="...", issue="URL")
       ↓
Create an issue to fix or remove the test
       ↓
In nightly: monitor whether it improves
       ↓
When it's fixed: remove quarantine

Strategy 4: Relax assertions

The most effective and most ignored solution: fix the assertion, not the test:

# ─── BEFORE (fragile) ─────────────────────────────────────────────
def test_sentiment_fragile():
    result = analyze_sentiment("This product is excellent")
    # ❌ Too specific — the LLM can give 0.88, 0.91, 0.95, etc.
    assert result["score"] == 0.92
    # ❌ The explanation can vary on each run
    assert result["explanation"] == "The text uses strong positive adjectives."
    # ❌ Keywords can vary in order or selection
    assert result["keywords"] == ["excellent", "product"]

# ─── AFTER (robust) ───────────────────────────────────────────────
def test_sentiment_robust(make_sentiment_client):
    # For a contract test → use a mock (M2)
    client = make_sentiment_client(
        sentiment="positive",
        score=0.92,
        keywords=["excellent"]
    )
    result = analyze_sentiment("This product is excellent", client=client)
    
    # ✅ Range instead of exact value
    assert 0.7 <= result["score"] <= 1.0
    # ✅ Property instead of exact text
    assert isinstance(result["explanation"], str) and len(result["explanation"]) > 0
    # ✅ Subset instead of exact list
    assert any(kw in result["keywords"] for kw in ["excellent", "product", "positive"])

# ─── FOR A REAL QUALITY TEST (with the real LLM) ─────────────────
@pytest.mark.integration
def test_sentiment_quality(integration_client):
    result = analyze_sentiment("This product is excellent", client=integration_client)
    
    # ✅ Only robust assertions for the real LLM
    assert result["sentiment"] == "positive"  # This one should be deterministic
    assert result["score"] >= 0.6  # Range, not an exact value
    assert len(result.get("explanation", "")) > 10  # Has some explanation

Detecting and monitoring flakiness

# Script to detect flaky tests by running multiple times:
# run_flaky_detector.sh
#!/bin/bash
FAIL_COUNT=0
TOTAL_RUNS=10

for i in $(seq 1 $TOTAL_RUNS); do
    if ! pytest tests/integration/ -q --tb=no 2>/dev/null; then
        FAIL_COUNT=$((FAIL_COUNT + 1))
    fi
done

echo "Failures: $FAIL_COUNT / $TOTAL_RUNS runs"
if [ $FAIL_COUNT -gt 2 ]; then
    echo "⚠️  There are flaky tests (more than 20% failure rate)"
    exit 1
fi

# In pytest: use pytest-repeat to detect flakiness
# pip install pytest-repeat

# Run each test 5 times:
# pytest --count=5 tests/integration/ -v

# If any test fails on any of the 5 runs, it's reported as failed

When to apply each strategy

The golden rule: never ignore a failure without investigating

# ❌ What a team that normalized flakiness does:
@pytest.mark.skip(reason="Always fails, I don't know why")
def test_important_behavior():
    ...

# ❌ What a resigned team does:
@pytest.mark.flaky(reruns=10)  # 10 retries because nobody investigated
def test_something():
    ...

# ✅ What a team that manages flakiness well does:
@pytest.mark.quarantine(
    reason="Fails ~15% of the time. Score varies 0.65-0.85 for this input. "
           "Investigating whether it's model variance or a bug in the prompt. "
           "Issue: github.com/repo/issues/456"
)
@pytest.mark.integration
def test_specific_score():
    """Test in quarantine while the intermittent failure is investigated."""
    result = analyze_sentiment("ambiguous text")
    assert result["score"] > 0.75  # Fails ~15% of the time

Exercises

Exercise 1: Diagnose the type of flakiness

For each case, identify the cause and the strategy:

1. Test that verifies result["explanation"] == "The text is positive" — fails 40% of the time
2. Test that fails when run after test_b but passes alone
3. Test that fails with APITimeoutError during traffic peaks
4. Test that verifies result["sentiment"] == "positive" — fails 5% of the time

Exercise 2: Implement tolerance

Write a test that verifies that analyze_sentiment correctly classifies positive texts "most of the time" (4 out of 5 runs):

@pytest.mark.integration
@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="Requires an API key")
def test_positive_classification_with_tolerance(integration_client):
    """
    For clearly positive text, the model must classify it as 'positive'
    at least 4 out of 5 times.
    """
    text = "I love this product, it's incredible, I totally recommend it"
    classifications = []
    
    for _ in range(5):
        result = analyze_sentiment(text, client=integration_client)
        classifications.append(result["sentiment"])
    
    n_positive = classifications.count("positive")
    
    assert n_positive >= 4, (
        f"Only {n_positive}/5 classifications were 'positive'.\n"
        f"Classifications: {classifications}"
    )

Exercise 3: Complete quarantine workflow

A test in your suite fails approximately 25% of the time. Describe the complete process you would follow to manage it:

Exercise 4: Fix this test

The following test fails ~30% of the time. Fix it so it's stable:

@pytest.mark.integration
def test_negative_sentiment():
    result = analyze_sentiment("This product is horrible, I regret buying it")
    assert result["sentiment"] == "negative"
    assert result["score"] == 0.05
    assert result["explanation"] == "The text expresses very strong dissatisfaction."
    assert result["keywords"] == ["horrible", "regret"]

# Fixed version: robust assertions
@pytest.mark.integration
@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="Requires an API key")
def test_negative_sentiment_robust(integration_client):
    """
    Fixed test: uses robust assertions for the real LLM.
    The 'negative' sentiment is deterministic for such explicit text.
    Score and explanation wording vary — we don't verify them exactly.
    """
    result = analyze_sentiment(
        "This product is horrible, I regret buying it",
        client=integration_client
    )
    
    # ✅ This one is deterministic for such clear text:
    assert result["sentiment"] == "negative"
    
    # ✅ Range instead of exact value:
    assert result["score"] <= 0.3, \
        f"For very negative text, score must be low (<=0.3), it is {result['score']}"
    
    # ✅ Property instead of exact text:
    assert isinstance(result["explanation"], str)
    assert len(result["explanation"]) > 5
    
    # ✅ Subset instead of exact list:
    assert any(kw in result["keywords"] for kw in ["horrible", "regret", "bad", "negative"])

Exercise 5: Team flakiness policy

Write a 5-point policy for your team on how to handle flaky tests:

## Flaky Tests Policy

1. **No flaky test on the main branch without prior investigation.**
   If a test fails, it must be investigated before merging it.

2. **We will not use `@pytest.mark.skip` without a documented reason.**
   A skipped test without a reason is a test that doesn't work. Always document why and when it will be resolved.

3. **Quarantine is temporary, not permanent.**
   A test in quarantine has an associated issue. If it's not fixed in 2 weeks, it gets removed.

4. **Maximum retry of 3 attempts, with a documented reason.**
   `@pytest.mark.flaky(reruns=3)` requires a comment explaining why the test is inherently variable.

5. **Monitor the failure rate of integration tests in nightly.**
   If >10% of tests fail in nightly, prioritize the fix before new features.

Summary

• Four strategies: retry, tolerance threshold, quarantine, relaxing assertions
• Two types of flakiness: from LLM variance (handle) vs from a poorly written test/bug (fix)
• Never normalize: "AI tests are flaky" is false — they can be managed
• Quarantine is temporary: always with an associated issue and deadline
• The most effective solution: relax the assertion instead of adding retry
• Golden rule: no failure is ignored without investigation

Additional resources

1. pytest-rerunfailures — Retry plugin
2. pytest-flaky — Alternative for marking flaky tests
3. Non-Determinism — Martin Fowler — General strategies for flakiness
4. Dealing with flaky tests at Google — How Google handles it
5. pytest-repeat — To detect flakiness by running tests multiple times
6. Quarantine pattern — The quarantine pattern in detail