feat(test-tools): Add run_tasks fixture (#85)

Author: khvn26

poetry.lock

@@ -19,7 +19,10 @@ dependencies = [
     "requests",
     "simplejson (>=3,<4)",
 ]
-optional-dependencies = { test-tools = ["pyfakefs (>=5,<6)"] }
+optional-dependencies = { test-tools = [
+    "pyfakefs (>=5,<6)",
+    "pytest-django (>=4,<5)",
+] }
 authors = [
     { name = "Matthew Elwell" },
     { name = "Gagan Trivedi" },

pyproject.toml

@@ -1,6 +1,11 @@
-from common.test_tools.types import AssertMetricFixture, SnapshotFixture
+from common.test_tools.types import (
+    AssertMetricFixture,
+    RunTasksFixture,
+    SnapshotFixture,
+)
 
 __all__ = (
     "AssertMetricFixture",
+    "RunTasksFixture",
     "SnapshotFixture",
 )

src/common/test_tools/__init__.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.test_tools.types import (
+    AssertMetricFixture,
+    RunTasksFixture,
+    Snapshot,
+    SnapshotFixture,
+)
+from task_processor.task_run_method import TaskRunMethod
 
 
 def pytest_addoption(parser: pytest.Parser) -> None:
@@ -68,6 +76,16 @@ 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
+    from common.prometheus.utils import reload_metrics
+
+    reload_metrics("task_processor.metrics")
+
+
 @pytest.fixture(autouse=True)
 def flagsmith_markers_marked(
     request: pytest.FixtureRequest,
@@ -77,6 +95,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,
+    transactional_db: None,
+    task_processor_mode: None,
+) -> RunTasksFixture:
+    settings.TASK_RUN_METHOD = TaskRunMethod.TASK_PROCESSOR
+
+    from task_processor.processor import run_tasks
+
+    return partial(run_tasks, database="default")
 
 
 @pytest.fixture

src/common/test_tools/plugin.py

@@ -1,8 +1,11 @@
 from pathlib import Path
-from typing import Protocol
+from typing import TYPE_CHECKING, Protocol
 
 import pytest
 
+if TYPE_CHECKING:
+    from task_processor.models import TaskRun
+
 
 class AssertMetricFixture(Protocol):
     def __call__(
@@ -14,6 +17,13 @@ def __call__(
     ) -> None: ...
 
 
+class RunTasksFixture(Protocol):
+    def __call__(
+        self,
+        num_tasks: int,
+    ) -> "list[TaskRun]": ...
+
+
 class SnapshotFixture(Protocol):
     def __call__(self, name: str = "") -> "Snapshot": ...
 

src/common/test_tools/types.py

@@ -1,9 +1,13 @@
+from datetime import datetime, timedelta
+
 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
+from task_processor.models import Task
 
 
 def test_assert_metrics__asserts_expected(
@@ -55,3 +59,44 @@ 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"
+
+
+def test_run_tasks__delay_inside_task__runs_expected(
+    run_tasks: RunTasksFixture,
+) -> None:
+    # Given
+    @register_task_handler()
+    def my_task() -> None: ...
+
+    @register_task_handler()
+    def calling_task() -> None:
+        my_task.delay(delay_until=datetime.now() + timedelta(hours=1))
+
+    calling_task.delay()
+
+    # When
+    task_runs = run_tasks(num_tasks=1)
+
+    # Then
+    assert len(task_runs) == 1
+    assert Task.objects.count() == 2
+    assert Task.objects.latest("scheduled_for").task_identifier == "test_plugin.my_task"

tests/unit/common/test_tools/test_plugin.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