tests/docs/TEST_ISOLATION_PATTERNS.md

# Test Isolation Patterns

**Purpose**: Comprehensive guide for writing isolated tests that run reliably in parallel.

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

---

## Overview

**Test isolation** ensures tests don't depend on each other. Isolated tests:
- Run independently (order doesn't matter)
- Run in parallel (no resource conflicts)
- Give consistent results (no false positives/negatives)
- Are easier to debug (failures are self-contained)

**Why It Matters**:
- **False positives**: Tests fail due to other tests' side effects
- **False negatives**: Tests pass only when run in specific order
- **Slow execution**: Sequential execution due to shared state
- **Flaky tests**: Intermittent failures from resource conflicts

**Project Infrastructure**:
- `pytest-xdist`: Parallel test execution
- `unique_resource_name` fixture: Collision-free resource names
- `db_session` fixture: Database transaction rollback
- Factories: Dynamic test data generation

---

## Why Test Isolation Matters

### False Positives from Shared State

**Problem**: Test fails because another test left behind state.

**Example**:
```python

# BAD: Test depends on global state

def test_create_agent_1():

    agent = AgentRegistry(id="test-agent", name="Agent 1")

    db.add(agent)

    db.commit()



def test_create_agent_2():

    # FAILS if test_create_agent_1 ran first (duplicate ID)

    agent = AgentRegistry(id="test-agent", name="Agent 2")

    db.add(agent)

    db.commit()  # IntegrityError: duplicate id

```

**Root Cause**: Hardcoded ID "test-agent" causes collision.

**Impact**: Test suite passes when run individually, fails in CI.

---

### False Negatives from Shared State

**Problem**: Test passes only because another test set up state.

**Example**:
```python

# BAD: Test depends on execution order

def test_update_agent():

    agent = db.query(AgentRegistry).first()

    # PASSES only if test_create_agent ran first

    agent.name = "Updated"

    db.commit()



# If run alone: agent = None → AttributeError

```

**Root Cause**: Assumes test_create_agent runs first.

**Impact**: Test passes in suite, fails individually.

---

### Flaky Tests from Resource Conflicts

**Problem**: Tests fail intermittently due to port/file/name conflicts.

**Example**:
```python

# BAD: Hardcoded port causes conflict

def test_api_server():

    server = start_server(port=8000)  # Port in use?



def test_websocket_server():

    server = start_server(port=8000)  # Fails if first test running

```

**Root Cause**: Both tests try to bind port 8000.

**Impact**: Tests pass in sequential execution, fail in parallel (`pytest -n auto`).

---

### Parallel Execution Requirements

**pytest-xdist** runs tests in parallel across multiple CPU cores.

**Requirement**: Tests must not share:
- Database rows (hardcoded IDs)
- File system paths (hardcoded filenames)
- Network ports (hardcoded ports)
- Global variables (mutable state)

**Verification**:
```bash

# Run tests sequentially

pytest tests/ -v



# Run tests in parallel (should give same results)

pytest tests/ -n auto -v



# If results differ → isolation problem

```

---

### Factory Session Enforcement

**Problem**: Forgetting to inject `db_session` into factory calls causes test data leaks.

**Example**:
```python

# BAD: Factory uses default session (data persists across tests)

def test_agent_creation():

    agent = AgentFactory.create()  # No _session parameter!

    # Data persists in database, affects other tests

```

**Root Cause**: BaseFactory doesn't enforce session injection, making it easy to forget.

**Solution**: BaseFactory enforces `_session` parameter in test environment (Pattern 2).

```python

# GOOD: Factory uses test session (transaction rollback)

def test_agent_creation(db_session):

    agent = AgentFactory.create(_session=db_session)

    # Data rolled back after test, no side effects

```

**Enforcement**: RuntimeError raised when `_session` missing in test environment.

**Impact**: Prevents test data collisions, ensures parallel execution safety.

---

## Isolation Patterns

### Pattern 1: unique_resource_name Fixture

**Purpose**: Generate collision-free resource names for parallel execution.

**Source**: `backend/tests/conftest.py`

**Implementation**:
```python

@pytest.fixture(scope="function")

def unique_resource_name():

    """

    Generate unique resource name for parallel test execution.

    Combines worker ID with UUID to ensure no collisions.

    """

    worker_id = os.environ.get('PYTEST_XDIST_WORKER_ID', 'master')

    unique_id = str(uuid.uuid4())[:8]

    return f"test_{worker_id}_{unique_id}"

```

**How It Works**:
1. `PYTEST_XDIST_WORKER_ID` set by pytest-xdist (e.g., "gw0", "gw1")
2. UUID ensures uniqueness within worker
3. Combined format: `test_gw0_a1b2c3d4`

**Usage Example**:
```python

def test_create_agent_with_unique_id(unique_resource_name):

    # GOOD: No collision with parallel tests

    agent = AgentRegistry(

        id=unique_resource_name,  # "test_gw0_a1b2c3d4"

        name="Test Agent"

    )

    db.add(agent)

    db.commit()

```

**Before (Bad)**:
```python

def test_create_agent():

    agent = AgentRegistry(id="test-agent", ...)  # Collision!

```

**After (Good)**:
```python

def test_create_agent(unique_resource_name):

    agent = AgentRegistry(id=unique_resource_name, ...)  # Unique!

```

---

### Pattern 2: Factory-Boy with Session Injection

**Purpose**: Generate test data with proper database isolation.

**Source**: `backend/tests/factories/base.py`

**Enforcement**: BaseFactory requires `_session` parameter in test environment (detected by `PYTEST_XDIST_WORKER_ID`).

**Implementation**:
```python

class BaseFactory(SQLAlchemyModelFactory):

    class Meta:

        abstract = True

        sqlalchemy_session = None

        sqlalchemy_session_persistence = "commit"



    @classmethod

    def _create(cls, model_class, *args, **kwargs):

        """Override to handle session injection with enforcement."""

        session = kwargs.pop('_session', None)

        is_test_env = os.environ.get('PYTEST_XDIST_WORKER_ID') is not None



        # Enforce _session in test environment

        if is_test_env and session is None:

            raise RuntimeError(

                "{0}.create() requires _session parameter in test environment. "

                "Usage: {0}.create(_session=db_session, ...)".format(cls.__name__)

            )



        if session:

            cls._meta.sqlalchemy_session = session

            cls._meta.sqlalchemy_session_persistence = "flush"

        else:

            if cls._meta.sqlalchemy_session is None:

                cls._meta.sqlalchemy_session = get_session()

            cls._meta.sqlalchemy_session_persistence = "commit"

        return super()._create(model_class, *args, **kwargs)

```

**How It Works**:
1. Test environment detected by `PYTEST_XDIST_WORKER_ID` environment variable
2. If `_session` not provided in test environment, raises `RuntimeError`
3. Error message includes factory name and correct usage example
4. When `_session` provided, uses "flush" mode (transaction rollback, not commit)

**Usage Example**:
```python

def test_agent_creation(unique_resource_name, db_session):

    # Each test gets isolated agent

    agent = AgentFactory.create(

        _session=db_session,  # REQUIRED in test environment

        id=unique_resource_name,

        name="Test Agent"

    )



    # Agent visible in test session

    assert db_session.query(AgentRegistry).filter_by(id=agent.id).count() == 1

```

**Error Handling**:
- Missing `_session` → `RuntimeError: AgentFactory.create() requires _session parameter`
- Solution: Add `_session=db_session` to factory call

**Why It Works**:
- `db_session` fixture uses transaction rollback (instant cleanup)
- Each worker (gw0, gw1, gw2, gw3) has separate database schema
- No data conflicts between parallel tests
- Enforcement prevents bugs from forgetting `_session` parameter

**Before (Bad)**:
```python

def test_agent_creation():

    agent = AgentFactory.create()  # RuntimeError in test environment!

```

**After (Good)**:
```python

def test_agent_creation(db_session):

    agent = AgentFactory.create(_session=db_session)  # Explicit isolation

```

**See also**: `backend/tests/factories/README.md` for complete factory documentation.

---

### Pattern 3: Database Transaction Rollback

**Purpose**: Isolate database operations with automatic cleanup.

**Implementation**:
```python

@pytest.fixture(scope="function")

def db_session():

    """

    Create isolated database session with automatic rollback.

    Ensures zero shared state between parallel tests.

    """

    from core.models import SessionLocal



    session = SessionLocal()

    transaction = session.begin()



    yield session



    # Rollback transaction to clean up test data

    session.rollback()

    session.close()

```

**How It Works**:
1. Begin transaction before test
2. Test executes with session
3. Rollback after test (undo all changes)
4. Close session

**Usage Example**:
```python

def test_agent_creation(db_session, unique_resource_name):

    # Create agent

    agent = AgentRegistry(

        id=unique_resource_name,

        name="Test Agent"

    )

    db_session.add(agent)

    db_session.commit()



    # Test logic...



    # Automatic rollback via fixture

    # No other test sees this agent

```

**Before (Bad)**:
```python

def test_agent_creation():

    agent = AgentRegistry(id="test-agent", ...)

    db.add(agent)

    db.commit()

    # Manual cleanup required (error-prone)

    db.delete(agent)

    db.commit()

```

**After (Good)**:
```python

def test_agent_creation(db_session, unique_resource_name):

    agent = AgentRegistry(id=unique_resource_name, ...)

    db_session.add(agent)

    db_session.commit()

    # Automatic rollback - no cleanup needed

```

**Benefits**:
- No manual cleanup
- No shared state between tests
- Fast (rollback is instant)
- Safe for parallel execution

---

### Pattern 4: Worker-Specific Database Isolation

**Purpose**: Enable parallel test execution without data conflicts.

**Source**: `backend/tests/conftest.py`

**Implementation**: Each pytest-xdist worker gets its own PostgreSQL database.

**Worker Databases**:
- gw0: `test_db_gw0`
- gw1: `test_db_gw1`
- gw2: `test_db_gw2`
- gw3: `test_db_gw3`

**For Local Development (SQLite)**:
Uses in-memory database (no worker isolation needed):
```python

# DATABASE_URL=sqlite:///:memory:

# No worker-specific schemas created

```

**For CI (PostgreSQL)**:
Creates and drops worker databases:
```python

# DATABASE_URL=postgresql://user:pass@host/test_db

# Creates: test_db_gw0, test_db_gw1, test_db_gw2, test_db_gw3

```

**Usage Example**:
```python

def test_parallel_isolation(unique_resource_name, db_session):

    # Each worker (gw0, gw1, gw2, gw3) creates unique agent

    agent = AgentFactory.create(

        _session=db_session,

        id=unique_resource_name  # test_gw0_a1b2c3d4

    )



    # Only this worker sees this agent (different database)

    assert db_session.query(AgentRegistry).filter_by(id=agent.id).count() == 1

```

**Transaction Rollback**:
- `db_session` fixture uses `begin_nested()` for instant rollback
- Changes discarded after test (<1ms cleanup)
- No foreign key cascade issues (transaction isolation)

**Verification**:
```bash

# Run tests in parallel (4 workers)

pytest tests/ -n auto -v



# All workers pass (no data conflicts)

# gw0: test_db_gw0

# gw1: test_db_gw1

# gw2: test_db_gw2

# gw3: test_db_gw3

```

**Cleanup**:
- Worker databases dropped after test session completes
- No manual cleanup required

---

### Pattern 5: Factory Pattern

**Purpose**: Generate dynamic test data with no hardcoded values.

**Source**: `backend/tests/factories/`

**Implementation**:
```python

class AgentFactory(BaseFactory):

    class Meta:

        model = AgentRegistry



    id = LazyFunction(lambda: str(uuid.uuid4()))

    name = LazyFunction(lambda: fake.company())

    status = "STUDENT"

    confidence = 0.5

```

**How It Works**:
- `LazyFunction`: Call function for each instance (not class-level)
- `Faker`: Generate realistic data (names, emails, companies)
- `SubFactory`: Handle relationships

**Usage Example**:
```python

from tests.factories import AgentFactory, UserFactory



def test_agent_governance(unique_resource_name):

    # GOOD: Unique IDs every time

    agent = AgentFactory.create(

        id=unique_resource_name,

        status="STUDENT",

        confidence=0.4

    )

    # agent.id is unique (UUID + worker ID)

    # agent.name is realistic (Faker-generated)



    user = UserFactory.create()

    # user.email is unique (Faker-generated)

    # No collision with parallel tests

```

**Before (Bad)**:
```python

def test_agent_governance():

    agent = AgentRegistry(

        id="test-agent-123",  # Hardcoded!

        name="Test Agent",     # Hardcoded!

        email="test@example.com"  # Hardcoded!

    )

```

**After (Good)**:
```python

def test_agent_governance(unique_resource_name):

    agent = AgentFactory.create(

        id=unique_resource_name  # Unique!

    )

    # All other fields auto-generated

```

**Benefits**:
- Unique data every time (no collisions)
- Realistic data (better than "test123")
- Easy to override specific fields
- Relationships handled automatically

**See**: `backend/tests/factories/README.md` for complete factory guide.

---

### Pattern 6: Mock External Dependencies

**Purpose**: Isolate tests from external systems (APIs, databases, file systems).

**Implementation**:
```python

from unittest.mock import patch, MagicMock



def test_external_api_call():

    # GOOD: Mock external API

    with patch('core.services.external_api_client') as mock_api:

        mock_api.call.return_value = {"status": "success"}



        result = service.process_data()



        assert result["status"] == "success"

        mock_api.call.assert_called_once()

```

**Before (Bad)**:
```python

def test_external_api_call():

    # BAD: Calls real API (slow, unreliable, requires auth)

    result = service.process_data()

    assert result["status"] == "success"

```

**After (Good)**:
```python

def test_external_api_call():

    # GOOD: Mocked API (fast, reliable, no auth)

    with patch('core.services.external_api_client') as mock_api:

        mock_api.call.return_value = {"status": "success"}

        result = service.process_data()

        assert result["status"] == "success"

```

**What to Mock**:
- External APIs (HTTP calls, third-party services)
- File system operations (if not testing file I/O)
- Database connections (use db_session instead)

- Time/date (use freezegun)

- Random values (use seed or mock)



**What NOT to Mock**:

- Business logic (you're testing this)

- Data structures (use real objects)

- Database operations (use transaction rollback)



---



### Pattern 7: Fixture Cleanup



**Purpose**: Ensure cleanup even if test fails.



**Implementation**:

```python

@pytest.fixture(scope="function")

def temp_file():
    """

    Create temporary file with automatic cleanup.

    Cleanup runs even if test fails.

    """

    import tempfile

    fd, path = tempfile.mkstemp(suffix=".txt")


    yield path


    # Cleanup runs regardless of test result

    os.close(fd)

    os.unlink(path)


# Usage
def test_file_operations(temp_file):

    with open(temp_file, 'w') as f:
        f.write("test data")


    # Test logic...


    # Automatic cleanup via fixture

```


**Alternative: yield + try/finally**
```python

@pytest.fixture(scope="function")

def temp_directory():

    temp_dir = tempfile.mkdtemp()

    try:

        yield temp_dir

    finally:

        shutil.rmtree(temp_dir)  # Always runs

```

**Benefits**:
- No manual cleanup in tests
- Cleanup runs even if test fails
- No resource leaks (files, ports, connections)

---

## Anti-Patterns to Avoid

### Anti-Pattern 1: Hardcoded Test Data

**Problem**: Hardcoded values cause collisions in parallel execution.

**Example**:
```python

# BAD: Hardcoded ID

def test_create_agent():

    agent = AgentRegistry(id="test-agent-123", ...)

    # Fails if another test uses same ID



# BAD: Hardcoded filename

def test_file_operations():

    with open("test_file.txt", 'w') as f:

        f.write("data")

    # Fails if another test creates same file



# BAD: Hardcoded port

def test_server():

    server = start_server(port=8000)

    # Fails if another test binds port 8000

```

**Solution**: Use unique_resource_name or factories.

```python

# GOOD: Unique ID

def test_create_agent(unique_resource_name):

    agent = AgentRegistry(id=unique_resource_name, ...)



# GOOD: Unique filename

def test_file_operations(unique_resource_name):

    filename = f"{unique_resource_name}.txt"

    with open(filename, 'w') as f:

        f.write("data")



# GOOD: Dynamic port

def test_server():

    port = get_ephemeral_port()  # Get free port

    server = start_server(port=port)

```

---

### Anti-Pattern 2: Global State Mutations

**Problem**: Tests modify global variables, affecting other tests.

**Example**:
```python

# BAD: Modifying global variable

SETTINGS = {"debug": False}



def test_enable_debug():

    global SETTINGS

    SETTINGS["debug"] = True  # Affects other tests



def test_feature_flag():

    # FAILS if test_enable_debug ran first

    assert SETTINGS["debug"] == False

```

**Solution**: Use fixtures to reset state.

```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 to this test



def test_feature_flag():

    assert SETTINGS["debug"] == False  # Passes (state reset)

```

---

### Anti-Pattern 3: Time-Based Tests Without Mocking

**Problem**: Tests depend on current time, causing non-deterministic results.

**Example**:
```python

# BAD: Depends on current time

def test_token_expiry():

    token = create_token(expiry_hours=24)

    assert token.expires_at > datetime.now()  # Flaky near boundary



# BAD: Uses sleep (slow, non-deterministic)

def test_cache_timeout():

    cache.set("key", "value", timeout=1)

    time.sleep(1)  # SLOW!

    assert cache.get("key") is None

```

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

```python

# GOOD: Freeze 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)



# GOOD: Freeze time for timeout tests

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

```

---

### Anti-Pattern 4: File System Operations Without Temp Directories

**Problem**: Tests create files in current directory, causing collisions.

**Example**:
```python

# BAD: Creates file in current directory

def test_export_data():

    export_to_file("data.json")  # Collisions with parallel tests



# BAD: Hardcoded path

def test_import_data():

    import_from_file("/tmp/test.json")  # May not exist, permissions issues

```

**Solution**: Use tempfile module.

```python

# GOOD: Temporary file

def test_export_data(unique_resource_name):

    temp_dir = tempfile.mkdtemp()

    try:

        filepath = os.path.join(temp_dir, f"{unique_resource_name}.json")

        export_to_file(filepath)

        assert os.path.exists(filepath)

    finally:

        shutil.rmtree(temp_dir)  # Cleanup



# GOOD: Fixture-based temp directory

@pytest.fixture

def temp_dir():

    dir_path = tempfile.mkdtemp()

    yield dir_path

    shutil.rmtree(dir_path)



def test_import_data(temp_dir):

    filepath = os.path.join(temp_dir, "test.json")

    import_from_file(filepath)

```

---

### Anti-Pattern 5: Database Commits in Tests

**Problem**: Tests commit data, affecting other tests.

**Example**:
```python

# BAD: Commits data to database

def test_create_agent():

    agent = AgentRegistry(id="test-agent", ...)

    db.add(agent)

    db.commit()  # Data persists after test



# BAD: Manual cleanup (error-prone)

def test_create_agent():

    agent = AgentRegistry(id="test-agent", ...)

    db.add(agent)

    db.commit()

    # Test logic...

    db.delete(agent)  # May not run if test fails

    db.commit()

```

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



```python

# GOOD: Transaction rollback

def test_create_agent(db_session, unique_resource_name):
    agent = AgentRegistry(id=unique_resource_name, ...)

    db_session.add(agent)

    db_session.commit()  # Transaction rolled back after test

    # No manual cleanup needed

```


**Key Insight**: Rollback is instant (no actual DELETE queries).

---

## Examples

### Good: Using unique_resource_name for Filenames

```python

import tempfile

import os



def test_file_export(unique_resource_name):

    """Test file export with unique filename."""

    temp_dir = tempfile.mkdtemp()

    try:

        # GOOD: Unique filename (no collisions)

        filename = os.path.join(temp_dir, f"{unique_resource_name}.csv")

        export_data_to_csv(filename, data=[1, 2, 3])



        # Verify file exists and has correct content

        assert os.path.exists(filename)

        with open(filename, 'r') as f:

            assert f.read() == "1,2,3\n"

    finally:

        shutil.rmtree(temp_dir)  # Cleanup

```

**Bad Alternative**:
```python

def test_file_export():

    # BAD: Hardcoded filename (collisions in parallel)

    filename = "test_export.csv"

    export_data_to_csv(filename, data=[1, 2, 3])

    # File persists after test

```

---

### Good: Using db_session with Automatic Rollback



```python

def test_agent_deletion(db_session, unique_resource_name):
    """Test agent deletion with database isolation."""

    # Create agent

    agent = AgentFactory.create(_session=db_session, id=unique_resource_name)


    # Delete agent

    db_session.delete(agent)

    db_session.commit()


    # Verify deletion

    assert db_session.query(AgentRegistry).filter_by(id=agent.id).first() is None


    # Automatic rollback: other tests don't see this deletion

```


**Bad Alternative**:
```python

def test_agent_deletion():

    # BAD: Manual cleanup (error-prone)

    agent = AgentRegistry(id="test-agent-123")

    db.add(agent)

    db.commit()



    db.delete(agent)

    db.commit()



    # Manual cleanup (may not run if test fails)

    if db.query(AgentRegistry).filter_by(id="test-agent-123").first():

        db.delete(agent)

        db.commit()

```

---

### Good: Mocking External APIs

```python

from unittest.mock import patch, MagicMock



def test_payment_processing(unique_resource_name):

    """Test payment processing with mocked external API."""

    # GOOD: Mock external payment API

    with patch('core.services.payment_gateway.charge') as mock_charge:

        mock_charge.return_value = {

            "status": "success",

            "transaction_id": "txn_123"

        }



        # Process payment

        result = process_payment(

            user_id=unique_resource_name,

            amount=100.00

        )



        # Verify result

        assert result["status"] == "success"

        assert result["transaction_id"] == "txn_123"



        # Verify API was called

        mock_charge.assert_called_once_with(amount=100.00)

```

**Bad Alternative**:
```python

def test_payment_processing():

    # BAD: Calls real payment API (charges actual money!)

    result = process_payment(user_id="test-user", amount=100.00)

    assert result["status"] == "success"

```

---

## Debugging Isolation Issues

### Identifying Shared State Problems

**Symptom**: Tests pass alone but fail in suite.

**Diagnosis**:
```bash

# Run test alone (passes)

pytest tests/test_agent.py::test_create_agent -v



# Run in suite (fails)

pytest tests/ -v



# Run with fresh database (passes)

pytest tests/ -v --create-db

```

**Root Cause**: Previous tests left behind data.

**Solution**:
1. Use `db_session` fixture (automatic rollback)
2. Use `unique_resource_name` fixture (unique IDs)
3. Use factories (dynamic data)

---

### Finding Resource Conflicts

**Symptom**: "Port already in use", "File exists", "Duplicate key" errors.

**Diagnosis**:
```bash

# Run tests in parallel (fails)

pytest tests/ -n auto -v



# Run tests sequentially (passes)

pytest tests/ -v



# Check for hardcoded values

grep -r "port=8000\|\"test-agent\|\"test.txt" tests/

```

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

**Solution**:
1. Use `unique_resource_name` for dynamic names
2. Use `tempfile.mkdtemp()` for temporary files
3. Use `get_ephemeral_port()` for dynamic ports

---

### Verifying Isolation

**Step 1: Run Tests 10 Times**
```bash

for i in {1..10}; do

    pytest tests/ -v --tb=short

done



# If results vary → flaky test (isolation issue)

```

**Step 2: Run in Random Order**
```bash

pip install pytest-randomly



pytest tests/ -v



# If results vary → order dependency

```

**Step 3: Run in Parallel**
```bash

pytest tests/ -n auto -v



# If failures don't reproduce sequentially → resource conflict

```

**Step 4: Check for Global State**
```bash

grep -r "global\|GLOBALS\|_state" tests/



# Find mutable globals and reset with fixtures

```

---

## pytest-xdist Integration

### Worker ID Environment Variable

**PYTEST_XDIST_WORKER_ID**: Set by pytest-xdist to identify worker.



**Values**:

- `master`: Main process (sequential execution)

- `gw0`, `gw1`, `gw2`, ...: Worker IDs (parallel execution)



**Usage**:

```python

# In conftest.py

def pytest_configure(config):

    if hasattr(config, 'workerinput'):

        worker_id = config.workerinput.get('workerid', 'master')

        os.environ['PYTEST_XDIST_WORKER_ID'] = worker_id



# In fixtures

@pytest.fixture

def worker_log_file():

    worker_id = os.environ.get('PYTEST_XDIST_WORKER_ID', 'master')

    return f"test_{worker_id}.log"

```



---



### Load Scope Scheduling



**`--dist loadscope`**: Group tests by scope (test module, class, function).



**Configuration** (pytest.ini):

```ini

[pytest]

addopts = -n auto --dist loadscope

```



**Benefits**:

- Tests in same module run on same worker

- Reduces database lock contention

- Better isolation for module-level fixtures



**Example**:

```bash

# All tests in test_agent.py run on worker gw0

# All tests in test_user.py run on worker gw1

# Reduces database conflicts (agent tests don't interfere with user tests)

```



---



### Parallel Execution Verification



**Step 1: Verify pytest-xdist Installed**

```bash

pip install pytest-xdist

pytest --version  # Should show "pytest-xdist"

```



**Step 2: Run Tests in Parallel**

```bash

pytest tests/ -n auto -v



# Should see worker IDs in output (gw0, gw1, ...)

```



**Step 3: Verify Speedup**

```bash

# Sequential execution

time pytest tests/ -v



# Parallel execution

time pytest tests/ -n auto -v



# Parallel should be ~2-4x faster (depending on CPU cores)

```



**Step 4: Verify Consistency**

```bash

# Run sequential

pytest tests/ -v > sequential.txt



# Run parallel

pytest tests/ -n auto -v > parallel.txt



# Compare results

diff sequential.txt parallel.txt



# Should be identical (except for worker IDs)

```



---



## Related Documentation



- **[COVERAGE_GUIDE.md](./COVERAGE_GUIDE.md)** - Coverage report interpretation

- **[FLAKY_TEST_GUIDE.md](./FLAKY_TEST_GUIDE.md)** - Flaky test prevention

- **[../conftest.py](../conftest.py)** - Fixture definitions

- **[../factories/README.md](../factories/README.md)** - Test data factory usage



---



## Summary



**Key Takeaways**:

1. **unique_resource_name**: Collision-free resource names (worker ID + UUID)

2. **db_session**: Database isolation via transaction rollback
3. **Factories**: Dynamic test data generation (no hardcoded values)
4. **Mocking**: Isolate external dependencies (APIs, file systems)
5. **Fixture cleanup**: Ensure cleanup even if test fails

**Anti-Patterns**:
- Hardcoded test data → Use factories
- Global state mutations → Reset with fixtures
- Time-based tests → Mock with freezegun
- File operations → Use tempfile
- Database commits → Use db_session



**Quick Reference**:

```python

# GOOD: Isolated test

def test_agent_creation(db_session, unique_resource_name):
    agent = AgentFactory.create(

        _session=db_session,

        id=unique_resource_name

    )

    # Automatic cleanup, no shared state


# BAD: Non-isolated test
def test_agent_creation():
    agent = AgentRegistry(id="test-agent-123", ...)

    db.commit()  # Persists data

    # Manual cleanup required

```


**Next Steps**:
1. Run tests in parallel: `pytest tests/ -n auto -v`
2. Fix isolation issues using patterns above
3. Verify tests pass 10 times in a row
4. Add to CI/CD for continuous validation