Improve test coverage

Author: RealOrangeOne

django_tasks/backends/base.py

@@ -41,17 +41,14 @@ def __init__(self, alias: str, params: dict) -> None:
     def _get_enqueue_on_commit_for_task(self, task: Task) -> bool:
         """
         Determine the correct `enqueue_on_commit` setting to use for a given task.
-
-        If the task defines it, use that, otherwise, fall back to the backend.
         """
-        # If this project doesn't use a database, there's nothing to commit to
-        if not connections.settings:
-            return False
-
-        if task.enqueue_on_commit is not None:
-            return task.enqueue_on_commit
 
-        return self.enqueue_on_commit
+        # If the task defines it, use that, otherwise, fall back to the backend.
+        return (
+            task.enqueue_on_commit
+            if task.enqueue_on_commit is not None
+            else self.enqueue_on_commit
+        )
 
     def validate_task(self, task: Task) -> None:
         """
@@ -121,8 +118,7 @@ async def aget_result(self, result_id: str) -> TaskResult:
 
     def check(self, **kwargs: Any) -> Iterable[messages.CheckMessage]:
         if self.enqueue_on_commit and not connections.settings:
-            yield messages.CheckMessage(
-                messages.ERROR,
+            yield messages.Error(
                 "`ENQUEUE_ON_COMMIT` cannot be used when no databases are configured",
                 hint="Set `ENQUEUE_ON_COMMIT` to False",
             )

django_tasks/backends/database/backend.py

@@ -88,8 +88,7 @@ def check(self, **kwargs: Any) -> Iterable[messages.CheckMessage]:
         backend_name = self.__class__.__name__
 
         if not apps.is_installed("django_tasks.backends.database"):
-            yield messages.CheckMessage(
-                messages.ERROR,
+            yield messages.Error(
                 f"{backend_name} configured as django_tasks backend, but database app not installed",
                 "Insert 'django_tasks.backends.database' in INSTALLED_APPS",
             )
@@ -102,8 +101,7 @@ def check(self, **kwargs: Any) -> Iterable[messages.CheckMessage]:
             django.VERSION >= (5, 1)
             and connection_requires_manual_exclusive_transaction(db_connection)
         ):
-            yield messages.CheckMessage(
-                messages.ERROR,
+            yield messages.Error(
                 f"{backend_name} is using SQLite non-exclusive transactions",
                 f"Set settings.DATABASES[{db_connection.alias!r}]['OPTIONS']['transaction_mode'] to 'EXCLUSIVE'",
             )

tests/tests/test_database_backend.py

@@ -264,6 +264,22 @@ def test_database_backend_app_missing(self) -> None:
         self.assertEqual(len(errors), 1)
         self.assertIn("django_tasks.backends.database", errors[0].hint)  # type:ignore[arg-type]
 
+    @skipIf(
+        connection.vendor != "sqlite", "Transaction mode is only applicable on SQLite"
+    )
+    def test_check_non_exclusive_transaction(self) -> None:
+        try:
+            with mock.patch.dict(
+                connection.settings_dict["OPTIONS"], {"transaction_mode": None}
+            ):
+                errors = list(default_task_backend.check())
+
+            self.assertEqual(len(errors), 1)
+            self.assertIn("['transaction_mode'] to 'EXCLUSIVE'", errors[0].hint)  # type:ignore[arg-type]
+        finally:
+            connection.close()
+            connection.get_connection_params()
+
     def test_priority_range_check(self) -> None:
         with self.assertRaises(IntegrityError):
             DBTaskResult.objects.create(

tests/tests/test_dummy_backend.py

@@ -1,13 +1,18 @@
 import json
 from typing import cast
+from unittest import mock
 
 from django.db import transaction
-from django.test import SimpleTestCase, TransactionTestCase, override_settings
+from django.test import (
+    SimpleTestCase,
+    TransactionTestCase,
+    override_settings,
+)
 from django.urls import reverse
 
 from django_tasks import ResultStatus, Task, default_task_backend, tasks
 from django_tasks.backends.dummy import DummyBackend
-from django_tasks.exceptions import ResultDoesNotExist
+from django_tasks.exceptions import InvalidTaskError, ResultDoesNotExist
 from tests import tasks as test_tasks
 
 
@@ -162,6 +167,13 @@ def test_exceptions(self) -> None:
         with self.assertRaisesMessage(ValueError, "Task has not finished yet"):
             result.traceback  # noqa: B018
 
+    def test_validate_disallowed_async_task(self) -> None:
+        with mock.patch.multiple(default_task_backend, supports_async_task=False):
+            with self.assertRaisesMessage(
+                InvalidTaskError, "Backend does not support async tasks"
+            ):
+                default_task_backend.validate_task(test_tasks.noop_task_async)
+
 
 class DummyBackendTransactionTestCase(TransactionTestCase):
     @override_settings(

tests/tests/test_tasks.py

@@ -176,6 +176,12 @@ def test_invalid_priority(self) -> None:
         test_tasks.noop_task.using(priority=-100)
         test_tasks.noop_task.using(priority=0)
 
+    def test_unknown_queue_name(self) -> None:
+        with self.assertRaisesMessage(
+            InvalidTaskError, "Queue 'queue-2' is not valid for backend"
+        ):
+            test_tasks.noop_task.using(queue_name="queue-2")
+
     def test_call_task(self) -> None:
         self.assertEqual(test_tasks.calculate_meaning_of_life.call(), 42)
 

tests/tests/test_utils.py

@@ -73,6 +73,14 @@ def test_keeps_return_value(self) -> None:
         self.assertTrue(utils.retry()(lambda: True)())
         self.assertFalse(utils.retry()(lambda: False)())
 
+    def test_skip_retry_on_keyboard_interrupt(self) -> None:
+        sentinel = Mock(side_effect=KeyboardInterrupt(""))
+
+        with self.assertRaises(KeyboardInterrupt):
+            utils.retry()(sentinel)()
+
+        self.assertEqual(sentinel.call_count, 1)
+
 
 class RandomIdTestCase(SimpleTestCase):
     def test_correct_length(self) -> None: