Perform one fine-grained incremental build step (after some file updates/deletions). Return (mypy output, triggered targets).
(
self,
operations: list[UpdateFile | DeleteFile],
server: Server,
options: Options,
build_options: Options,
testcase: DataDrivenTestCase,
main_src: str,
step: int,
num_regular_incremental_steps: int,
)
| 198 | return 1 |
| 199 | |
| 200 | def perform_step( |
| 201 | self, |
| 202 | operations: list[UpdateFile | DeleteFile], |
| 203 | server: Server, |
| 204 | options: Options, |
| 205 | build_options: Options, |
| 206 | testcase: DataDrivenTestCase, |
| 207 | main_src: str, |
| 208 | step: int, |
| 209 | num_regular_incremental_steps: int, |
| 210 | ) -> tuple[list[str], list[list[str]]]: |
| 211 | """Perform one fine-grained incremental build step (after some file updates/deletions). |
| 212 | |
| 213 | Return (mypy output, triggered targets). |
| 214 | """ |
| 215 | perform_file_operations(operations) |
| 216 | sources = self.parse_sources(main_src, step, options) |
| 217 | |
| 218 | if step <= num_regular_incremental_steps: |
| 219 | new_messages = self.build(build_options, sources) |
| 220 | else: |
| 221 | new_messages = self.run_check(server, sources) |
| 222 | |
| 223 | updated: list[str] = [] |
| 224 | changed: list[str] = [] |
| 225 | targets: list[str] = [] |
| 226 | triggered = [] |
| 227 | if server.fine_grained_manager: |
| 228 | if CHECK_CONSISTENCY: |
| 229 | check_consistency(server.fine_grained_manager) |
| 230 | triggered.append(server.fine_grained_manager.triggered) |
| 231 | |
| 232 | updated = server.fine_grained_manager.updated_modules |
| 233 | changed = [mod for mod, file in server.fine_grained_manager.changed_modules] |
| 234 | targets = server.fine_grained_manager.processed_targets |
| 235 | |
| 236 | expected_stale = testcase.expected_stale_modules.get(step - 1) |
| 237 | if expected_stale is not None: |
| 238 | assert_module_equivalence("stale" + str(step - 1), expected_stale, changed) |
| 239 | |
| 240 | expected_rechecked = testcase.expected_rechecked_modules.get(step - 1) |
| 241 | if expected_rechecked is not None: |
| 242 | assert_module_equivalence("rechecked" + str(step - 1), expected_rechecked, updated) |
| 243 | |
| 244 | expected = testcase.expected_fine_grained_targets.get(step) |
| 245 | if expected: |
| 246 | assert_target_equivalence("targets" + str(step), expected, targets) |
| 247 | |
| 248 | new_messages = normalize_messages(new_messages) |
| 249 | |
| 250 | a = new_messages |
| 251 | assert testcase.tmpdir is not None |
| 252 | a.extend(self.maybe_suggest(step, server, main_src, testcase.tmpdir)) |
| 253 | a.extend(self.maybe_inspect(step, server, main_src)) |
| 254 | |
| 255 | return a, triggered |
| 256 | |
| 257 | def parse_sources( |
no test coverage detected