0
# Results Management
1
2
Efficient caching and management system for mypy execution results. This system ensures mypy runs only once per session while supporting both single-process and xdist parallel execution modes.
3
4
## Capabilities
5
6
### Results Data Structure
7
8
Core data structure for storing and managing mypy execution results.
9
10
```python { .api }
11
@dataclass(frozen=True)
12
class MypyResults:
13
"""
14
Parsed and cached mypy execution results.
15
16
Stores all information from a mypy run including command arguments,
17
output, exit status, and per-file error lines for efficient access
18
by multiple test items.
19
"""
20
21
opts: List[str] # Mypy command options used
22
args: List[str] # File arguments passed to mypy
23
stdout: str # Complete mypy stdout output
24
stderr: str # Complete mypy stderr output
25
status: int # Mypy exit code
26
path_lines: Dict[Optional[Path], List[str]] # Error lines grouped by file
27
28
def dump(self, results_f: IO[bytes]) -> None:
29
"""
30
Cache results to file for persistence across processes.
31
32
Serializes the complete results structure to JSON format
33
for storage in the temporary results file. Path objects are
34
converted to strings for JSON compatibility.
35
36
Parameters:
37
- results_f: Binary file handle for writing cached results
38
"""
39
40
@classmethod
41
def load(cls, results_f: IO[bytes]) -> MypyResults:
42
"""
43
Load cached results from file.
44
45
Deserializes results from JSON format and reconstructs
46
the complete MypyResults structure.
47
48
Parameters:
49
- results_f: Binary file handle containing cached results
50
51
Returns:
52
Fully reconstructed MypyResults instance
53
"""
54
55
@classmethod
56
def from_mypy(
57
cls,
58
paths: List[Path],
59
*,
60
opts: Optional[List[str]] = None
61
) -> MypyResults:
62
"""
63
Execute mypy and create results from output.
64
65
Runs mypy with specified options on the given paths,
66
parses the output, and groups error lines by file.
67
68
Parameters:
69
- paths: List of file paths to check with mypy
70
- opts: Optional mypy command line options
71
72
Returns:
73
MypyResults containing complete execution results
74
"""
75
76
@classmethod
77
def from_session(cls, session: pytest.Session) -> MypyResults:
78
"""
79
Load or generate cached mypy results for a pytest session.
80
81
Uses file locking to ensure thread safety. If cached results
82
exist, loads them; otherwise runs mypy and caches results.
83
84
Parameters:
85
- session: Pytest session containing configuration and items
86
87
Returns:
88
MypyResults for all MypyFileItems in the session
89
"""
90
```
91
92
### Configuration Storage
93
94
Stash mechanism for storing plugin configuration across pytest processes.
95
96
```python { .api }
97
@dataclass(frozen=True)
98
class MypyConfigStash:
99
"""
100
Plugin configuration data stored in pytest.Config stash.
101
102
Maintains the path to cached mypy results and provides
103
serialization for xdist worker communication.
104
"""
105
106
mypy_results_path: Path # Path to temporary results cache file
107
108
@classmethod
109
def from_serialized(cls, serialized: str) -> MypyConfigStash:
110
"""
111
Reconstruct stash from serialized string.
112
113
Used by xdist workers to recreate configuration
114
from controller-provided serialized data.
115
116
Parameters:
117
- serialized: String representation of results path
118
119
Returns:
120
MypyConfigStash instance with reconstructed path
121
"""
122
123
def serialized(self) -> str:
124
"""
125
Serialize stash for transmission to workers.
126
127
Converts the results path to string format for
128
communication between xdist controller and workers.
129
130
Returns:
131
String representation suitable for deserialization
132
"""
133
```
134
135
### Session Integration
136
137
Integration with pytest's session lifecycle for results management.
138
139
```python { .api }
140
# Global stash key for accessing cached configuration
141
stash_key = {
142
"config": pytest.StashKey[MypyConfigStash](),
143
}
144
```
145
146
## File Locking and Concurrency
147
148
The results management system uses file locking to ensure thread safety:
149
150
```python
151
# Example of how results are safely accessed
152
def get_results_safely(session):
153
mypy_results_path = session.config.stash[stash_key["config"]].mypy_results_path
154
with FileLock(str(mypy_results_path) + ".lock"):
155
try:
156
with open(mypy_results_path, mode="rb") as results_f:
157
return MypyResults.load(results_f)
158
except FileNotFoundError:
159
# First access - run mypy and cache results
160
results = MypyResults.from_mypy(collected_paths)
161
with open(mypy_results_path, mode="wb") as results_f:
162
results.dump(results_f)
163
return results
164
```
165
166
## xdist Parallel Execution Support
167
168
Special handling for pytest-xdist parallel execution:
169
170
### Controller Plugin
171
```python { .api }
172
class MypyXdistControllerPlugin:
173
"""
174
Plugin active only on xdist controller processes.
175
176
Responsible for passing configuration to worker processes
177
so they can access the shared results cache.
178
"""
179
180
def pytest_configure_node(self, node: WorkerController) -> None:
181
"""
182
Pass configuration stash to xdist workers.
183
184
Serializes the results path and includes it in worker
185
input so workers can access the shared cache.
186
187
Parameters:
188
- node: xdist worker controller being configured
189
"""
190
```
191
192
### Controller Plugin
193
```python { .api }
194
class MypyControllerPlugin:
195
"""
196
Plugin for main/controller processes (not xdist workers).
197
198
Handles terminal reporting and cleanup of temporary files.
199
"""
200
201
def pytest_terminal_summary(
202
self,
203
terminalreporter: TerminalReporter,
204
config: pytest.Config
205
) -> None:
206
"""
207
Report mypy results in terminal summary.
208
209
Displays mypy output including errors, notes, and status
210
information in the pytest terminal report.
211
212
Parameters:
213
- terminalreporter: Pytest terminal reporter instance
214
- config: Pytest configuration object
215
"""
216
217
def pytest_unconfigure(self, config: pytest.Config) -> None:
218
"""
219
Clean up temporary mypy results file.
220
221
Removes the cached results file when pytest session ends.
222
223
Parameters:
224
- config: Pytest configuration object
225
"""
226
```
227
228
## Usage Examples
229
230
### Accessing Results from Test Items
231
```python
232
class CustomMypyFileItem(MypyFileItem):
233
def runtest(self):
234
# Access cached mypy results
235
results = MypyResults.from_session(self.session)
236
237
# Get errors for this specific file
238
file_errors = results.path_lines.get(self.path.resolve(), [])
239
240
# Custom error processing
241
if file_errors:
242
print(f"Found {len(file_errors)} errors in {self.path}")
243
for error in file_errors:
244
print(f" {error}")
245
```
246
247
### Custom Results Processing
248
```python
249
def analyze_mypy_results(session):
250
"""Analyze mypy results for custom reporting."""
251
results = MypyResults.from_session(session)
252
253
total_errors = sum(len(lines) for lines in results.path_lines.values())
254
files_with_errors = len([lines for lines in results.path_lines.values() if lines])
255
256
print(f"Mypy found {total_errors} issues across {files_with_errors} files")
257
print(f"Exit status: {results.status}")
258
```