Dramatiq actors run inline with StubBroker and results middleware: working versions (tested 2026-10-15)

dramatiq==2.2.1
pytest==9.1.1
pytest-asyncio==1.4.0

import asyncio

import dramatiq
from dramatiq.brokers.stub import StubBroker
from dramatiq.middleware.asyncio import AsyncIO
from dramatiq.results import Results
from dramatiq.results.backends.stub import StubBackend

broker = StubBroker()
broker.add_middleware(AsyncIO())
broker.add_middleware(Results(backend=StubBackend()))
dramatiq.set_broker(broker)


@dramatiq.actor(store_results=True)
def add(x, y):
    return x + y


@dramatiq.actor
def square(x):
    return x * x


@dramatiq.actor(store_results=True, max_retries=0)
def divide(x, y):
    return x / y


@dramatiq.actor(store_results=True)
async def upper(text):
    await asyncio.sleep(0)
    return text.upper()

import asyncio

import pytest
from dramatiq.results.errors import ResultFailure, ResultMissing
from dramatiq.worker import Worker

from tasks import add, broker, divide, square, upper


@pytest.fixture
def worker():
    w = Worker(broker, worker_timeout=100, worker_threads=1)
    w.start()
    yield w
    w.stop()
    broker.dead_letters_by_queue.clear()


def test_actor_runs_on_worker(worker):
    message = add.send(2, 3)
    broker.join(add.queue_name)
    assert message.get_result(backend=broker.get_results_backend()) == 5


def test_result_missing_without_store_results(worker):
    message = square.send(4)
    broker.join(square.queue_name)
    with pytest.raises(ResultMissing):
        message.get_result(backend=broker.get_results_backend())


def test_join_fail_fast_reraises_actor_error(worker):
    divide.send(1, 0)
    with pytest.raises(ZeroDivisionError):
        broker.join(divide.queue_name, fail_fast=True)


def test_result_failure_after_non_fail_fast_join(worker):
    message = divide.send(1, 0)
    broker.join(divide.queue_name, fail_fast=False)
    with pytest.raises(ResultFailure) as exc:
        message.get_result(backend=broker.get_results_backend(), block=True)
    assert "ZeroDivisionError" in str(exc.value)


async def test_async_actor_result_is_retrieved_from_a_thread(worker):
    message = upper.send("abc")
    await asyncio.to_thread(broker.join, upper.queue_name)
    result = await asyncio.to_thread(message.get_result, backend=broker.get_results_backend(), block=True)
    assert result == "ABC"

[pytest]
pythonpath = .
asyncio_mode = auto