Handle case when options.retry.on is empty

src/aiotaskq/exceptions.py

@@ -19,3 +19,7 @@ class ConcurrencyTypeNotSupported(Exception):
 
 class InvalidArgument(Exception):
     """A task is applied with invalid arguments."""
+
+
+class InvalidRetryOptions(Exception):
+    """A task is defined with invalid retry options."""

src/aiotaskq/interfaces.py

@@ -168,19 +168,23 @@ class RetryOptions(t.TypedDict):
     max_retries int | None: The number times to keep retrying the execution of the task
                             until the task executes successfully. Counting starts from
                             0 so if max_retries = 2 for example, then the task will execute
-                            1 + 2 times (1 time for first execution, 2 times for re-try).
+                            1 + 2 times (1 time for first execution, 2 times for re-try) in the
+                            worst case scenario.
     on tuple[type[Exception], ...]: The tuple of exception classes to retry on. The task will
                                     will only be retried if that exception that is raised
                                     during task execution is an instance of one of the listed
                                     exception classes.
 
                                     Examples:
 
-                                        If on=(Exception,) then any kind of exception will trigger
+                                        If `on=(Exception,)` then any kind of exception will trigger
                                         a retry.
 
-                                        If on=(ExceptionA, ExceptionB,) and during task
+                                        If `on=(ExceptionA, ExceptionB,)` and during task
                                         execution ExceptionC was raised, then retry is not triggered.
+
+                                        If `on=tuple()` then during task definition aiotaskq will raise
+                                        `InvalidRetryOptions`
     """
 
     max_retries: int | None

src/aiotaskq/task.py

@@ -11,7 +11,7 @@
 
 from .config import Config
 from .constants import Constants
-from .exceptions import InvalidArgument, ModuleInvalidForTask
+from .exceptions import InvalidArgument, InvalidRetryOptions, ModuleInvalidForTask
 from .interfaces import PollResponse, TaskOptions
 from .pubsub import PubSub
 
@@ -113,7 +113,11 @@ def __init__(
         Store the underlying function and an automatically generated task_id in the Task instance.
         """
         self.func = func
+
+        if retry and len(retry.get("on", [])) == 0:
+            raise InvalidRetryOptions('retry.on should not be empty')
         self.retry = retry
+
         self.args = args
         self.kwargs = kwargs
         self.id = task_id
@@ -134,8 +138,10 @@ def with_retry(self, max_retries: int, on: tuple[type[Exception], ...]) -> "Task
 
         We return a copy so that we don't overwrite the original task definition.
         """
-        task_ = copy.deepcopy(self)
-        retry = {"max_retries": max_retries, "on": on}
+        task_: Task = copy.deepcopy(self)
+        if len(on) == 0:
+            raise InvalidRetryOptions
+        retry: RetryOptions = {"max_retries": max_retries, "on": on}
         task_.retry = retry
         return task_
 

src/tests/test_task.py

@@ -3,7 +3,8 @@
 
 import pytest
 
-from aiotaskq.exceptions import InvalidArgument
+from aiotaskq.task import task as task_decorator
+from aiotaskq.exceptions import InvalidArgument, InvalidRetryOptions
 from tests.apps import simple_app
 
 if TYPE_CHECKING:
@@ -224,3 +225,42 @@ async def test_retry_until_successful(worker: "WorkerFixture", some_file: str):
     with open(some_file, mode="r", encoding="utf-8") as fi:
         num_lines = len(fi.read().rstrip("\n").split("\n"))
     assert num_lines == 3  # (first call + 2 retries)
+
+
+def test_empty_retry_on_during_task_definition__invalid():
+    # When a task is defined with options.retry.on = tuple()
+    exception = None
+    try:
+        @task_decorator(
+            options={
+                "retry": {
+                    "on": tuple(),
+                    "max_retries": 1,
+                },
+            },
+        )
+        def _():
+            return "Hello world"
+    except Exception as e:  # pylint: disable=broad-except
+        exception = e
+    finally:
+        # Then InvalidRetryOptions should be raised during task definition
+        assert isinstance(exception, InvalidRetryOptions), (
+            "Task definition should fail with InvalidRetryOptions"
+        )
+
+
+@pytest.mark.asyncio
+async def test_empty_retry_on_during_task_call__invalid(some_file: str):
+    # Give a task that is defined without error
+    some_task = simple_app.append_to_file
+
+    exception = None
+    try:
+        # When the task is called with options.retry.on = empty tuple
+        await some_task.with_retry(max_retries=1, on=tuple()).apply_async(some_file)
+    except Exception as e:  # pylint: disable=broad-except
+        exception = e
+    finally:
+        # Then InvalidRetryOptions should be raised during task call
+        assert isinstance(exception, InvalidRetryOptions), "Task call should fail with InvalidRetryOptions"