# Flaky Test Prevention Guide

**Purpose**: Comprehensive guide for preventing, detecting, and fixing flaky tests.

**Last Updated**: 2026-02-11

---

## Overview

**Flaky Test**: A test that passes or fails non-deterministically without any code changes.

**Impact**:
- **Erodes confidence**: Developers ignore test failures
- **Wastes time**: Debugging non-existent bugs
- **Masks real failures**: Legitimate failures hidden among flakes
- **Slows development**: Re-running tests increases cycle time

**Common Symptoms**:
- Passes locally, fails in CI
- Passes alone, fails in suite
- Passes in suite, fails alone
- Intermittent failures (passes 9/10 times)

---

## What Are Flaky Tests

### Definition

**Flaky Test**: A test with non-deterministic outcome due to factors other than code changes.

**Non-Flaky Test**: Deterministic outcome (always passes or always fails for same code).

**Examples**:
```python
# FLAKY: Depends on timing
def test_async_operation():
    result = async_operation()
    time.sleep(0.1)  # May not be enough
    assert result.is_ready

# FLAKY: Depends on execution order
def test_user_count():
    assert User.query.count() == 0  # Fails if other test created users

# FLAKY: Depends on external service
def test_api_call():
    response = requests.get("https://api.example.com/data")
    assert response.status_code == 200  # Network may be slow/down
```

---

### Common Symptoms

| Symptom | Likely Cause | Fix |
|---------|--------------|-----|
| Passes locally, fails in CI | Environment differences (speed, resources, timezone) | Use mocks, increase timeouts |
| Passes alone, fails in suite | Shared state between tests | Use unique_resource_name, db_session |
| Passes in suite, fails alone | Depends on other tests' setup | Make tests independent |
| Intermittent failures | Race conditions, resource contention | Synchronize, use explicit waits |
| Fails in parallel, passes sequentially | Resource conflicts (ports, files) | Use unique resource names |
| Random failures | Time dependencies, randomness | Mock time, seed random generator |

---

### Impact

**Wasted Time**:
- Developer: 30 minutes debugging flaky test
- Team: 10 developers × 30 minutes = 5 hours/week
- CI: Re-running tests = slower builds

**Eroded Confidence**:
```
Scenario: Test fails in CI
Developer A: "Probably just flaky, ignore it"
Developer B: "No, that's a real bug!"
Result: Legitimate failures ignored
```

**Masked Real Failures**:
```
Build #123: FAILED (flaky)
Build #124: FAILED (flaky)
Build #125: FAILED (real bug) ← Hidden by flakes
Build #126: FAILED (flaky)
```

**Slowed Development**:
- Re-run tests 3-4 times to get "green" build
- Disable tests to unblock CI
- Merge without full test suite

---

## Common Causes

### 1. Race Conditions

**What**: Tests depend on timing between concurrent operations.

**Example**:
```python
# FLAKY: Race condition
def test_async_processing():
    queue.add_task("process_data")
    # May execute before task completes
    assert queue.get_result("process_data") is not None
```

**Root Cause**: No guarantee task completes before assertion.

**Fix**: Use explicit synchronization (events, barriers).

```python
# GOOD: Explicit synchronization
def test_async_processing():
    queue.add_task("process_data")
    queue.wait_for_completion("process_data", timeout=5.0)
    assert queue.get_result("process_data") is not None
```

---

### 2. Shared State

**What**: Tests modify global variables, database rows, or file system state.

**Example**:
```python
# FLAKY: Shared database state
def test_create_user():
    user = User(id=1, name="Alice")  # Hardcoded ID
    db.add(user)
    db.commit()

def test_update_user():
    user = db.query(User).get(1)  # Fails if test_create_user didn't run
    user.name = "Bob"
    db.commit()
```

**Root Cause**: Tests depend on execution order.

**Fix**: Use db_session with automatic rollback.

```python
# GOOD: Isolated database state
def test_create_user(db_session, unique_resource_name):
    user = UserFactory.create(_session=db_session, id=unique_resource_name)
    # Automatic rollback, no shared state
```

---

### 3. External Dependencies

**What**: Tests call external APIs, databases, or services.

**Example**:
```python
# FLAKY: External API call
def test_payment_processing():
    result = payment_gateway.charge(amount=100.00)
    assert result["status"] == "success"  # API may be down/slow
```

**Root Cause**: Network issues, service downtime, rate limits.

**Fix**: Mock external dependencies.

