[DEV-16665] Don't allow tasks to be defined without args and kwargs

- Fixes for tasks with types

flake8_routable.py

@@ -427,6 +427,7 @@ def task_args_kwargs_and_priority(self) -> None:
         kwargs_found = False
         last_star = -1
         last_star_star = -1
+        last_close_paren = -1
 
         for i, (token_type, token_str, start_indices, end_indices, line) in enumerate(self._file_tokens):
             # Start of a contextmanager
@@ -436,11 +437,13 @@ def task_args_kwargs_and_priority(self) -> None:
             elif handler_start and token_type == tokenize.NAME and token_str == "shared_task":
                 in_task_definition = True
 
-            # Track * and ** positions
+            # Track *, **, and ) positions
             elif in_task_definition and token_type == tokenize.OP and token_str == "*":
                 last_star = i
             elif in_task_definition and token_type == tokenize.OP and token_str == "**":
                 last_star_star = i
+            elif in_task_definition and token_type == tokenize.OP and token_str == ")":
+                last_close_paren = i
 
             # Look for *args and **kwargs
             elif in_task_definition and token_type == tokenize.NAME and token_str == "args" and last_star == i - 1:
@@ -455,7 +458,7 @@ def task_args_kwargs_and_priority(self) -> None:
                 self.errors.append((*start_indices, ROU113))
 
             # End of method, are *args or **kwargs missing?
-            elif token_type == tokenize.OP and token_str == ":":
+            elif token_type == tokenize.OP and token_str == ":" and last_close_paren == i - 1:
                 if in_task_definition and (not args_found or not kwargs_found):
                     self.errors.append((*start_indices, ROU112))
 

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = flake8_routable
-version = 0.1.5
+version = 0.1.6
 
 [options]
 py_modules = flake8_routable

tests/test_flake8_routable.py

@@ -659,6 +659,54 @@ def test_with_invalid_comment(self, comment):
 
 
 class TestROU112:
+    TASK_DECORATOR_WITH_PARAMS_WITH_ARGS_KWARGS = (
+        "\n"
+        "@shared_task(autoretry_for=(Exception,), default_retry_delay=20)\n"
+        "def task_method(\n"
+        "    field_1,\n"
+        "    field_2,\n"
+        "    *args,\n"
+        "    **kwargs,\n"
+        "):"
+        "    pass\n"
+        "\n"
+    )
+
+    TASK_DECORATOR_WITH_PARAMS_MISSING_ARGS_KWARGS = (
+        "\n"
+        "@shared_task(autoretry_for=(Exception,), default_retry_delay=20)\n"
+        "def task_method(\n"
+        "    field_1,\n"
+        "    field_2,\n"
+        "):"
+        "    pass\n"
+        "\n"
+    )
+
+    TASK_WITH_TYPES_WITH_PARAMS_WITH_ARGS_KWARGS = (
+        "\n"
+        "@shared_task\n"
+        "def task_method(\n"
+        "    field_1: str,\n"
+        "    field_2: int,\n"
+        "    *args,\n"
+        "    **kwargs,\n"
+        "):"
+        "    pass\n"
+        "\n"
+    )
+
+    TASK_WITH_TYPES_WITH_PARAMS_MISSING_ARGS_KWARGS = (
+        "\n"
+        "@shared_task\n"
+        "def task_method(\n"
+        "    field_1: str,\n"
+        "    field_2: int,\n"
+        "):"
+        "    pass\n"
+        "\n"
+    )
+
     TASK_MULTILINE_WITH_ARGS_KWARGS = (
         "\n"
         "@shared_task\n"
@@ -717,6 +765,8 @@ class TestROU112:
     @pytest.mark.parametrize(
         "code",
         (
+            TASK_DECORATOR_WITH_PARAMS_WITH_ARGS_KWARGS,
+            TASK_WITH_TYPES_WITH_PARAMS_WITH_ARGS_KWARGS,
             TASK_MULTILINE_WITH_ARGS_KWARGS,
             TASK_SINGLELINE_WITH_ARGS_KWARGS,
         ),
@@ -728,6 +778,8 @@ def test_correct_signature(self, code):
     @pytest.mark.parametrize(
         ("code", "location"),
         (
+            (TASK_DECORATOR_WITH_PARAMS_MISSING_ARGS_KWARGS, "6:1"),
+            (TASK_WITH_TYPES_WITH_PARAMS_MISSING_ARGS_KWARGS, "6:1"),
             (TASK_MULTILINE_MISSING_ARGS, "7:1"),
             (TASK_SINGLELINE_MISSING_ARGS, "3:43"),
             (TASK_MULTILINE_MISSING_KWARGS, "7:1"),