0
# Worker Detection
1

2
Functions to detect and control the execution environment, allowing tests and plugins to behave differently when running in distributed mode versus single-process mode.
3

4
## Capabilities
5

6
### Worker Process Detection
7

8
Determine if the current process is a worker process in a distributed test run.
9

10
```python { .api }
11
def is_xdist_worker(request_or_session: pytest.FixtureRequest | pytest.Session) -> bool:
12
    """
13
    Return True if this is an xdist worker, False otherwise.
14
    
15
    Args:
16
        request_or_session: pytest request fixture or session object
17
        
18
    Returns:
19
        bool: True if running in worker process
20
    """
21
```
22

23
**Usage:**
24
- Check if test is running in a worker process
25
- Conditional logic for worker-only or controller-only code
26
- Resource management (e.g., shared resources only in controller)
27

28
### Controller Process Detection
29

30
Determine if the current process is the controller (master) process coordinating workers.
31

32
```python { .api }
33
def is_xdist_controller(request_or_session: pytest.FixtureRequest | pytest.Session) -> bool:
34
    """
35
    Return True if this is the xdist controller, False otherwise.
36
    
37
    Note: this method also returns False when distribution has not been
38
    activated at all.
39
    
40
    Args:
41
        request_or_session: pytest request fixture or session object
42
        
43
    Returns:
44
        bool: True if running in controller process with distribution active
45
    """
46
```
47

48
**Behavior:**
49
- Returns `True` only when distribution is active AND running in controller
50
- Returns `False` for single-process runs (no distribution)
51
- Returns `False` for worker processes
52

53
### Legacy Master Detection (Deprecated)
54

55
Deprecated alias for controller detection. Use `is_xdist_controller` instead.
56

57
```python { .api }
58
def is_xdist_master(request_or_session: pytest.FixtureRequest | pytest.Session) -> bool:
59
    """
60
    Deprecated alias for is_xdist_controller.
61
    
62
    Args:
63
        request_or_session: pytest request fixture or session object
64
        
65
    Returns:
66
        bool: True if running in controller process
67
    """
68
```
69

70
### Worker ID Retrieval
71

72
Get the unique identifier for the current worker process.
73

74
```python { .api }
75
def get_xdist_worker_id(request_or_session: pytest.FixtureRequest | pytest.Session) -> str:
76
    """
77
    Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
78
    if running on the controller node.
79
    
80
    If not distributing tests (for example passing -n0 or not passing -n at all)
81
    also return 'master'.
82
    
83
    Args:
84
        request_or_session: pytest request fixture or session object
85
        
86
    Returns:
87
        str: Worker ID ('gw0', 'gw1', etc.) or 'master' for controller/single-process
88
    """
89
```
90

91
**Return Values:**
92
- `'gw0'`, `'gw1'`, `'gw2'`, etc. for worker processes
93
- `'master'` for controller process or single-process runs
94

95
### Session-Scoped Fixtures
96

97
Convenient fixtures for accessing worker information in tests.
98

99
```python { .api }
100
@pytest.fixture(scope="session")
101
def worker_id(request: pytest.FixtureRequest) -> str:
102
    """
103
    Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
104
    if running on the master node.
105
    
106
    Returns:
107
        str: Worker ID or 'master'
108
    """
109

110
@pytest.fixture(scope="session")  
111
def testrun_uid(request: pytest.FixtureRequest) -> str:
112
    """
113
    Return the unique id of the current test run.
114
    
115
    This ID is shared across all workers in the same test run,
116
    making it useful for coordinating between workers or creating
117
    unique resource names.
118
    
119
    Returns:
120
        str: Unique test run identifier (hex string)
121
    """
122
```
123

124
## Usage Examples
125

126
### Conditional Test Behavior
127

128
```python
129
import pytest
130
from xdist import is_xdist_worker, get_xdist_worker_id
131

132
def test_with_worker_logic(request):
133
    if is_xdist_worker(request):
134
        # This code only runs in worker processes
135
        worker_id = get_xdist_worker_id(request)
136
        print(f"Running in worker {worker_id}")
137
        
138
        # Worker-specific setup
139
        setup_worker_resources()
140
    else:
141
        # This code runs in controller or single-process mode
142
        print("Running in controller process")
143
        
144
        # Controller-specific setup
145
        setup_shared_resources()
146
```
147

148
### Resource Management
149

150
```python
151
import pytest
152
from xdist import is_xdist_controller
153

154
@pytest.fixture(scope="session")
155
def database_connection(request):
156
    if is_xdist_controller(request):
157
        # Only controller sets up the database
158
        db = create_test_database()
159
        yield db
160
        cleanup_test_database(db)
161
    else:
162
        # Workers get connection info via other means
163
        yield get_worker_db_connection()
164
```
165

166
### Using Fixtures
167

168
```python
169
def test_worker_coordination(worker_id, testrun_uid):
170
    # worker_id is 'gw0', 'gw1', etc. for workers, 'master' for controller
171
    # testrun_uid is the same across all workers in this test run
172
    
173
    temp_file = f"/tmp/test_{testrun_uid}_{worker_id}.tmp"
174
    with open(temp_file, 'w') as f:
175
        f.write(f"Worker {worker_id} data")
176
    
177
    # Do test work
178
    assert os.path.exists(temp_file)
179
    
180
    # Cleanup
181
    os.unlink(temp_file)
182
```
183

184
### Plugin Development
185

186
```python
187
class MyPlugin:
188
    def pytest_runtest_setup(self, item):
189
        if is_xdist_worker(item.session):
190
            # Worker-specific test setup
191
            self.setup_worker_test_env()
192
        elif is_xdist_controller(item.session):
193
            # Controller-specific test setup  
194
            self.setup_controller_test_env()
195
        else:
196
            # Single-process test setup
197
            self.setup_single_process_test_env()
198
```
199

200
### Shared State Management
201

202
```python
203
import pytest
204
from xdist import is_xdist_worker, get_xdist_worker_id
205

206
# Global state that should only be initialized once per worker
207
_worker_state = None
208

209
def get_worker_state(request):
210
    global _worker_state
211
    if _worker_state is None:
212
        if is_xdist_worker(request):
213
            worker_id = get_xdist_worker_id(request)
214
            _worker_state = WorkerState(worker_id)
215
        else:
216
            _worker_state = ControllerState()
217
    return _worker_state
218

219
def test_using_worker_state(request):
220
    state = get_worker_state(request)
221
    # Use worker-specific or controller state
222
    state.do_something()
223
```