Testing resources

test_mode(resources=...), MockResource, pytest fixtures, propagating errors.

The test_mode() context manager runs tasks synchronously in the calling thread without starting a worker. Pass mock resources to override real factories during tests.

Coming from Celery, test_mode() replaces task_always_eager — but as a context manager it's scoped, so it's safe to use in parallel test runs.

Injecting mock resources

python

from unittest.mock import MagicMock
from myapp.tasks import queue, process_order

def test_process_order():
    mock_db = MagicMock()
    mock_db.return_value.get.return_value = Order(id=42, total=99.0)

    with queue.test_mode(resources={"db": mock_db}) as results:
        process_order.delay(42)

    assert results[0].succeeded
    mock_db.return_value.get.assert_called_once_with(Order, 42)

The resources= dict maps resource names to mock values. Any plain Python object works — MagicMock, a real instance, a simple dict, whatever your test needs.

When test_mode(resources=...) is active:

Resources are taken directly from the dict — no factories are called.
Proxy reconstruction is bypassed. Proxy markers in arguments are passed through unchanged, so tests don't fail because of missing files or network connections.
The previous resource runtime is restored on context exit.

`MockResource`

MockResource adds call tracking on top of a plain mock value:

python

from flexiq import MockResource

real_session_factory = sessionmaker(create_engine("sqlite:///test.db"))

def test_with_spy():
    spy_db = MockResource("db", wraps=real_session_factory, track_calls=True)

    with queue.test_mode(resources={"db": spy_db}) as results:
        process_order.delay(42)

    assert spy_db.call_count == 1
    assert results[0].succeeded

Parameter	Description
`name`	Resource name (informational, used in repr).
`return_value`	Value returned when the resource is accessed.
`wraps`	Wrap a real object — it is returned as-is when the resource is accessed.
`track_calls`	If `True`, increment `call_count` each time the resource is accessed.

MockResource attributes:

Attribute	Description
`call_count`	Number of times the resource was accessed during the test.
`calls`	List of call tuples (currently `[]` — tracking is count-only).

MockResource wraps a value — it is not callable by default. If your task calls db() to obtain a session, your return_value or wraps must be callable (e.g., a MagicMock or a real session factory).

Explicit kwargs override injection

If a test calls .delay() with an explicit kwarg that matches an injected resource name, the explicit value wins:

python

@queue.task()
def my_task(db: Inject["db"]):
    db.do_something()

# Override injection for this one call:
mock_db = MagicMock()
my_task.delay(db=mock_db)

This also works inside test_mode() — the resource dict is the default, but explicit call-site kwargs always take precedence.

pytest fixture pattern

python

# conftest.py
import pytest
from unittest.mock import MagicMock
from myapp.tasks import queue


@pytest.fixture
def mock_db():
    session = MagicMock()
    session.return_value.get.return_value = None
    return session


@pytest.fixture
def task_results(mock_db):
    with queue.test_mode(resources={"db": mock_db}) as results:
        yield results, mock_db

python

# test_orders.py
def test_order_processed(task_results):
    results, mock_db = task_results
    mock_db.return_value.get.return_value = Order(id=1, status="pending")

    process_order.delay(1)

    assert results[0].succeeded
    order = mock_db.return_value.get.return_value
    assert order.status == "processed"

Propagating errors

By default, task exceptions are captured in TestResult.error. To re-raise them immediately:

python

with queue.test_mode(propagate_errors=True) as results:
    process_order.delay(999)  # raises if the task raises

Use propagate_errors=False (the default) when you want to test error handling — check results[0].failed and results[0].error.