```python
# GOOD: Mocked API
@patch('core.services.payment_gateway.charge')
def test_payment_processing(mock_charge):
    mock_charge.return_value = {"status": "success"}
    result = payment_gateway.charge(amount=100.00)
    assert result["status"] == "success"
```

---

### 4. Time Dependencies

**What**: Tests depend on current time or use time.sleep().

**Example**:
```python
# FLAKY: Time-based assertion
def test_token_expiry():
    token = create_token(expiry_hours=24)
    assert token.expires_at > datetime.now()  # Flaky near boundary

# FLAKY: Uses sleep (non-deterministic)
def test_cache_timeout():
    cache.set("key", "value", timeout=1)
    time.sleep(1)  # May not be enough (slow CI)
    assert cache.get("key") is None
```

**Root Cause**: Time passes non-deterministically.

**Fix**: Use freezegun to mock time.

```python
# GOOD: Frozen time
from freezegun import freeze_time

def test_token_expiry():
    with freeze_time("2026-02-11 10:00:00"):
        token = create_token(expiry_hours=24)
        assert token.expires_at == datetime(2026, 2, 12, 10, 0, 0)

def test_cache_timeout():
    with freeze_time("2026-02-11 10:00:00"):
        cache.set("key", "value", timeout=3600)
    with freeze_time("2026-02-11 11:00:00"):
        assert cache.get("key") is None  # Instant, no sleep
```

---

### 5. Resource Contention

**What**: Tests compete for limited resources (ports, files, memory).

**Example**:
```python
# FLAKY: Port already in use
def test_server_start():
    server = start_server(port=8000)  # Fails if another test using port

# FLAKY: File already exists
def test_log_processing():
    with open("test.log", 'w') as f:  # Fails if another test created file
        f.write("log data")
```

**Root Cause**: Hardcoded resource names/ports.

**Fix**: Use unique resource names or ephemeral ports.

```python
# GOOD: Unique filename
def test_log_processing(unique_resource_name):
    filename = f"{unique_resource_name}.log"
    with open(filename, 'w') as f:
        f.write("log data")

# GOOD: Ephemeral port
def test_server_start():
    port = get_ephemeral_port()  # Get free port
    server = start_server(port=port)
```

---

### 6. Order Dependency

**What**: Tests assume specific execution order.

**Example**:
```python
# FLAKY: Depends on test_setup_data running first
def test_process_data():
    data = db.query(Data).first()  # None if test_setup_data didn't run
    assert process(data) == "result"

# FLAKY: Depends on global variable being set
def test_feature_flag():
    assert SETTINGS["feature_enabled"] == True  # Fails if previous test didn't set
```

**Root Cause**: Implicit dependencies between tests.

**Fix**: Make tests independent (use fixtures for setup).

```python
# GOOD: Explicit setup
def test_process_data(unique_resource_name):
    data = DataFactory.create(_session=db_session, id=unique_resource_name)
    assert process(data) == "result"

# GOOD: Fixture resets state
@pytest.fixture(autouse=True)
def reset_settings():
    original = SETTINGS.copy()
    yield
    SETTINGS.clear()
    SETTINGS.update(original)

def test_feature_flag():
    SETTINGS["feature_enabled"] = True  # Isolated
```

---

## Prevention Patterns

### Pattern 1: Explicit Synchronization

**Instead of sleep**: Use events, barriers, or condition variables.

**Bad**:
```python
# FLAKY: Arbitrary sleep
def test_worker_pool():
    pool = WorkerPool(size=2)
    pool.submit(task)
    time.sleep(0.5)  # Hope task completes
    assert pool.results[0] == "done"
```

**Good**:
```python
# GOOD: Explicit wait
def test_worker_pool():
    pool = WorkerPool(size=2)
    future = pool.submit(task)
    result = future.wait(timeout=5.0)  # Explicit synchronization
    assert result == "done"
```

---

### Pattern 2: Mock External Dependencies

**Instead of real API**: Use mocks, fixtures, or test doubles.

**Bad**:
```python
# FLAKY: Real API call
def test_send_email():
    result = email_service.send("user@example.com", "Hello")
    assert result["status"] == "sent"  # API may be down
```

**Good**:
```python
# GOOD: Mocked API
@patch('core.services.email_service.send')
def test_send_email(mock_send):
    mock_send.return_value = {"status": "sent"}
    result = email_service.send("user@example.com", "Hello")
    assert result["status"] == "sent"
```

