Walk positional-only, keyword-only and annotated parameters in patchedast

CHANGELOG.md

@@ -1,4 +1,5 @@
 # **Upcoming release**
+- Walk positional-only, keyword-only and annotated parameters in patchedast
 
 - #895 patchedast Starred and keyword now consumes their syntactically expected and ** (@lieryan)
 - #896 patchedast cleanup and refactoring (@lieryan)

rope/contrib/autoimport/sqlite.py

@@ -11,7 +11,7 @@
 import warnings
 from collections import OrderedDict
 from concurrent.futures import Future, ProcessPoolExecutor, as_completed
-from datetime import datetime
+from datetime import datetime, timezone
 from hashlib import sha256
 from itertools import chain
 from pathlib import Path
@@ -503,7 +503,7 @@ def clear_cache(self):
             data = (
                 versioning.calculate_version_hash(self.project),
                 json.dumps(versioning.get_version_hash_data(self.project)),
-                datetime.utcnow().isoformat(),
+                datetime.now(timezone.utc).isoformat(),
             )
             assert models.Metadata.columns == [
                 "version_hash",

rope/refactor/extract.py

@@ -485,14 +485,6 @@ def multi_line_conditions(self, info):
             )
         if usefunction._yield_count(node):
             raise RefactoringError("Extracted piece cannot have yield statements.")
-        if not hasattr(
-            ast, "PyCF_ALLOW_TOP_LEVEL_AWAIT"
-        ) and _AsyncStatementFinder.has_errors(extracted):
-            raise RefactoringError(
-                "Extracted piece can only have async/await "
-                "statements if Rope is running on Python "
-                "3.8 or higher"
-            )
         if count == 1 and not usefunction._returns_last(node):
             raise RefactoringError("Return should be the last statement.")
         if info.region != info.lines_region:
@@ -1086,23 +1078,6 @@ def _ClassDef(self, node):
         pass
 
 
-class _AsyncStatementFinder(_BaseErrorFinder):
-    def __init__(self):
-        self.error = False
-
-    def _AsyncFor(self, node):
-        self.error = True
-
-    def _AsyncWith(self, node):
-        self.error = True
-
-    def _FunctionDef(self, node):
-        pass
-
-    def _ClassDef(self, node):
-        pass
-
-
 class _UnbalancedRegionFinder(_BaseErrorFinder):
     """
     Flag an error if we are including the start of a block without the end.

rope/refactor/patchedast.py

@@ -466,22 +466,34 @@ def _AsyncFunctionDef(self, node):
 
     def _arguments(self, node):
         children = []
-        args = list(node.args)
-        defaults = [None] * (len(args) - len(node.defaults)) + list(node.defaults)
-        for index, (arg, default) in enumerate(zip(args, defaults)):
+        positional = list(node.posonlyargs) + list(node.args)
+        defaults = [None] * (len(positional) - len(node.defaults)) + list(node.defaults)
+        for index, (arg, default) in enumerate(zip(positional, defaults)):
             if index > 0:
                 children.append(",")
             self._add_args_to_children(children, arg, default)
-        if node.vararg is not None:
-            if args:
+            if node.posonlyargs and index + 1 == len(node.posonlyargs):
+                children.extend([",", "/"])
+        if node.vararg is not None or node.kwonlyargs:
+            if positional:
                 children.append(",")
-            children.extend(["*", node.vararg.arg])
+            children.append("*")
+            if node.vararg is not None:
+                children.append(self._star_parameter(node.vararg))
+        for arg, default in zip(node.kwonlyargs, node.kw_defaults):
+            children.append(",")
+            self._add_args_to_children(children, arg, default)
         if node.kwarg is not None:
-            if args or node.vararg is not None:
+            if positional or node.vararg is not None or node.kwonlyargs:
                 children.append(",")
-            children.extend(["**", node.kwarg.arg])
+            children.extend(["**", self._star_parameter(node.kwarg)])
         self._handle(node, children)
 
+    @staticmethod
+    def _star_parameter(arg):
+        """An annotated ``*args``/``**kwargs`` is walked as a node, else by name."""
+        return arg if arg.annotation is not None else arg.arg
+
     def _add_args_to_children(self, children, arg, default):
         if isinstance(arg, (list, tuple)):
             self._add_tuple_parameter(children, arg)
@@ -590,7 +602,10 @@ def _NameConstant(self, node):
         self._handle(node, [str(node.value)])
 
     def _arg(self, node):
-        self._handle(node, [node.arg])
+        children = [node.arg]
+        if node.annotation is not None:
+            children.extend([":", node.annotation])
+        self._handle(node, children)
 
     def _Pass(self, node):
         self._handle(node, ["pass"])

ropetest/refactor/patchedasttest.py

@@ -168,6 +168,42 @@ def test_ellipsis(self):
         start = source.index("...")
         checker.check_region("Ellipsis", start, start + len("..."))
 
+    def test_positional_only_and_keyword_only_parameters(self):
+        source = "def f(a, /, b, *, c):\n    pass\n"
+        ast_frag = patchedast.get_patched_ast(source, True)
+        checker = _ResultChecker(self, ast_frag)
+        checker.check_children(
+            "arguments",
+            ["arg", "", ",", " ", "/", "", ",", " ", "arg", "", ",", " ", "*",
+             "", ",", " ", "arg"],
+        )
+
+    def test_bare_star_keyword_only_parameter_with_default(self):
+        source = "def f(*, key=None):\n    pass\n"
+        ast_frag = patchedast.get_patched_ast(source, True)
+        checker = _ResultChecker(self, ast_frag)
+        checker.check_children(
+            "arguments", ["*", "", ",", " ", "arg", "", "=", "", "NameConstant"]
+        )
+
+    def test_annotated_parameters_are_walked_as_nodes(self):
+        source = "def f(a: g(1), b: h(2)) -> None:\n    pass\n"
+        ast_frag = patchedast.get_patched_ast(source, True)
+        checker = _ResultChecker(self, ast_frag)
+        checker.check_children("arguments", ["arg", "", ",", " ", "arg"])
+        start = source.index("g(1)")
+        checker.check_region("Call", start, start + len("g(1)"))
+        self.assertEqual(source, patchedast.write_ast(ast_frag))
+
+    def test_annotated_star_parameters(self):
+        source = "def f(*args: int, **kwargs: str):\n    pass\n"
+        ast_frag = patchedast.get_patched_ast(source, True)
+        checker = _ResultChecker(self, ast_frag)
+        checker.check_children(
+            "arguments", ["*", "", "arg", "", ",", " ", "**", "", "arg"]
+        )
+        self.assertEqual(source, patchedast.write_ast(ast_frag))
+
     def test_ass_name_node(self):
         source = "a = 10\n"
         ast_frag = patchedast.get_patched_ast(source, True)