send_task_mock

Scope: function | Returns: MagicMock

The send_task_mock fixture replaces lime_task.send_task with a mock and returns it. By default each call returns a Status in the STARTED state. The returned status can be controlled via indirect parametrization.

Basic usage

While the fixture is active, any call to lime_task.send_task returns a Status in the STARTED state instead of enqueuing a task:

import lime_task

def test_task_is_started(send_task_mock, lime_app):
    status = lime_task.send_task("my_task", lime_app)

    assert status.status == lime_task.STARTED

Parametrizing the result

Pass a FakeTaskResult to control what status is returned:

import lime_task
from lime_test.common.helpers import FakeTaskResult

@pytest.mark.parametrize(
    "send_task_mock",
    [FakeTaskResult(lime_task.SUCCESS, 42)],
    indirect=True,
)
def test_task_succeeds(send_task_mock, lime_app):
    status = lime_task.send_task("my_task", lime_app)

    assert status.status == lime_task.SUCCESS
    assert status.result == 42

Parametrizing a sequence of results

Each successive call to lime_task.send_task returns the next result in the sequence:

@pytest.mark.parametrize(
    "send_task_mock",
    [(FakeTaskResult(lime_task.SUCCESS, 42), FakeTaskResult(lime_task.FAILURE))],
    indirect=True,
)
def test_second_send_fails(send_task_mock, lime_app):
    first = lime_task.send_task("my_task", lime_app)
    second = lime_task.send_task("my_task", lime_app)

    assert first.status == lime_task.SUCCESS
    assert second.status == lime_task.FAILURE

Parametrizing results per task

Use a dict keyed by task name to control results for individual tasks independently:

@pytest.mark.parametrize(
    "send_task_mock",
    [
        {
            "task_1": FakeTaskResult(lime_task.SUCCESS, 42),
            "task_2": FakeTaskResult(lime_task.FAILURE),
        }
    ],
    indirect=True,
)
def test_each_task_has_its_own_result(send_task_mock, lime_app):
    status_1 = lime_task.send_task("task_1", lime_app)
    status_2 = lime_task.send_task("task_2", lime_app)

    assert status_1.status == lime_task.SUCCESS
    assert status_2.status == lime_task.FAILURE

Per-task sequences are also supported — pass a list as the value for a task name to return results in order for that task.