AST visitor that finds subprocess.run() calls with output capture.
| 34 | |
| 35 | |
| 36 | class SubprocessCaptureVisitor(ast.NodeVisitor): |
| 37 | """AST visitor that finds subprocess.run() calls with output capture.""" |
| 38 | |
| 39 | def __init__(self, source_lines: list[str] | None = None) -> None: |
| 40 | self.violations: list[tuple[int, str]] = [] |
| 41 | self.source_lines = source_lines or [] |
| 42 | |
| 43 | def visit_Call(self, node: ast.Call) -> None: # noqa: N802 |
| 44 | if not self._is_subprocess_run(node): |
| 45 | self.generic_visit(node) |
| 46 | return |
| 47 | |
| 48 | # Skip if this call uses input= (stdin piping is not a deadlock risk) |
| 49 | if any(kw.arg == "input" for kw in node.keywords): |
| 50 | self.generic_visit(node) |
| 51 | return |
| 52 | |
| 53 | # Skip if this line has a noqa comment |
| 54 | if self._has_noqa(node.lineno): |
| 55 | self.generic_visit(node) |
| 56 | return |
| 57 | |
| 58 | # Check keyword arguments for capture patterns |
| 59 | for kw in node.keywords: |
| 60 | if kw.arg == "capture_output" and self._is_true(kw.value): |
| 61 | self.violations.append((node.lineno, _VIOLATION_MSG_CAPTURE)) |
| 62 | break |
| 63 | if kw.arg in ("stdout", "stderr") and self._is_pipe(kw.value): |
| 64 | self.violations.append((node.lineno, _VIOLATION_MSG_PIPE)) |
| 65 | break |
| 66 | |
| 67 | self.generic_visit(node) |
| 68 | |
| 69 | def _is_subprocess_run(self, node: ast.Call) -> bool: |
| 70 | """Check if the call is subprocess.run(...).""" |
| 71 | if isinstance(node.func, ast.Attribute) and node.func.attr == "run": |
| 72 | if ( |
| 73 | isinstance(node.func.value, ast.Name) |
| 74 | and node.func.value.id == "subprocess" |
| 75 | ): |
| 76 | return True |
| 77 | return False |
| 78 | |
| 79 | def _is_true(self, node: ast.expr) -> bool: |
| 80 | """Check if node is the literal True.""" |
| 81 | return isinstance(node, ast.Constant) and node.value is True |
| 82 | |
| 83 | def _is_pipe(self, node: ast.expr) -> bool: |
| 84 | """Check if node is subprocess.PIPE.""" |
| 85 | if isinstance(node, ast.Attribute) and node.attr == "PIPE": |
| 86 | if isinstance(node.value, ast.Name) and node.value.id == "subprocess": |
| 87 | return True |
| 88 | return False |
| 89 | |
| 90 | def _has_noqa(self, lineno: int) -> bool: |
| 91 | """Check if a line has a noqa comment.""" |
| 92 | if not self.source_lines or lineno < 1 or lineno > len(self.source_lines): |
| 93 | return False |