feat(test-tools): Add run_tasks fixture

Author: khvn26

src/common/test_tools/plugin.py

@@ -1,11 +1,19 @@
+from functools import partial
 from typing import Generator
 
 import prometheus_client
 import pytest
 from prometheus_client.metrics import MetricWrapperBase
 from pyfakefs.fake_filesystem import FakeFilesystem
+from pytest_django.fixtures import SettingsWrapper
 
-from common.test_tools.types import AssertMetricFixture, Snapshot, SnapshotFixture
+from common.prometheus.utils import reload_metrics
+from common.test_tools.types import (
+    AssertMetricFixture,
+    RunTasksFixture,
+    Snapshot,
+    SnapshotFixture,
+)
 
 
 def pytest_addoption(parser: pytest.Parser) -> None:
@@ -68,6 +76,14 @@ def enterprise_mode(fs: FakeFilesystem) -> Generator[None, None, None]:
     is_enterprise.cache_clear()
 
 
+@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 flagsmith_markers_marked(
     request: pytest.FixtureRequest,
@@ -77,6 +93,21 @@ def flagsmith_markers_marked(
             request.getfixturevalue("saas_mode")
         if marker.name == "enterprise_mode":
             request.getfixturevalue("enterprise_mode")
+        if marker.name == "task_processor_mode":
+            request.getfixturevalue("task_processor_mode")
+
+
+@pytest.fixture(name="run_tasks")
+def run_tasks_impl(
+    settings: SettingsWrapper,
+    db: None,
+    task_processor_mode: None,
+) -> RunTasksFixture:
+    settings.TASK_RUN_METHOD = "TASK_PROCESSOR"
+
+    from task_processor.processor import run_tasks
+
+    return partial(run_tasks, database="default")
 
 
 @pytest.fixture

tests/unit/common/test_tools/test_plugin.py

@@ -1,9 +1,10 @@
 import prometheus_client
 import pytest
 
-from common.test_tools import AssertMetricFixture
+from common.test_tools import AssertMetricFixture, RunTasksFixture
 from common.test_tools.plugin import assert_metric_impl
 from common.test_tools.utils import edition_printer
+from task_processor.decorators import register_task_handler
 
 
 def test_assert_metrics__asserts_expected(
@@ -55,3 +56,22 @@ def test_saas_mode_marker__is_saas_returns_expected() -> None:
 def test_enterprise_mode_marker__is_enterprise_returns_expected() -> None:
     # When & Then
     assert edition_printer() == "enterprise!"
+
+
+def test_run_tasks__runs_expected_tasks(
+    run_tasks: RunTasksFixture,
+) -> None:
+    # Given
+    @register_task_handler()
+    def my_task() -> None:
+        return None
+
+    my_task.delay()
+
+    # When
+    task_runs = run_tasks(num_tasks=1)
+
+    # Then
+    assert len(task_runs) == 1
+    assert task_runs[0].task.task_identifier == "test_plugin.my_task"
+    assert task_runs[0].result == "SUCCESS"

tests/unit/task_processor/conftest.py

@@ -3,7 +3,6 @@
 import pytest
 from pytest_django.fixtures import SettingsWrapper
 
-from common.prometheus.utils import reload_metrics
 from task_processor.task_registry import RegisteredTask
 
 
@@ -35,22 +34,6 @@ def current_database(
     return database
 
 
-@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")
-            return
-
-
 @pytest.fixture(autouse=True)
 def task_registry() -> typing.Generator[dict[str, RegisteredTask], None, None]:
     from task_processor.task_registry import registered_tasks