Spaces:

techprotrade
/

annator-command-center

Running

App Files Files Community

annator-command-center / tests /docs /FLAKY_TEST_GUIDE.md

techprotrade

Deploy ATOM FastAPI command center runtime (part 7)

cc036ff verified 12 days ago

preview code

Raw

History Blame Contribute Delete

22.4 kB

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:

# 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:

# 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).

# 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:

# 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.

# 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:

# 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.

# 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:

# 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.

# 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:

# 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.

# 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:

# 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).

# 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:

# 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:

# 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:

# 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:

# 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:

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

Good:

# 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:

# 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:

# 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:

# FLAKY: Global state
SETTINGS = {"debug": False}

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

Good:

# 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:

# 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:

# 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:

pip install pytest-rerunfailures

Usage:

# 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:

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:

# 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:

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):

pytest tests/ -v
# All tests pass (no resource conflicts)

Parallel (may fail):

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:

pip install pytest-randomly

Usage:

pytest tests/ -v
# Tests run in random order each time

# If results vary → order dependency

Seeded Random Order (reproducible):

pytest tests/ -v --randomly-seed=1234
# Same random order each time with seed=1234

Fixing Flaky Tests

Step 1: Identify the Cause

Add Logging:

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:

# 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:

# 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).

# 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.

# 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.

# 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.

# 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:

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

# Should pass 100/100 times

Run in Parallel:

pytest tests/ -n auto -v

# Should pass in parallel too

Run in Random Order:

pytest tests/ -v  # With pytest-randomly

# Should pass regardless of order

Step 4: Document Fix

Add Comment:

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:

## 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:

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:

Must include TODO comment
Must link to GitHub issue
Must document root cause (if known)
Must NOT be permanent solution

Known Issues Filed as Bugs

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

Example:

@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:

# 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:

# 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:

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

Fixed:

# 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:

# 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:

# 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:

# 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:

# 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)

Summary

Key Takeaways:

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

Quick Reference:

# 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:

Run tests 10 times to detect flakiness
Use pytest-rerunfailures to identify intermittent failures
Fix root causes (don't just add retries)
Verify fixes by running 100 times
Document fixes with comments and issue links

Flaky Test Prevention Guide

Overview

What Are Flaky Tests

Definition

Common Symptoms

Impact

Common Causes

1. Race Conditions

2. Shared State

3. External Dependencies

4. Time Dependencies

5. Resource Contention

6. Order Dependency

Prevention Patterns

Pattern 1: Explicit Synchronization

Pattern 2: Mock External Dependencies

Pattern 3: Unique Resource Names

Pattern 4: Transaction Rollback

Pattern 5: Avoid Global State

Pattern 6: Make Tests Order-Independent

Detection Strategies

Strategy 1: pytest-rerunfailures

Strategy 2: Run Tests Multiple Times

Strategy 3: Run Tests in Parallel

Strategy 4: Run Tests in Random Order

Fixing Flaky Tests

Step 1: Identify the Cause

Step 2: Fix the Root Cause

Step 3: Verify Fix

Step 4: Document Fix

When to Use @pytest.mark.flaky

Temporary Workaround

Known Issues Filed as Bugs

NEVER a Permanent Solution

Real Examples from Codebase

Example 1: Database Row Collision

Example 2: Race Condition in Async Test

Example 3: Time-Dependent Token Expiry

Related Documentation

Summary