---

### Pattern 3: Unique Resource Names

**Instead of hardcoded names**: Use unique_resource_name fixture.

**Bad**:
```python
# FLAKY: Hardcoded ID
def test_create_user():
    user = User(id="test-user", ...)  # Collision in parallel
```

**Good**:
```python
# GOOD: Unique ID
def test_create_user(unique_resource_name):
    user = UserFactory.create(id=unique_resource_name, ...)
```

---

### Pattern 4: Transaction Rollback

**Instead of manual cleanup**: Use db_session with rollback.

**Bad**:
```python
# FLAKY: Manual cleanup (may not run)
def test_create_agent():
    agent = Agent(id="test-agent", ...)
    db.add(agent)
    db.commit()
    try:
        # Test logic...
        pass
    finally:
        db.delete(agent)  # May not run
        db.commit()
```

**Good**:
```python
# GOOD: Automatic rollback
def test_create_agent(db_session, unique_resource_name):
    agent = AgentFactory.create(_session=db_session, id=unique_resource_name)
    # Test logic...
    # Automatic rollback via fixture
```

---

### Pattern 5: Avoid Global State

**Instead of globals**: Use fixtures to reset state.

**Bad**:
```python
# FLAKY: Global state
SETTINGS = {"debug": False}

def test_enable_debug():
    global SETTINGS
    SETTINGS["debug"] = True  # Affects other tests
```

**Good**:
```python
# GOOD: Fixture with cleanup
@pytest.fixture(autouse=True)
def reset_settings():
    original = SETTINGS.copy()
    yield
    SETTINGS.clear()
    SETTINGS.update(original)

def test_enable_debug():
    SETTINGS["debug"] = True  # Isolated
```

---

### Pattern 6: Make Tests Order-Independent

**Instead of order dependency**: Each test sets up its own data.

**Bad**:
```python
# FLAKY: Depends on test_create_user running first
def test_update_user():
    user = db.query(User).first()  # None if test_create_user didn't run
    user.name = "Bob"
    db.commit()
```

**Good**:
```python
# GOOD: Independent test
def test_update_user(unique_resource_name):
    user = UserFactory.create(id=unique_resource_name, name="Alice")
    user.name = "Bob"
    db.commit()
    assert user.name == "Bob"
```

---

## Detection Strategies

### Strategy 1: pytest-rerunfailures

**Purpose**: Automatically retry failed tests to detect flakiness.

**Installation**:
```bash
pip install pytest-rerunfailures
```

**Usage**:
```bash
# Retry all failed tests 3 times
pytest tests/ --reruns 3

# Retry with delay between retries
pytest tests/ --reruns 3 --reruns-delay 1

# Retry only specific tests
pytest tests/ --reruns 3 -m "flaky"

# Run without retries (to detect real failures)
pytest tests/ --reruns 0
```

**Marker Usage**:
```python
import pytest

@pytest.mark.flaky(reruns=3, reruns_delay=1)
def test_external_api():
    # May fail intermittently, will retry up to 3 times
    result = external_api_call()
    assert result["status"] == "success"
```

**When to Use**: Temporary workaround while fixing root cause (not permanent).

---

### Strategy 2: Run Tests Multiple Times

**Purpose**: Detect intermittent failures by running tests repeatedly.

**Bash Script**:
```bash
# Run tests 10 times, count failures
FAILURES=0
for i in {1..10}; do
    pytest tests/test_agent.py -v --tb=short
    if [ $? -ne 0 ]; then
        FAILURES=$((FAILURES + 1))
    fi
done

echo "Failures: $FAILURES/10"

# If FAILURES > 0 and < 10 → Flaky test
```

**Python Script**:
```python
import subprocess

failures = 0
runs = 10

for i in range(runs):
    result = subprocess.run(["pytest", "tests/test_agent.py", "-v"])
    if result.returncode != 0:
        failures += 1

if 0 < failures < runs:
    print(f"FLAKY TEST: {failures}/{runs} runs failed")
elif failures == runs:
    print("REAL FAILURE: All runs failed")
else:
    print("STABLE: All runs passed")
```

---

### Strategy 3: Run Tests in Parallel

**Purpose**: Reveal resource conflicts and shared state issues.

**Sequential (may pass)**:
```bash
pytest tests/ -v
# All tests pass (no resource conflicts)
```

**Parallel (may fail)**:
```bash
pytest tests/ -n auto -v
# Some tests fail (resource conflicts, shared state)
```

