0
# Coverage Controllers
1

2
pytest-cov uses different coverage controller classes to handle various testing scenarios including single-process testing, distributed master coordination, and distributed worker execution. Each controller manages coverage measurement, data collection, and reporting appropriate to its context.
3

4
## Capabilities
5

6
### Base Controller
7

8
The foundational controller class that provides common functionality for all coverage scenarios.
9

10
```python { .api }
11
class CovController:
12
    """
13
    Base class for coverage controller implementations.
14
    
15
    Provides common functionality for coverage measurement including environment
16
    variable management, directory handling, and report generation.
17
    """
18
    
19
    def __init__(self, options: argparse.Namespace, config: Union[None, object], nodeid: Union[None, str]):
20
        """
21
        Initialize coverage controller with configuration.
22
        
23
        Args:
24
            options: Parsed command-line options containing coverage settings
25
            config: pytest configuration object (may be None)
26
            nodeid: Node identifier for distributed testing (may be None)
27
        """
28
    
29
    def start(self):
30
        """
31
        Start coverage measurement.
32
        
33
        Marks the controller as started. Specific implementations override
34
        this method to initialize their coverage objects and begin measurement.
35
        """
36
    
37
    def finish(self):
38
        """
39
        Finish coverage measurement.
40
        
41
        Marks the controller as stopped. Specific implementations override
42
        this method to stop coverage and save data.
43
        """
44
    
45
    def pause(self):
46
        """
47
        Pause coverage measurement temporarily.
48
        
49
        Stops coverage collection and removes environment variables.
50
        Used when running code that should not be measured (e.g., tests marked with no_cover).
51
        """
52
    
53
    def resume(self):
54
        """
55
        Resume coverage measurement after pausing.
56
        
57
        Restarts coverage collection and restores environment variables.
58
        """
59
    
60
    def summary(self, stream) -> Optional[float]:
61
        """
62
        Generate coverage reports and return total coverage percentage.
63
        
64
        Args:
65
            stream: Output stream for report text
66
            
67
        Returns:
68
            Optional[float]: Total coverage percentage, or None if reporting disabled
69
        """
70
```
71

72
### Environment Management
73

74
Methods for managing environment variables that enable subprocess coverage.
75

76
```python { .api }
77
def set_env(self):
78
    """
79
    Set environment variables for subprocess coverage.
80
    
81
    Sets COV_CORE_SOURCE, COV_CORE_CONFIG, COV_CORE_DATAFILE, and 
82
    COV_CORE_BRANCH environment variables so that forked processes
83
    and subprocesses can automatically enable coverage measurement.
84
    """
85

86
def unset_env(self):
87
    """
88
    Remove coverage-related environment variables.
89
    
90
    Cleans up COV_CORE_* environment variables to prevent interference
91
    with subsequent processes.
92
    """
93
```
94

95
### Utility Methods
96

97
Helper methods for controller operation and display formatting.
98

99
```python { .api }
100
def ensure_topdir(self):
101
    """
102
    Context manager ensuring operations run in the correct directory.
103
    
104
    Returns:
105
        Context manager that changes to topdir and restores original directory
106
    """
107

108
def get_node_desc(platform: str, version_info: tuple) -> str:
109
    """
110
    Generate description string for a testing node.
111
    
112
    Args:
113
        platform: Platform identifier (e.g., 'linux', 'win32')
114
        version_info: Python version info tuple
115
        
116
    Returns:
117
        str: Formatted node description
118
    """
119

120
def get_width() -> int:
121
    """
122
    Get terminal width for report formatting.
123
    
124
    Returns:
125
        int: Terminal width in characters (minimum 40, default 80)
126
    """
127

128
def sep(self, stream, s: str, txt: str):
129
    """
130
    Write separator line to output stream.
131
    
132
    Args:
133
        stream: Output stream
134
        s: Separator character
135
        txt: Text to center in separator line
136
    """
137
```
138

139
### Single-Process Controller
140

141
Controller for standard single-process test execution.
142

143
```python { .api }
144
class Central(CovController):
145
    """
146
    Coverage controller for single-process (centralized) test execution.
147
    
148
    Handles standard pytest runs without distributed testing, managing
149
    a single coverage instance and combining data from any subprocesses.
150
    """
151
    
152
    def start(self):
153
        """
154
        Initialize and start centralized coverage measurement.
155
        
156
        Creates coverage.Coverage instance with configured options,
157
        sets up combining coverage for subprocess data, and begins
158
        coverage measurement. Warns if dynamic_context=test_function
159
        is configured (recommends --cov-context instead).
160
        """
161
    
162
    def finish(self):
163
        """
164
        Stop coverage and prepare data for reporting.
165
        
166
        Stops coverage measurement, saves data, creates combining coverage
167
        instance, loads and combines all coverage data (including subprocess
168
        data), and adds node description for reporting.
169
        """
170
```
171

172
### Distributed Master Controller
173

174
Controller for the master process in distributed testing scenarios.
175

