Repository: asottile/dead

Issue:
Detect unused arguments
A related feature that could be useful would be to detect "dead arguments," or function arguments not used by the function body.

Here is an example of a dead argument found in pip: https://github.com/pypa/pip/pull/7015

Discussion:

Patch:
diff --git a/dead.py b/dead.py
index 7879cc5..019ee8e 100644
--- a/dead.py
+++ b/dead.py
@@ -8,6 +8,7 @@ import subprocess
 import tokenize
 from typing import DefaultDict
 from typing import Generator
+from typing import List
 from typing import NewType
 from typing import Optional
 from typing import Pattern
@@ -28,13 +29,19 @@ TYPE_FUNC_RE = re.compile(r'^(\(.*?\))\s*->\s*(.*)$')
 DISABLE_COMMENT_RE = re.compile(r'\bdead\s*:\s*disable')
 
 
-class Visitor(ast.NodeVisitor):
+class Scope:
     def __init__(self) -> None:
-        self.filename = ''
-        self.is_test = False
         self.reads: UsageMap = collections.defaultdict(set)
         self.defines: UsageMap = collections.defaultdict(set)
         self.reads_tests: UsageMap = collections.defaultdict(set)
+
+
+class Visitor(ast.NodeVisitor):
+    def __init__(self) -> None:
+        self.filename = ''
+        self.is_test = False
+        self.previous_scopes: List[Scope] = []
+        self.scopes = [Scope()]
         self.disabled: Set[FileLine] = set()
 
     @contextlib.contextmanager
@@ -52,9 +59,13 @@ class Visitor(ast.NodeVisitor):
             self.filename = orig_filename
             self.is_test = orig_is_test
 
-    @property
-    def reads_target(self) -> UsageMap:
-        return self.reads_tests if self.is_test else self.reads
+    @contextlib.contextmanager
+    def scope(self) -> Generator[None, None, None]:
+        self.scopes.append(Scope())
+        try:
+            yield
+        finally:
+            self.previous_scopes.append(self.scopes.pop())
 
     def _file_line(self, filename: str, line: int) -> FileLine:
         return FileLine(f'{filename}:{line}')
@@ -62,31 +73,77 @@ class Visitor(ast.NodeVisitor):
     def definition_str(self, node: ast.AST) -> FileLine:
         return self._file_line(self.filename, node.lineno)
 
+    def define(self, name: str, node: ast.AST) -> None:
+        if not self.is_test:
+            self.scopes[-1].defines[name].add(self.definition_str(node))
+
+    def read(self, name: str, node: ast.AST) -> None:
+        for scope in self.scopes:
+            if self.is_test:
+                scope.reads_tests[name].add(self.definition_str(node))
+            else:
+                scope.reads[name].add(self.definition_str(node))
+
     def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
         for name in node.names:
-            self.reads_target[name.name].add(self.definition_str(node))
-            if not self.is_test and name.asname:
-                self.defines[name.asname].add(self.definition_str(node))
+            self.read(name.name, node)
+            if name.asname:
+                self.define(name.asname, node)
 
         self.generic_visit(node)
 
     def visit_ClassDef(self, node: ast.ClassDef) -> None:
-        if not self.is_test:
-            self.defines[node.name].add(self.definition_str(node))
+        self.define(node.name, node)
         self.generic_visit(node)
 
+    def _is_stub_function(self, node: ast.FunctionDef) -> bool:
+        for stmt in node.body:
+            if (
+                    isinstance(stmt, ast.Expr) and
+                    isinstance(stmt.value, (ast.Str, ast.Ellipsis))
+            ):
+                continue  # docstring or ...
+            elif isinstance(stmt, ast.Pass):
+                continue  # pass
+            elif (
+                    isinstance(stmt, ast.Raise) and
+                    stmt.cause is None and (
+                        (
+                            isinstance(stmt.exc, ast.Name) and
+                            stmt.exc.id == 'NotImplementedError'
+                        ) or (
+                            isinstance(stmt.exc, ast.Call) and
+                            isinstance(stmt.exc.func, ast.Name) and
+                            stmt.exc.func.id == 'NotImplementedError'
+                        )
+                    )
+            ):
+                continue  # raise NotImplementedError
+            else:
+                return False
+        else:
+            return True
+
     def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
-        if not self.is_test:
-            self.defines[node.name].add(self.definition_str(node))
-        self.generic_visit(node)
+        self.define(node.name, node)
+        with self.scope():
+            if not self._is_stub_function(node):
+                for arg in (
+                        *node.args.args,
+                        node.args.vararg,
+                        *node.args.kwonlyargs,
+                        node.args.kwarg,
+                ):
+                    if arg is not None:
+                        self.define(arg.arg, arg)
+            self.generic_visit(node)
 
     visit_AsyncFunctionDef = visit_FunctionDef
 
     def visit_Assign(self, node: ast.Assign) -> None:
-        if not self.is_test:
-            for target in node.targets:
-                if isinstance(target, ast.Name):
-                    self.defines[target.id].add(self.definition_str(node))
+        for target in node.targets:
+            if isinstance(target, ast.Name):
+                self.define(target.id, node)
 
         if (
                 len(node.targets) == 1 and
@@ -96,7 +153,7 @@ class Visitor(ast.NodeVisitor):
         ):
             for elt in node.value.elts:
                 if isinstance(elt, ast.Str):
-                    self.reads_target[elt.s].add(self.definition_str(elt))
+                    self.read(elt.s, elt)
 
         self.generic_visit(node)
 
@@ -104,13 +161,13 @@ class Visitor(ast.NodeVisitor):
 
     def visit_Name(self, node: ast.Name) -> None:
         if isinstance(node.ctx, ast.Load):