**Diagnosis**: Tests that fail in parallel but pass sequentially have isolation issues.

---

### Strategy 4: Run Tests in Random Order

**Purpose**: Detect order dependencies.

**Installation**:
```bash
pip install pytest-randomly
```

**Usage**:
```bash
pytest tests/ -v
# Tests run in random order each time

# If results vary → order dependency
```

**Seeded Random Order** (reproducible):
```bash
pytest tests/ -v --randomly-seed=1234
# Same random order each time with seed=1234
```

---

## Fixing Flaky Tests

### Step 1: Identify the Cause

**Add Logging**:
```python
def test_async_operation():
    result = async_operation()
    print(f"Result status: {result.status}")  # Debug output
    print(f"Result data: {result.data}")
    assert result.is_ready
```

**Run in Isolation**:
```bash
# Run test alone
pytest tests/test_agent.py::test_async_operation -v -s

# Run with pdb on failure
pytest tests/test_agent.py::test_async_operation -v --pdb
```

**Run Multiple Times**:
```bash
# Run 100 times to see intermittent pattern
for i in {1..100}; do
    pytest tests/test_agent.py::test_async_operation -v
done
```

---

### Step 2: Fix the Root Cause

**Race Condition**: Add synchronization (events, barriers).
```python
# Before: Flaky
def test_async_processing():
    queue.add_task("process_data")
    assert queue.get_result("process_data") is not None

# After: Fixed
def test_async_processing():
    queue.add_task("process_data")
    queue.wait_for_completion("process_data", timeout=5.0)
    assert queue.get_result("process_data") is not None
```

**Shared State**: Use db_session or unique_resource_name.
```python
# Before: Flaky
def test_create_user():
    user = User(id="test-user", ...)
    db.commit()

# After: Fixed
def test_create_user(db_session, unique_resource_name):
    user = UserFactory.create(_session=db_session, id=unique_resource_name)
```

**External Dependency**: Mock the dependency.
```python
# Before: Flaky
def test_api_call():
    result = requests.get("https://api.example.com/data")
    assert result.status_code == 200

# After: Fixed
@patch('requests.get')
def test_api_call(mock_get):
    mock_get.return_value.status_code = 200
    result = requests.get("https://api.example.com/data")
    assert result.status_code == 200
```

**Time Dependency**: Mock time with freezegun.
```python
# Before: Flaky
def test_token_expiry():
    token = create_token(expiry_hours=24)
    assert token.expires_at > datetime.now()

# After: Fixed
def test_token_expiry():
    with freeze_time("2026-02-11 10:00:00"):
        token = create_token(expiry_hours=24)
        assert token.expires_at == datetime(2026, 2, 12, 10, 0, 0)
```

---

### Step 3: Verify Fix

**Run 100 Times**:
```bash
for i in {1..100}; do
    pytest tests/test_agent.py::test_async_operation -v
done

# Should pass 100/100 times
```

**Run in Parallel**:
```bash
pytest tests/ -n auto -v

# Should pass in parallel too
```

**Run in Random Order**:
```bash
pytest tests/ -v  # With pytest-randomly

# Should pass regardless of order
```

---

### Step 4: Document Fix

**Add Comment**:
```python
def test_async_operation():
    """
    Test async operation completion.

    Note: Previously flaky due to race condition. Fixed by adding
    explicit wait_for_completion() instead of time.sleep().

    Bug: https://github.com/example/project/issues/123
    Fix: commit abc123 (2026-02-11)
    """
    queue.add_task("process_data")
    queue.wait_for_completion("process_data", timeout=5.0)
    assert queue.get_result("process_data") is not None
```

**Add to Changelog**:
```markdown
## Fixes
- Fix flaky test_async_operation by adding explicit synchronization
- Fix flaky test_create_user by using db_session fixture
```

---

## When to Use @pytest.mark.flaky

### Temporary Workaround

**Use Case**: Known flaky test while investigating root cause.

**Example**:
```python
import pytest

@pytest.mark.flaky(reruns=3, reruns_delay=1)
def test_external_api_integration():
    """
    Test external API integration.

    TODO: Fix flakiness (currently 10% failure rate)
    Issue: https://github.com/example/project/issues/456
    Root cause: API rate limiting
    Proposed fix: Add retry logic with exponential backoff
    """
    result = external_api_call()
    assert result["status"] == "success"
```