176
```python { .api }
177
class DistMaster(CovController):
178
    """
179
    Coverage controller for distributed testing master process.
180
    
181
    Coordinates coverage measurement across multiple worker processes,
182
    collecting and combining coverage data from all workers while
183
    managing distributed testing configuration.
184
    """
185
    
186
    def start(self):
187
        """
188
        Initialize master coverage for distributed testing.
189
        
190
        Creates master coverage instance, validates configuration
191
        (raises DistCovError if dynamic_context=test_function with xdist),
192
        configures path mapping, and starts coverage measurement.
193
        """
194
    
195
    def configure_node(self, node):
196
        """
197
        Configure a worker node for distributed testing.
198
        
199
        Sends master configuration to worker including hostname,
200
        directory paths, and rsync roots for proper file path
201
        mapping and coverage data collection.
202
        
203
        Args:
204
            node: pytest-xdist node object
205
        """
206
    
207
    def testnodedown(self, node, error):
208
        """
209
        Handle worker node shutdown and collect coverage data.
210
        
211
        Retrieves coverage data from worker node, handles both collocated
212
        and remote worker scenarios, creates coverage instances for
213
        remote worker data, and updates path mappings for data combination.
214
        
215
        Args:
216
            node: pytest-xdist node object  
217
            error: Any error that occurred during shutdown
218
        """
219
    
220
    def finish(self):
221
        """
222
        Combine coverage data from all workers.
223
        
224
        Stops master coverage, saves data, switches to combining coverage
225
        instance, loads and combines data from all workers, and saves
226
        final combined coverage data.
227
        """
228
```
229

230
### Distributed Worker Controller
231

232
Controller for worker processes in distributed testing scenarios.
233

234
```python { .api }
235
class DistWorker(CovController):
236
    """
237
    Coverage controller for distributed testing worker processes.
238
    
239
    Handles coverage measurement in worker processes, managing path
240
    mapping for remote workers and sending coverage data back to master.
241
    """
242
    
243
    def start(self):
244
        """
245
        Initialize worker coverage for distributed testing.
246
        
247
        Determines if worker is collocated with master, adjusts source
248
        paths and config paths for remote workers, creates worker coverage
249
        instance with unique data suffix, and starts coverage measurement.
250
        """
251
    
252
    def finish(self):
253
        """
254
        Finish worker coverage and send data to master.
255
        
256
        Stops coverage measurement and handles data transmission:
257
        - Collocated workers: Save data file and send node ID to master
258
        - Remote workers: Combine data, serialize coverage data, and send
259
          complete data payload to master including path mapping info
260
        """
261
    
262
    def summary(self, stream):
263
        """
264
        No-op summary method for workers.
265
        
266
        Workers don't generate reports - only the master generates
267
        coverage summaries after combining all worker data.
268
        """
269
```
270

271
## Internal Utilities
272

273
Helper functions and classes that support controller functionality.
274

275
```python { .api }
276
class BrokenCovConfigError(Exception):
277
    """Exception raised when coverage configuration is invalid."""
278

279
class _NullFile:
280
    """File-like object that discards all writes."""
281
    
282
    @staticmethod
283
    def write(v):
284
        """Discard written content."""
285

286
def _ensure_topdir(meth):
287
    """
288
    Decorator ensuring method runs in the correct directory.
289
    
290
    Changes to controller's topdir before method execution and
291
    restores original directory afterward, handling cases where
292
    original directory no longer exists.
293
    """
294

295
def _backup(obj, attr):
296
    """
297
    Context manager for backing up and restoring object attributes.
298
    
299
    Args:
300
        obj: Object whose attribute will be backed up
301
        attr: Attribute name to backup
302
        
303
    Returns:
304
        Context manager that creates copy of attribute and restores it
305
    """
306

307
def _data_suffix(name: str) -> str:
308
    """
309
    Generate data file suffix for coverage files.
310
    
311
    Args:
312
        name: Suffix name identifier
313
        
314
    Returns:
315
        str: Complete data file suffix including random component
316
    """
317
```
318

319
## Usage Patterns
320

321
### Controller Selection
322

323
Controllers are automatically selected based on testing configuration:
324

325
```python
326
# Single-process testing
327
if not distributed:
328
    controller = Central(options, config, nodeid)
329

330
# Distributed master  
331
if distributed and master:
332
    controller = DistMaster(options, config, nodeid)
333

334
# Distributed worker
335
if distributed and worker:
336
    controller = DistWorker(options, config, nodeid)
337
```
338

339
### Lifecycle Management
340

341
All controllers follow the same lifecycle pattern:
342

343
```python
344
controller = ControllerClass(options, config, nodeid)
345
controller.start()
346
# ... test execution ...
347
controller.finish() 
348
total_coverage = controller.summary(output_stream)
349
```
350

351
### Path Mapping
352

353
For distributed testing with remote workers, controllers handle path mapping:
354

355
```python
356
# Master sends configuration
357
master.configure_node(worker_node)
358

359
# Worker adjusts paths if not collocated
360
if not worker.is_collocated:
361
    worker.cov_source = adjust_paths(worker.cov_source, master_path, worker_path)
362
```