Testing Tasks & Workers¶

litestar-queues is designed for fast, deterministic unit and integration testing without requiring external brokers or background daemon threads in test suites.

Runnable Example¶

docs/examples/learn/testing_tasks.py

"""Testing tasks with immediate execution and deterministic consume_one claims."""

from litestar_queues import QueueConfig, QueueService, TaskExitCode, WorkerConfig, consume_one, task

EXPECTED_DIRECT_TAX = 825
EXPECTED_IMMEDIATE_TAX = 1650
EXPECTED_QUEUED_TAX = 4125


@task("orders.calculate_tax", queue="orders")
async def calculate_tax(subtotal_cents: int, rate_bps: int) -> int:
    """Compute tax in cents from a subtotal and basis-point rate."""
    return (subtotal_cents * rate_bps) // 10_000


async def main() -> None:
    """Demonstrate direct callable invocation, immediate execution, and consume_one."""
    direct_value = await calculate_tax(10_000, 825)
    assert direct_value == EXPECTED_DIRECT_TAX

    immediate_config = QueueConfig(
        queue_backend="memory", execution_backend="immediate", worker=WorkerConfig(placement="external")
    )
    async with QueueService(config=immediate_config) as service:
        handle = await service.enqueue(calculate_tax, 20_000, 825)
        await handle.refresh()
        assert handle.status == "completed"
        assert handle.result == EXPECTED_IMMEDIATE_TAX

    local_config = QueueConfig(
        queue_backend="memory", execution_backend="local", worker=WorkerConfig(placement="external")
    )
    async with QueueService(config=local_config) as service:
        queued = await service.enqueue(calculate_tax, 50_000, 825)
        assert queued.status == "pending"
        exit_code = await consume_one(service, queued.id)
        assert exit_code == TaskExitCode.SUCCESS
        await queued.refresh()
        assert queued.result == EXPECTED_QUEUED_TAX

Testing Layers¶

Direct Callable Invocation¶

Because @task preserves the underlying function signature and await task_fn(*args, **kwargs) delegates directly to the wrapped callable, pure domain logic can be unit tested without initializing QueueService:

result = await calculate_tax(10_000, 825)
assert result == 825

Inline Execution with Immediate Mode¶

To test the full enqueue-to-completion pipeline (argument serialization, uniqueness checks, task_dependency_provider lifecycle, event publication, and TaskResult updates) synchronously inside an HTTP test request, combine queue_backend="memory", execution_backend="immediate", and WorkerConfig(placement="external"):

from litestar_queues import QueueConfig, WorkerConfig

test_config = QueueConfig(
    queue_backend="memory",
    execution_backend="immediate",
    worker=WorkerConfig(placement="external"),
)

With execution_backend="immediate", every await queue_service.enqueue(...) persists the record, claims it, runs execute_record() inline before enqueue() returns, and transitions the record to completed (or failed / cancelled). Setting WorkerConfig(placement="external") prevents AsyncTestClient from spawning a background polling worker during the test.

Deterministic Single-Claim Testing with consume_one¶

When testing queued state transitions, retries, expiration deadlines, or custom worker hooks where the record must first sit in pending or scheduled status, use execution_backend="local" with consume_one:

from litestar_queues import TaskExitCode, consume_one

handle = await service.enqueue(calculate_tax, 50_000, 825)
assert handle.status == "pending"

exit_code = await consume_one(service, handle.id)
assert exit_code == TaskExitCode.SUCCESS

await handle.refresh()
assert handle.status == "completed"

consume_one(service, task_id) claims the specific record by UUID, executes a single attempt with heartbeat management, and returns a TaskExitCode (SUCCESS, FAILURE, MISSING_RECORD, UNKNOWN_TASK, CLAIM_LOST, or CANCELLED).

Isolating the Task Registry in Pytest Suites¶

If tests dynamically register temporary @task definitions, snapshot and restore the task registry or call clear_task_registry() in a fixture so temporary task names do not leak across test modules. For event assertions with InMemoryQueueEventSink and EventCapture, see Testing Queue Events.