**Requirements**:
1. Must include TODO comment
2. Must link to GitHub issue
3. Must document root cause (if known)
4. Must NOT be permanent solution

---

### Known Issues Filed as Bugs

**Use Case**: Bug in production code, not test code.

**Example**:
```python
@pytest.mark.flaky(reruns=3, reruns_delay=1)
def test_database_transaction():
    """
    Test database transaction rollback.

    FLAKY: Known bug in transaction handling (issue #789)
    Root cause: Race condition in connection pool
    Workaround: Retry test up to 3 times
    """
    with db.transaction():
        db.execute("INSERT INTO users (name) VALUES ('Alice')")
    # Bug: Transaction sometimes doesn't rollback
```

**Action**: Fix bug in production code, remove @flaky marker.

---

### NEVER a Permanent Solution

**Anti-Pattern**: Using @flaky to mask real issues.

**Bad**:
```python
# BAD: Permanent workaround (don't do this)
@pytest.mark.flaky(reruns=10)
def test_critical_payment_flow():
    result = process_payment(amount=100.00)
    assert result["status"] == "success"
```

**Why Bad**:
- Hides real bugs (payment failures!)
- Erodes confidence in test suite
- Slows CI (re-running flaky tests)
- Developers ignore failures

**Correct Approach**:
```python
# GOOD: Fix root cause
@patch('payment_gateway.charge')
def test_critical_payment_flow(mock_charge):
    mock_charge.return_value = {"status": "success"}
    result = process_payment(amount=100.00)
    assert result["status"] == "success"
```

---

## Real Examples from Codebase

### Example 1: Database Row Collision

**Flaky Test**:
```python
# FLAKY: Hardcoded ID causes collision
def test_create_agent():
    agent = AgentRegistry(id="test-agent", name="Test")
    db.add(agent)
    db.commit()
```

**Fixed**:
```python
# GOOD: Unique ID
def test_create_agent(unique_resource_name):
    agent = AgentFactory.create(id=unique_resource_name)
```

---

### Example 2: Race Condition in Async Test

**Flaky Test**:
```python
# FLAKY: Race condition
def test_episode_retrieval():
    episode = create_episode()
    results = search_service.search(episode.id)
    # May not be indexed yet
    assert episode.id in results
```

**Fixed**:
```python
# GOOD: Explicit wait
def test_episode_retrieval():
    episode = create_episode()
    search_service.wait_for_indexing(episode.id, timeout=5.0)
    results = search_service.search(episode.id)
    assert episode.id in results
```

---

### Example 3: Time-Dependent Token Expiry

**Flaky Test**:
```python
# FLAKY: Time-dependent
def test_token_expires_after_24_hours():
    token = create_token(expiry_hours=24)
    assert token.expires_at > datetime.now()  # Flaky near boundary
```

**Fixed**:
```python
# GOOD: Frozen time
def test_token_expires_after_24_hours():
    with freeze_time("2026-02-11 10:00:00"):
        token = create_token(expiry_hours=24)
        assert token.expires_at == datetime(2026, 2, 12, 10, 0, 0)
```

---

## Related Documentation

- **[TEST_ISOLATION_PATTERNS.md](./TEST_ISOLATION_PATTERNS.md)** - Test isolation patterns
- **[COVERAGE_GUIDE.md](./COVERAGE_GUIDE.md)** - Coverage interpretation
- **[../conftest.py](../conftest.py)** - Fixture definitions

---

## Summary

**Key Takeaways**:
1. **Flaky tests erode confidence**: Developers ignore failures, masking real bugs
2. **Root causes**: Race conditions, shared state, external deps, time, resources, order
3. **Prevention**: Explicit sync, mocks, unique names, rollback, no globals
4. **Detection**: pytest-rerunfailures, run 100x, parallel, random order
5. **Fixing**: Identify cause → Fix root → Verify 100x → Document fix

**Quick Reference**:
```python
# BAD: Flaky test
def test_async_operation():
    result = async_operation()
    time.sleep(0.1)  # Arbitrary
    assert result.is_ready

# GOOD: Fixed test
def test_async_operation():
    result = async_operation()
    result.wait_for_completion(timeout=5.0)  # Explicit
    assert result.is_ready
```

**Next Steps**:
1. Run tests 10 times to detect flakiness
2. Use pytest-rerunfailures to identify intermittent failures
3. Fix root causes (don't just add retries)
4. Verify fixes by running 100 times
5. Document fixes with comments and issue links