beam: AST name-shadowing linter with tests

wizard-work/beam/shadow_lint.py

@@ -0,0 +1,259 @@
+"""AST-based detector for name shadowing in Python source.
+
+A 'shadow' occurs when a name bound in an inner scope reuses a name
+already bound in an enclosing scope, hiding the outer binding. This is a
+common source of subtle bugs (e.g. loop variable clobbering an outer
+variable, or a comprehension leaking its target name in older Pythons).
+
+Only *binding* occurrences are considered (assignments, function params,
+import names, for/comprehension targets, with-as targets, except-as).
+"""
+import ast
+import sys
+from dataclasses import dataclass
+from typing import List, Set, Tuple
+
+
+@dataclass(frozen=True)
+class Shadow:
+    name: str
+    outer_line: int
+    inner_line: int
+    kind: str  # 'function-param', 'assignment', 'import', 'loop-target', 'comprehension', 'with-as', 'except-as'
+
+
+class Scope:
+    def __init__(self, parent=None, kind="module"):
+        self.parent = parent
+        self.kind = kind
+        self.bindings: Set[str] = set()
+        self.binding_lines: dict = {}
+
+    def is_bound_outer(self, name: str) -> bool:
+        s = self.parent
+        while s is not None:
+            if name in s.bindings:
+                return True
+            s = s.parent
+        return False
+
+    def outer_line_for(self, name: str) -> int:
+        s = self.parent
+        while s is not None:
+            if name in s.bindings:
+                return s.binding_lines.get(name, 0)
+            s = s.parent
+        return 0
+
+    def record(self, name: str, line: int = 0):
+        self.bindings.add(name)
+        if name not in self.binding_lines:
+            self.binding_lines[name] = line
+
+
+def _binding_names(target) -> List[str]:
+    """Collect names bound by an assignment target / pattern."""
+    out = []
+    if isinstance(target, ast.Name):
+        out.append(target.id)
+    elif isinstance(target, (ast.Tuple, ast.List)):
+        for elt in target.elts:
+            out.extend(_binding_names(elt))
+    elif isinstance(target, ast.Starred):
+        out.extend(_binding_names(target.value))
+    return out
+
+
+class ShadowFinder(ast.NodeVisitor):
+    def __init__(self):
+        self.shadows: List[Shadow] = []
+        self.scope: Scope = Scope(kind="module")
+
+    def _enter_scope(self, kind, node_line):
+        new = Scope(parent=self.scope, kind=kind)
+        self._scope_stack_push(new)
+        return new
+
+    def _scope_stack_push(self, s):
+        self._prev_scope = self.scope
+        self.scope = s
+
+    def _leave_scope(self):
+        self.scope = self._prev_scope
+
+    # --- function defs create a new scope; params are bindings ---
+    def visit_FunctionDef(self, node):
+        # params bind in the new function scope
+        new_scope = Scope(parent=self.scope, kind="function")
+        old = self.scope
+        self.scope = new_scope
+        for arg in (*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs):
+            if arg.arg in ("",):
+                continue
+            if self.scope.parent.is_bound_outer(arg.arg) or arg.arg in self.scope.parent.bindings:
+                # shadow check against enclosing
+                if arg.arg in self._enclosing_bindings():
+                    self.shadows.append(Shadow(arg.arg, self.scope.outer_line_for(arg.arg), node.lineno, "function-param"))
+            new_scope.record(arg.arg)
+        if node.args.vararg:
+            new_scope.record(node.args.vararg.arg)
+        if node.args.kwarg:
+            new_scope.record(node.args.kwarg.arg)
+        self.generic_visit(node)
+        self.scope = old
+
+    visit_AsyncFunctionDef = visit_FunctionDef
+
+    def _enclosing_bindings(self) -> Set[str]:
+        s = self.scope.parent
+        acc: Set[str] = set()
+        while s is not None:
+            acc |= s.bindings
+            s = s.parent
+        return acc
+
+    def _outer_line(self, name: str) -> int:
+        # find the line where name was first bound in an enclosing scope
+        s = self.scope.parent
+        while s is not None:
+            # we don't store lines; approximate by searching AST? Instead store line map.
+            s = s.parent
+        return 0
+
+    # We need outer line. Rework: track binding lines in scope.
+    # (handled below by patching Scope to store lines)
+
+    def visit_Assign(self, node):
+        for t in node.targets:
+            for name in _binding_names(t):
+                if name in self._enclosing_bindings():
+                    self.shadows.append(Shadow(name, self.scope.outer_line_for(name), node.lineno, "assignment"))
+                self.scope.bindings.add(name)
+        self.generic_visit(node)
+
+    def visit_AnnAssign(self, node):
+        if isinstance(node.target, ast.Name):
+            name = node.target.id
+            if name in self._enclosing_bindings():
+                self.shadows.append(Shadow(name, self.scope.outer_line_for(name), node.lineno, "assignment"))
+            self.scope.bindings.add(name)
+        self.generic_visit(node)
+
+    def visit_AugAssign(self, node):
+        # aug-assign to existing name is not a new binding in our model; skip
+        self.generic_visit(node)
+
+    def visit_Import(self, node):
+        for alias in node.names:
+            name = alias.asname if alias.asname else alias.name.split(".")[0]
+            if name in self._enclosing_bindings():
+                self.shadows.append(Shadow(name, self.scope.outer_line_for(name), node.lineno, "import"))
+            self.scope.bindings.add(name)
+        self.generic_visit(node)
+
+    def visit_ImportFrom(self, node):
+        for alias in node.names:
+            if alias.name == "*":
+                continue
+            name = alias.asname if alias.asname else alias.name
+            if name in self._enclosing_bindings():
+                self.shadows.append(Shadow(name, self.scope.outer_line_for(name), node.lineno, "import"))
+            self.scope.bindings.add(name)
+        self.generic_visit(node)
+
+    def visit_For(self, node):
+        for name in _binding_names(node.target):
+            if name in self._enclosing_bindings():
+                self.shadows.append(Shadow(name, self.scope.outer_line_for(name), node.lineno, "loop-target"))
+            self.scope.bindings.add(name)
+        # body in same scope
+        self.generic_visit(node)
+
+    def visit_AsyncFor(self, node):
+        self.visit_For(node)
+
+    def visit_With(self, node):
+        for item in node.items:
+            if item.optional_vars is not None:
+                for name in _binding_names(item.optional_vars):
+                    if name in self._enclosing_bindings():
+                        self.shadows.append(Shadow(name, self.scope.outer_line_for(name), node.lineno, "with-as"))
+                    self.scope.bindings.add(name)
+        self.generic_visit(node)
+
+    def visit_ExceptHandler(self, node):
+        if node.name:
+            if node.name in self._enclosing_bindings():
+                self.shadows.append(Shadow(node.name, self.scope.outer_line_for(node.name), node.lineno, "except-as"))
+            self.scope.bindings.add(node.name)
+        self.generic_visit(node)
+
+    # comprehensions create their own scope
+    def visit_ListComp(self, node):
+        self._comp_scope(node)
+    def visit_SetComp(self, node):
+        self._comp_scope(node)
+    def visit_GeneratorExp(self, node):
+        self._comp_scope(node)
+    def visit_DictComp(self, node):
+        self._comp_scope(node)
+
+    def _comp_scope(self, node):
+        new_scope = Scope(parent=self.scope, kind="comprehension")
+        old = self.scope
+        self.scope = new_scope
+        # targets of all generators
+        for gen in node.generators:
+            for name in _binding_names(gen.target):
+                if name in self._enclosing_bindings() or name in self.scope.parent.bindings:
+                    # shadow relative to enclosing (parent of comp scope)
+                    if name in self._enclosing_bindings():
+                        self.shadows.append(Shadow(name, self.scope.outer_line_for(name), gen.target.lineno if hasattr(gen.target, "lineno") else node.lineno, "comprehension"))
+                new_scope.bindings.add(name)
+        # visit body with this scope
+        # for ListComp/SetComp/GenExp: elt; DictComp: key/value
+        if isinstance(node, ast.DictComp):
+            self.visit(node.key)
+            self.visit(node.value)
+        elif isinstance(node, ast.GeneratorExp):
+            self.visit(node.elt)
+        else:
+            self.visit(node.elt)
+        for gen in node.generators:
+            self.visit(gen.iter)
+            # conditions
+            for c in gen.ifs:
+                self.visit(c)
+        self.scope = old
+
+
+def find_shadows(source: str) -> List[Shadow]:
+    tree = ast.parse(source)
+    finder = ShadowFinder()
+    finder.visit(tree)
+    return finder.shadows
+
+
+def main(argv=None):
+    import argparse
+    p = argparse.ArgumentParser(description="Detect name shadowing in Python files.")
+    p.add_argument("paths", nargs="+")
+    args = p.parse_args(argv)
+    rc = 0
+    for path in args.paths:
+        try:
+            src = open(path).read()
+        except OSError as e:
+            print(f"{path}: {e}", file=sys.stderr)
+            rc = 1
+            continue
+        sh = find_shadows(src)
+        for s in sh:
+            print(f"{path}:{s.inner_line}: shadowed name {s.name!r} ({s.kind})")
+        if sh:
+            rc = 1
+    return rc
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

