(
tree: ast.Module, patches: Patches
)
| 241 | |
| 242 | |
| 243 | def _iter_patch_lines( |
| 244 | tree: ast.Module, patches: Patches |
| 245 | ) -> "Iterator[tuple[int, str]]": |
| 246 | import sys |
| 247 | |
| 248 | # Build cache of all classes (for Phase 2 to find classes without methods) |
| 249 | cache = {} |
| 250 | # Build per-class set of async method names (for Phase 2 to generate correct override) |
| 251 | async_methods: dict[str, set[str]] = {} |
| 252 | # Track class bases for inherited async method lookup |
| 253 | class_bases: dict[str, list[str]] = {} |
| 254 | all_classes = {node.name for node in tree.body if isinstance(node, ast.ClassDef)} |
| 255 | for node in tree.body: |
| 256 | if isinstance(node, ast.ClassDef): |
| 257 | cache[node.name] = node.end_lineno |
| 258 | class_bases[node.name] = [ |
| 259 | base.id |
| 260 | for base in node.bases |
| 261 | if isinstance(base, ast.Name) and base.id in all_classes |
| 262 | ] |
| 263 | cls_async: set[str] = set() |
| 264 | for item in node.body: |
| 265 | if isinstance(item, ast.AsyncFunctionDef): |
| 266 | cls_async.add(item.name) |
| 267 | if cls_async: |
| 268 | async_methods[node.name] = cls_async |
| 269 | |
| 270 | # Phase 1: Iterate and mark existing tests |
| 271 | for cls_node, fn_node in iter_tests(tree): |
| 272 | specs = patches.get(cls_node.name, {}).pop(fn_node.name, None) |
| 273 | if not specs: |
| 274 | continue |
| 275 | |
| 276 | lineno = min( |
| 277 | (dec_node.lineno for dec_node in fn_node.decorator_list), |
| 278 | default=fn_node.lineno, |
| 279 | ) |
| 280 | indent = " " * fn_node.col_offset |
| 281 | patch_lines = "\n".join(spec.as_decorator() for spec in specs) |
| 282 | yield (lineno - 1, textwrap.indent(patch_lines, indent)) |
| 283 | |
| 284 | # Phase 2: Iterate and mark inherited tests |
| 285 | for cls_name, tests in sorted(patches.items()): |
| 286 | lineno = cache.get(cls_name) |
| 287 | if not lineno: |
| 288 | print(f"WARNING: {cls_name} does not exist in remote file", file=sys.stderr) |
| 289 | continue |
| 290 | |
| 291 | for test_name, specs in sorted(tests.items()): |
| 292 | decorators = "\n".join(spec.as_decorator() for spec in specs) |
| 293 | # Check current class and ancestors for async method |
| 294 | is_async = False |
| 295 | queue = [cls_name] |
| 296 | visited: set[str] = set() |
| 297 | while queue: |
| 298 | cur = queue.pop(0) |
| 299 | if cur in visited: |
| 300 | continue |
no test coverage detected