test for TASK_PROCESSOR_MODE

- consistent usage of `settings.TASK_PROCESSOR_MODE`
- test for failure running tasks when `TASK_PROCESSOR_MODE` is `False`
- add `task_processor_mode` pytest marker
- consistent usage of `django_db` marker

Author: khvn26

settings/dev.py

@@ -46,7 +46,7 @@
 TASK_DELETE_RETENTION_DAYS = 15
 TASK_DELETE_RUN_EVERY = timedelta(days=1)
 TASK_DELETE_RUN_TIME = time(5, 0, 0)
-TASK_PROCESSOR_MODE = True
+TASK_PROCESSOR_MODE = False
 TASK_RUN_METHOD = TaskRunMethod.TASK_PROCESSOR
 
 # Avoid models.W042 warnings

src/common/prometheus/utils.py

@@ -1,3 +1,4 @@
+import importlib
 import typing
 
 import prometheus_client
@@ -16,3 +17,26 @@ def get_registry() -> prometheus_client.CollectorRegistry:
     registry = prometheus_client.CollectorRegistry()
     MultiProcessCollector(registry)  # type: ignore[no-untyped-call]
     return registry
+
+
+def reload_metrics(*metric_module_names: str) -> None:
+    """
+    Clear the registry of all collectors from the given modules
+    and reload the modules to register the collectors again.
+
+    Used in tests to reset the state of the metrics module
+    when needed.
+    """
+
+    for module_name in metric_module_names:
+        metrics_module = importlib.import_module(module_name)
+
+        _collectors = metrics_module.__dict__.values()
+
+        for collector in [
+            *(registry := prometheus_client.REGISTRY)._collector_to_names
+        ]:
+            if collector in _collectors:
+                registry.unregister(collector)
+
+        importlib.reload(metrics_module)

src/task_processor/decorators.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import typing
 from datetime import datetime, time, timedelta
 from threading import Thread
@@ -178,7 +177,7 @@ def register_recurring_task(
     first_run_time: time | None = None,
     timeout: timedelta | None = timedelta(minutes=30),
 ) -> typing.Callable[[TaskCallable[TaskParameters]], TaskCallable[TaskParameters]]:
-    if not os.environ.get("RUN_BY_PROCESSOR"):
+    if not settings.TASK_PROCESSOR_MODE:
         # Do not register recurring tasks if not invoked by task processor
         return lambda f: f
 

tests/unit/task_processor/conftest.py

@@ -2,13 +2,25 @@
 import typing
 
 import pytest
+from pytest_django.fixtures import SettingsWrapper
 
+from common.prometheus.utils import reload_metrics
 from task_processor.task_registry import RegisteredTask
 
 
-@pytest.fixture
-def run_by_processor(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setenv("RUN_BY_PROCESSOR", "True")
+@pytest.fixture()
+def task_processor_mode(settings: SettingsWrapper) -> None:
+    settings.TASK_PROCESSOR_MODE = True
+    # The setting is supposed to be set before the metrics module is imported,
+    # so reload it
+    reload_metrics("task_processor.metrics")
+
+
+@pytest.fixture(autouse=True)
+def task_processor_mode_marked(request: pytest.FixtureRequest) -> None:
+    for marker in request.node.iter_markers():
+        if marker.name == "task_processor_mode":
+            request.getfixturevalue("task_processor_mode")
 
 
 class GetTaskProcessorCaplog(typing.Protocol):

tests/unit/task_processor/test_unit_task_processor_decorators.py

@@ -99,10 +99,10 @@ def my_function(*args: typing.Any, **kwargs: typing.Any) -> None:
     )
 
 
+@pytest.mark.django_db
+@pytest.mark.task_processor_mode
 def test_register_recurring_task(
     mocker: MockerFixture,
-    db: None,
-    run_by_processor: None,
 ) -> None:
     # Given
     mock = mocker.Mock()
@@ -128,10 +128,8 @@ def test_register_recurring_task(
     assert task.callable is mock
 
 
-def test_register_recurring_task_does_nothing_if_not_run_by_processor(
-    mocker: MockerFixture,
-    db: None,
-) -> None:
+@pytest.mark.django_db
+def test_register_recurring_task_does_nothing_if_not_run_by_processor() -> None:
     # Given
 
     task_kwargs = {"first_arg": "foo", "second_arg": "bar"}
@@ -187,9 +185,8 @@ class NonSerializableObj:
         my_function.delay(args=(NonSerializableObj(),))
 
 
-def test_delay_returns_none_if_task_queue_is_full(
-    settings: SettingsWrapper, db: None
-) -> None:
+@pytest.mark.django_db
+def test_delay_returns_none_if_task_queue_is_full(settings: SettingsWrapper) -> None:
     # Given
     settings.TASK_RUN_METHOD = TaskRunMethod.TASK_PROCESSOR
 
@@ -209,9 +206,9 @@ def my_function(*args: typing.Any, **kwargs: typing.Any) -> None:
     assert task is None
 
 
+@pytest.mark.django_db
 def test_delay__expected_metrics(
     settings: SettingsWrapper,
-    db: None,
     assert_metric: AssertMetricFixture,
 ) -> None:
     # Given
@@ -232,7 +229,8 @@ def my_function(*args: typing.Any, **kwargs: typing.Any) -> None:
     )
 
 
-def test_can_create_task_with_priority(settings: SettingsWrapper, db: None) -> None:
+@pytest.mark.django_db
+def test_can_create_task_with_priority(settings: SettingsWrapper) -> None:
     # Given
     settings.TASK_RUN_METHOD = TaskRunMethod.TASK_PROCESSOR
 

tests/unit/task_processor/test_unit_task_processor_health.py

@@ -1,3 +1,4 @@
+import pytest
 from pytest_django.fixtures import SettingsWrapper
 from pytest_mock import MockerFixture
 
@@ -25,8 +26,8 @@ def test_is_processor_healthy_returns_false_if_task_not_processed(
     assert result is False
 
 
+@pytest.mark.django_db
 def test_is_processor_healthy_returns_true_if_task_processed(
-    db: None,
     settings: SettingsWrapper,
 ) -> None:
     # Given

tests/unit/task_processor/test_unit_task_processor_monitoring.py

@@ -1,12 +1,14 @@
 from datetime import timedelta
 
+import pytest
 from django.utils import timezone
 
 from task_processor.models import Task
 from task_processor.monitoring import get_num_waiting_tasks
 
 
-def test_get_num_waiting_tasks(db: None) -> None:
+@pytest.mark.django_db
+def test_get_num_waiting_tasks() -> None:
     # Given
     now = timezone.now()