-            self.reads_target[node.id].add(self.definition_str(node))
+            self.read(node.id, node)
 
         self.generic_visit(node)
 
     def visit_Attribute(self, node: ast.Attribute) -> None:
         if isinstance(node.ctx, ast.Load):
-            self.reads_target[node.attr].add(self.definition_str(node))
+            self.read(node.attr, node)
 
         self.generic_visit(node)
 
@@ -176,8 +233,7 @@ class ParsesEntryPoints(ast.NodeVisitor):
     def visit_Str(self, node: ast.Str) -> None:
         match = ENTRYPOINT_RE.match(node.s)
         if match:
-            location = self.visitor.definition_str(node)
-            self.visitor.reads[match.group(1)].add(location)
+            self.visitor.read(match.group(1), node)
         self.generic_visit(node)
 
 
@@ -236,25 +292,29 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
 
     retv = 0
 
+    visitor.previous_scopes.append(visitor.scopes.pop())
     unused_ignores = visitor.disabled.copy()
-    for k, v in visitor.defines.items():
-        if k not in visitor.reads:
-            unused_ignores.difference_update(v)
-            v = v - visitor.disabled
-
-        if k.startswith('__') and k.endswith('__'):
-            pass  # skip magic methods, probably an interface
-        elif k not in visitor.reads and not v - visitor.disabled:
-            pass  # all references disabled
-        elif k not in visitor.reads and k not in visitor.reads_tests:
-            print(f'{k} is never read, defined in {", ".join(sorted(v))}')
-            retv = 1
-        elif k not in visitor.reads:
-            print(
-                f'{k} is only referenced in tests, '
-                f'defined in {", ".join(sorted(v))}',
-            )
-            retv = 1
+    for scope in visitor.previous_scopes:
+        for k, v in scope.defines.items():
+            if k not in scope.reads:
+                unused_ignores.difference_update(v)
+                v = v - visitor.disabled
+
+            if k.startswith('__') and k.endswith('__'):
+                pass  # skip magic methods, probably an interface
+            elif k in {'cls', 'self'}:
+                pass  # ignore conventional cls / self
+            elif k not in scope.reads and not v:
+                pass  # all references disabled
+            elif k not in scope.reads and k not in scope.reads_tests:
+                print(f'{k} is never read, defined in {", ".join(sorted(v))}')
+                retv = 1
+            elif k not in scope.reads:
+                print(
+                    f'{k} is only referenced in tests, '
+                    f'defined in {", ".join(sorted(v))}',
+                )
+                retv = 1
 
     if unused_ignores:
         for ignore in sorted(unused_ignores):


Test patch:
diff --git a/tests/dead_test.py b/tests/dead_test.py
index 9ee3a09..f8b63ba 100644
--- a/tests/dead_test.py
+++ b/tests/dead_test.py
@@ -140,3 +140,71 @@ def test_partially_disabled(git_dir, capsys):
     assert dead.main(())
     out, _ = capsys.readouterr()
     assert out == 'x is never read, defined in f.py:1, f.py:3\n'
+
+
+def test_unused_argument(git_dir, capsys):
+    git_dir.join('f.py').write('def f(a, *b, c, **d): return 1\nf')
+    subprocess.check_call(('git', 'add', '.'))
+    assert dead.main(())
+    out, _ = capsys.readouterr()
+    assert out == (
+        'a is never read, defined in f.py:1\n'
+        'b is never read, defined in f.py:1\n'
+        'c is never read, defined in f.py:1\n'
+        'd is never read, defined in f.py:1\n'
+    )
+
+
+def test_unused_argument_in_scope(git_dir, capsys):
+    git_dir.join('f.py').write('def f(g): return 1\ndef g(): pass\ng\nf\n')
+    subprocess.check_call(('git', 'add', '.'))
+    assert dead.main(())
+    out, _ = capsys.readouterr()
+    assert out == 'g is never read, defined in f.py:1\n'
+
+
+def test_using_an_argument(git_dir):
+    git_dir.join('f.py').write('def f(g): return g\nf')
+    subprocess.check_call(('git', 'add', '.'))
+    assert not dead.main(())
+
+
+def test_ignore_unused_arguments_stubs(git_dir):
+    git_dir.join('f.py').write(
+        'import abc\n'
+        'from typing import overload\n'
+        'class C:\n'
+        '    @abc.abstractmethod\n'
+        '    def func(self, arg1):\n'
+        '        pass\n'
+        'def func2(arg2):\n'
+        '    pass\n'
+        'def func3(arg3):\n'
+        '    pass\n'
+        '@overload\n'
+        'def func4(arg4):\n'
+        '    ...\n'
+        'def func5(arg5):\n'
+        '    """docstring but trivial"""\n'
+        'def func6(arg6):\n'
+        '    raise NotImplementedError\n'
+        'def func7(arg7):\n'
+        '    raise NotImplementedError()\n'
+        'def func8(arg8):\n'
+        '    """docstring plus raise"""\n'
+        '    raise NotImplementedError()\n'
+        'C.func, func2, func3, func4, func5, func6, func7, func8\n',
+    )
+    subprocess.check_call(('git', 'add', '.'))
+    assert not dead.main(())
+
+
+def test_ignored_arguments(git_dir):
+    git_dir.join('f.py').write(
+        'class C:\n'
+        '    @classmethod\n'
+        '    def f(cls): return 1\n'  # allow conventional `cls` method
+        'C.f',
+    )
+    subprocess.check_call(('git', 'add', '.'))
+    assert not dead.main(())