wizard-work/beam/test_shadow.py

@@ -0,0 +1,44 @@
+import subprocess, sys, os
+HERE = os.path.dirname(os.path.abspath(__file__))
+SCRIPT = os.path.join(HERE, "shadow_lint.py")
+
+def run(src):
+    p = "/tmp/scratchpad/test_shadow.py"
+    os.makedirs("/tmp/scratchpad", exist_ok=True)
+    open(p,"w").write(src)
+    r = subprocess.run([sys.executable, SCRIPT, p], capture_output=True, text=True)
+    return r.returncode, r.stdout + r.stderr
+
+def test_param_shadow():
+    rc, out = run("x = 1\ndef f(x):\n    return x\n")
+    assert rc == 1
+    assert "function-param" in out and "'x'" in out
+
+def test_no_shadow():
+    rc, out = run("x = 1\ndef f(y):\n    return x + y\n")
+    assert rc == 0, out
+
+def test_import_shadow():
+    rc, out = run("import os\ndef f():\n    import os\n    return os\n")
+    assert rc == 1 and "import" in out
+
+def test_loop_target_shadow():
+    rc, out = run("def outer():\n    i = 0\n    def inner():\n        for i in range(3):\n            pass\n")
+    assert rc == 1 and "loop-target" in out
+
+def test_comprehension_shadow():
+    rc, out = run("x = 1\ny = [x for x in range(3)]\n")
+    assert rc == 1 and "comprehension" in out
+
+def test_with_as_shadow():
+    rc, out = run("def outer():\n    f = 1\n    def inner():\n        with open('/dev/null') as f:\n            pass\n")
+    assert rc == 1 and "with-as" in out
+
+if __name__ == "__main__":
+    for name in [n for n in globals() if n.startswith("test_")]:
+        globals()[name]()
+        print("PASS", name)
+    print("ALL PASS")
+
+if __name__ == "__main__":
+    pass