task_status_mock

Scope: function | Returns: None

The task_status_mock fixture replaces the internal task status retrieval with a mock. By default each status lookup returns a Status in the STARTED state. The returned status can be controlled via indirect parametrization.

Basic usage

While the fixture is active, any status lookup — for example through lime_task.TaskStatusService — returns a Status in the STARTED state:

import lime_task

def test_task_status_is_started(task_status_mock, lime_app):
    status = lime_task.TaskStatusService(lime_app.id).get("some-task-id")

    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(
    "task_status_mock",
    [FakeTaskResult(lime_task.SUCCESS, 42)],
    indirect=True,
)
def test_task_status_succeeds(task_status_mock, lime_app):
    status = lime_task.TaskStatusService(lime_app.id).get("some-task-id")

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

Parametrizing a sequence of results

Each successive status lookup returns the next result in the sequence:

@pytest.mark.parametrize(
    "task_status_mock",
    [(FakeTaskResult(lime_task.SUCCESS, 42), FakeTaskResult(lime_task.FAILURE))],
    indirect=True,
)
def test_second_lookup_fails(task_status_mock, lime_app):
    service = lime_task.TaskStatusService(lime_app.id)

    first = service.get("some-task-id")
    second = service.get("some-task-id")

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