0
# Subprocess Support
1
2
pytest-cov provides automatic coverage measurement for subprocesses and forked processes through environment variable propagation and embedded coverage initialization. This enables comprehensive coverage collection even when tests spawn additional processes.
3
4
## Capabilities
5
6
### Subprocess Coverage Initialization
7
8
Automatic coverage activation in subprocess environments based on environment variable configuration.
9
10
```python { .api }
11
def init():
12
"""
13
Initialize coverage in subprocess if environment variables are set.
14
15
Checks for COV_CORE_* environment variables set by parent process
16
and automatically starts coverage measurement in subprocess using
17
the same configuration. Creates coverage instance with proper
18
source filtering, branch coverage, and data file settings.
19
20
Returns:
21
coverage.Coverage: Active coverage instance, or None if not initialized
22
23
Environment Variables:
24
COV_CORE_SOURCE: Colon-separated source paths (empty means no filtering)
25
COV_CORE_CONFIG: Coverage config file path (colon means use default)
26
COV_CORE_DATAFILE: Coverage data file path
27
COV_CORE_BRANCH: 'enabled' to enable branch coverage
28
COV_CORE_CONTEXT: Coverage context name
29
"""
30
```
31
32
**Usage Examples:**
33
34
```python
35
# Automatic usage - no explicit calls needed
36
# When parent process has coverage enabled:
37
import subprocess
38
39
# This subprocess will automatically have coverage enabled
40
subprocess.run(['python', 'script.py'])
41
42
# Manual initialization in subprocess code
43
from pytest_cov.embed import init
44
cov = init() # Returns coverage object or None
45
```
46
47
### Coverage Cleanup
48
49
Proper cleanup of coverage measurement in subprocess environments.
50
51
```python { .api }
52
def cleanup():
53
"""
54
Clean up active coverage measurement in subprocess.
55
56
Stops active coverage instance, saves coverage data, disables
57
auto-save to prevent double-saving, and unregisters atexit
58
handlers. Handles signal-based cleanup gracefully.
59
"""
60
```
61
62
### Signal-based Cleanup
63
64
Automatic coverage cleanup when subprocess receives termination signals.
65
66
```python { .api }
67
def cleanup_on_signal(signum: int):
68
"""
69
Set up signal handler for coverage cleanup.
70
71
Installs a signal handler that will clean up coverage data
72
when the specified signal is received, ensuring coverage
73
data is saved even if subprocess is terminated unexpectedly.
74
75
Args:
76
signum: Signal number to handle (e.g., signal.SIGTERM, signal.SIGINT)
77
"""
78
79
def cleanup_on_sigterm():
80
"""
81
Set up SIGTERM handler for coverage cleanup.
82
83
Convenience function that sets up cleanup for SIGTERM signal,
84
the most common signal used to terminate processes gracefully.
85
"""
86
```
87
88
**Usage Examples:**
89
90
```python
91
from pytest_cov.embed import cleanup_on_sigterm
92
93
# Ensure coverage is saved if process receives SIGTERM
94
cleanup_on_sigterm()
95
96
# Handle custom signals
97
from pytest_cov.embed import cleanup_on_signal
98
import signal
99
cleanup_on_signal(signal.SIGUSR1)
100
```
101
102
### Internal Implementation
103
104
Low-level functions that support subprocess coverage functionality.
105
106
```python { .api }
107
def _cleanup(cov):
108
"""
109
Internal cleanup implementation for coverage instance.
110
111
Args:
112
cov: coverage.Coverage instance to clean up
113
"""
114
115
def _signal_cleanup_handler(signum: int, frame):
116
"""
117
Internal signal handler for coverage cleanup.
118
119
Handles cleanup during signal processing, manages pending signals
120
during cleanup, and properly chains to previous signal handlers.
121
122
Args:
123
signum: Signal number received
124
frame: Signal frame
125
"""
126
```
127
128
## Environment Variable Protocol
129
130
pytest-cov uses specific environment variables to communicate coverage configuration to subprocess environments:
131
132
### Coverage Configuration Variables
133
134
```python { .api }
135
# Environment variables set by parent process
136
COV_CORE_SOURCE: str # Colon-separated source paths for filtering
137
COV_CORE_CONFIG: str # Coverage configuration file path
138
COV_CORE_DATAFILE: str # Coverage data file path
139
COV_CORE_BRANCH: str # 'enabled' for branch coverage
140
COV_CORE_CONTEXT: str # Coverage context name
141
```
142
143
**Environment Variable Usage:**
144
145
```bash
146
# Example environment variables set by pytest-cov
147
export COV_CORE_SOURCE="/path/to/src:/path/to/lib"
148
export COV_CORE_CONFIG="/path/to/.coveragerc"
149
export COV_CORE_DATAFILE="/path/to/.coverage"
150
export COV_CORE_BRANCH="enabled"
151
export COV_CORE_CONTEXT="test_context"
152
153
# These are automatically set - no manual configuration needed
154
```
155
156
### Variable Processing
157
158
How environment variables are processed during subprocess initialization:
159
160
```python
161
# Source path handling
162
if cov_source == os.pathsep:
163
cov_source = None # No source filtering
164
else:
165
cov_source = cov_source.split(os.pathsep) # Split paths
166
167
# Config file handling
168
if cov_config == os.pathsep:
169
cov_config = True # Use default config discovery
170
# Otherwise use specified path
171
172
# Branch coverage
173
cov_branch = True if os.environ.get('COV_CORE_BRANCH') == 'enabled' else None
174
```
175
176
## Integration Patterns
177
178
### Automatic Subprocess Coverage
179
180
pytest-cov automatically enables subprocess coverage without requiring code changes:
181
182
```python
183
import subprocess
184
import os
185
186
# This subprocess will automatically have coverage
187
result = subprocess.run(['python', '-c', 'import mymodule; mymodule.function()'])
188
189
# Forked processes also get coverage
190
pid = os.fork()
191
if pid == 0:
192
# Child process automatically has coverage
193
import mymodule
194
mymodule.child_function()
195
os._exit(0)
196
```
197
198
### Manual Control
199
200
For advanced scenarios where manual control is needed:
201
202
```python
203
from pytest_cov.embed import init, cleanup, cleanup_on_sigterm
204
205
# Manual initialization with signal handling
206
cleanup_on_sigterm()
207
cov = init()
208
209
try:
210
# Subprocess code that should be measured
211
import mymodule
212
mymodule.do_work()
213
finally:
214
# Manual cleanup (also happens automatically on process exit)
215
cleanup()
216
```
217
218
### Testing Subprocess Coverage
219
220
Verifying that subprocess coverage is working correctly:
221
222
```python
223
import subprocess
224
import os
225
226
def test_subprocess_coverage():
227
"""Test that subprocesses get coverage measurement."""
228
229
# Run subprocess that imports and uses covered code
230
script = '''
231
import sys
232
sys.path.insert(0, "src")
233
import mypackage
234
mypackage.function_to_cover()
235
'''
236
237
result = subprocess.run([
238
'python', '-c', script
239
], capture_output=True)
240
241
assert result.returncode == 0
242
# Coverage data will automatically include subprocess execution
243
```
244
245
## Signal Handling
246
247
pytest-cov implements robust signal handling to ensure coverage data is preserved:
248
249
### Signal Handler Management
250
251
```python
252
# Global state for signal handling
253
_previous_handlers = {}
254
_pending_signal = None
255
_cleanup_in_progress = False
256
```
257
258
### Signal Processing
259
260
The signal handling system ensures coverage data is saved even during unexpected process termination:
261
262
1. **Handler Installation**: Previous handlers are saved and custom handler is installed
263
2. **Cleanup Coordination**: Prevents race conditions during cleanup
264
3. **Handler Chaining**: Previous handlers are called after coverage cleanup
265
4. **Graceful Exit**: Processes exit with appropriate codes after cleanup
266
267
### Signal Handler Behavior
268
269
```python
270
def _signal_cleanup_handler(signum, frame):
271
# If already cleaning up, defer signal
272
if _cleanup_in_progress:
273
_pending_signal = (signum, frame)
274
return
275
276
# Clean up coverage
277
cleanup()
278
279
# Call previous handler if exists
280
previous = _previous_handlers.get(signum)
281
if previous and previous != current_handler:
282
previous(signum, frame)
283
elif signum == signal.SIGTERM:
284
os._exit(128 + signum) # Standard exit code
285
elif signum == signal.SIGINT:
286
raise KeyboardInterrupt
287
```