Getting Started
Testing
Fake clients, inline execution, and bounded drain helpers.
Consumer tests can replace the producer facade without starting a worker or database.
from taskq.testing import FakeTaskQClient
with tq.replace_client(FakeTaskQClient()) as fake:
await application_call()
fake.assert_enqueued(
"mail.send",
where={"payload.recipient": "me@example.test"},
)
Inline mode
Execute registered handlers immediately in-process:
from taskq.testing import inline_mode
async with inline_mode(tq) as recorder:
await application_call()
assert recorder.settled("mail.send")[0].is_complete
Drain
Run a bounded claim/settle loop against a real (or fake) transport without sleeps:
from taskq.testing import drain
report = await drain(tq, queue="mail", max_jobs=100)
assert report.completed == 1
Also available: work(task=..., payload=...) for direct handler invocation and require_enqueued(...) for assertion helpers.
These are consumer-test conveniences, not production modes. The fake does not model PostgreSQL fencing, privileges, budgets, or transaction isolation. Use a scratch PostgreSQL transaction with
work, require_enqueued, or drain(..., connection=connection) when those contracts matter.