Testing¶

TaskQ ships a dedicated taskq.testing package with deterministic fakes, pytest fixtures, OTel helpers, chaos wrappers, and assertion utilities. Unit tests run against InMemoryBackend with a FakeClock: no Postgres, no Redis, no sleeping. Integration tests use testcontainers to spin up real Postgres and Dragonfly (Redis-compatible) containers. The Postgres image defaults to postgres:18-alpine and is overridable per run via TASKQ_TEST_PG_IMAGE — the seam CI's supported-version matrix (PG 15–18, see the README's support statement) uses to point the whole suite at another major.

Every symbol in taskq.testing lives outside the production import path so application code never pulls in test-only helpers.

InMemoryBackend¶

InMemoryBackend is a single-threaded, in-memory implementation of the Backend protocol. It simulates the full enqueue → dispatch → execute → terminal-write cycle including unique_for dedup, singleton enforcement, max_pending backpressure, retry/snooze transitions, cancellation, sweeps, schedule CRUD, and archive/expiry simulation.

from datetime import datetime, timezone
from taskq.testing.in_memory import InMemoryBackend
from taskq.testing.clock import FakeClock

clock = FakeClock(start=datetime.now(timezone.utc))
backend = InMemoryBackend(clock=clock)

Two InMemoryBackend instances in the same process are fully isolated; state is per-instance, never module-level. Single-threaded by contract: do not share across threads or event loops.

Registering actor stubs¶

run_until_drained executes jobs by calling registered stubs, not the real actor handlers. A stub is a callable that receives (payload: dict, ctx) and returns a result dict (or raises a control-flow exception):

import pytest
from datetime import datetime, timezone
from pydantic import BaseModel
from taskq import actor
from taskq.client import JobsClient
from taskq.testing.in_memory import InMemoryBackend
from taskq.testing.clock import FakeClock


class MyPayload(BaseModel):
    value: int


class MyResult(BaseModel):
    doubled: int


@actor
async def double_value(payload: MyPayload) -> MyResult:
    return MyResult(doubled=payload.value * 2)


async def test_double_value() -> None:
    clock = FakeClock(start=datetime.now(timezone.utc))
    backend = InMemoryBackend(clock=clock)
    client = JobsClient(backend)

    # The stub receives payload as a dict and ctx as a minimal duck-typed
    # object with job_id, attempt, payload, and cancel_event.
    backend.register_stub(
        double_value,
        lambda payload, ctx: {"doubled": payload["value"] * 2},
    )

    handle = await client.enqueue(double_value, MyPayload(value=21))
    await backend.run_until_drained()
    result = await handle.wait()
    assert result.doubled == 42

Pass the ActorRef itself (not double_value.name) so the runner can see the actor's declared contract: when payload_type is omitted it is auto-resolved from the ref, and run_until_drained validates every payload against the real model before invoking the stub, the same validation a production worker runs on every dispatch. A payload the actor's model would reject fails the job in-memory exactly as it would in production, so a payload-schema change cannot pass the in-memory suite by accident.

A bare name (register_stub("double_value", ...)) carries no declared model; the runner falls back to the permissive PassthroughPayload (extra="allow") and emits a StubPayloadTypeWarning, because that default accepts payload shapes the actor's real model would reject. Pass payload_type=MyPayload explicitly, or payload_type=PassthroughPayload to opt into the permissive behaviour deliberately and silence the warning.

register_stub also accepts retry, non_retryable_exceptions, retry_classifier, on_retry_exhausted, on_retry_exhausted_timeout, on_success (+ on_success_timeout), on_cancel (+ on_cancel_timeout), result_ttl, and payload_type to match the actor's configured behaviour.

Direct invocation (no queue)¶

For actors with no DI dependencies, call the ActorRef directly to bypass the queue entirely:

async def test_actor_direct() -> None:
    result = await double_value(MyPayload(value=21))
    assert result.doubled == 42

This runs the real handler in-process and is the simplest option when you do not need to test enqueue/dispatch behaviour. See Actors: direct invocation.

Actor-config registration¶

When using JobsClient with InMemoryBackend, register actor configs so the backend knows each actor's max_concurrent, queue, and metadata:

from taskq.testing.pg import DEFAULT_ACTORS

backend = InMemoryBackend(clock=clock)
for cfg in DEFAULT_ACTORS:
    backend.register_actor_config(actor=cfg.actor)

The memory_jobs fixture (below) does this automatically.

Archive and expiry simulation¶

InMemoryBackend exposes synchronous archive/expiry methods that mirror the maintenance leader's prune (Sweep 5) and archive-expiry (Sweep 6) sweeps:

from datetime import timedelta

result = backend.archive_terminal_jobs(
    retention=timedelta(days=30),
    archive_retention=timedelta(days=365),
)
expired = backend.expire_archived_jobs()
archived_row = await backend.get_archived(job_id)

FakeClock¶

FakeClock is a deterministic clock for tests. Its canonical home is taskq.clock, a production module that imports nothing from taskq.testing, so import taskq never pulls in the test-double backend. taskq.testing.clock re-exports it for backcompat. All four import paths (taskq, taskq.clock, taskq.testing, taskq.testing.clock) resolve to the same class object — see FakeClock: one class, four import paths.

from datetime import UTC, datetime, timedelta
from taskq.testing.clock import FakeClock

clock = FakeClock(datetime(2025, 1, 1, tzinfo=UTC))

clock.advance(timedelta(seconds=1))  # move forward 1s
clock.move_to(datetime(2025, 1, 2, tzinfo=UTC))  # jump to an instant
clock.now()  # → datetime(2025, 1, 2, tzinfo=UTC)
clock.monotonic()  # → elapsed seconds since an internal epoch (always non-zero)

advance(delta) adds delta to the clock; move_to(when) sets it to an absolute instant. Backward steps are safe; elapsed-time guards clamp to zero. monotonic() returns elapsed seconds from a fixed epoch so duration guards see a plausible non-zero starting value.

Inject FakeClock anywhere a Clock is accepted: InMemoryBackend, JobsClient, rate-limit primitives with backend="memory", and ConcurrencyReservation(clock=clock).

run_until_drained¶

backend.run_until_drained() is the test-only entry point that drives the dispatch-then-execute loop to completion. The loop:

Promotes scheduled → pending.
Dispatches the next highest-priority job (dispatch_batch with limit=1).
If nothing is dispatchable but future-scheduled jobs exist, advances the FakeClock to the earliest scheduled_at and continues.
Delegates per-job execution to consume_one_job, which handles Snooze, RetryAfter, ReservationUnavailable, generic exceptions, cancellation, and success.
Terminates when no jobs are pending, running, or scheduled.

await backend.run_until_drained()

Stubs are required

run_until_drained raises RuntimeError if it dispatches a job whose actor has no registered stub. Register stubs with backend.register_stub() before calling it. It also raises RuntimeError when the drain would otherwise end beside work that can never be dispatched, scheduled job whose actor has no registered stub or actor config at all gets zero dispatch capacity, so without the raise the job would sit pending forever while the drain reported success (and a caller in handle.wait() would hang with no diagnostic). The raise detects only; it never rewrites the stranded row.

When the clock is a FakeClock, the loop auto-advances through snoozes and scheduled jobs so a single call drains the entire queue. With a real clock (non-test) the loop returns instead of advancing: run_until_drained is intended for tests only.

The drain accepts one opt-in keyword: cancel_polling. When set, the drain drives the cancel poller itself (see Testing cancellation); the default keeps the historical behaviour, the drain never ticks cancel polling and tests drive cancellation by hand.

await backend.run_until_drained(cancel_polling=True)

Batch-policy simulation

run_until_drained invokes apply_batch_terminal_outcome after each job reaches a terminal write, so abort/completion semantics are tested identically to production. A batch enqueued with failure_policy=AbortBatchAfter(...) will have its consecutive-failure counter incremented, threshold checked, and remaining jobs cancelled on abort, all within the in-memory runner, no Postgres required.

Testing cancellation¶

Cooperative cancellation is a one-liner under the in-memory runner: register the job's cancel event, opt the drain into cancel polling, and cancel mid-drain. With run_until_drained(cancel_polling=True) the drain drives backend.tick_cancel_polling() itself, both between dispatch iterations and, while an attempt is in flight, from a concurrent ticker task, so the poller observes the request while the actor is running: it fires the registered event, the actor exits through its cancellation check, and the job ends terminal cancelled in-memory.

import asyncio
from datetime import UTC, datetime

from taskq._ids import new_job_id
from taskq.backend import EnqueueArgs
from taskq.testing.clock import FakeClock
from taskq.testing.in_memory import InMemoryBackend


async def test_cancel_mid_drain() -> None:
    clock = FakeClock(start=datetime(2025, 1, 1, tzinfo=UTC))
    backend = InMemoryBackend(clock=clock)

    async def victim(payload, ctx):
        await ctx.cancel_event.wait()  # park until the poller fires it
        ctx.check_cancelled()  # documented cooperative exit

    backend.register_stub("victim", victim)

    job_id = new_job_id()
    backend.register_cancel_event(job_id, asyncio.Event())
    await backend.enqueue(
        EnqueueArgs(
            id=job_id,
            actor="victim",
            queue="default",
            payload={},
            max_attempts=3,
            retry_kind="transient",
            scheduled_at=clock.now(),
        )
    )

    drain = asyncio.create_task(backend.run_until_drained(cancel_polling=True))

    # Cancel mid-drain: wait until the job is running, then request the
    # cancel the same way JobsClient.cancel() would.
    while (await backend.get(job_id)).status != "running":
        await asyncio.sleep(0.01)
    await backend.write_cancel_request(job_id, None)

    await drain
    row = await backend.get(job_id)
    assert row.status == "cancelled"

Without the flag the drain never ticks cancel polling; tests that want manual control can still call backend.write_cancel_request() and backend.tick_cancel_polling() by hand (advancing the FakeClock between ticks drives the phase escalation and the force-cancel arms), the same moves the runner performs automatically under the flag.

Asserting on progress¶

The stub context records every await ctx.progress(...) call on ctx.progress_reports, the in-memory progress assertion surface: one dict per call, {"seq": 1, "step": ..., "percent": ..., "detail": ..., "data": ...} with a strictly monotone seq, the harness half of the production progress contract (production publishes to Redis/Postgres; the runner records, so a test asserts on the list).

def progress_actor(payload, ctx):
    ...
    reports = ctx.progress_reports
    assert reports[0]["seq"] == 1

The stub context also wires the production sub-job and logging surfaces: await ctx.jobs.enqueue(actor_ref, payload) enqueues a real row the same drain dispatches, and ctx.log is a structlog logger bound with the job scope (job_id, actor, queue, attempt).

Pytest fixtures¶

Async tests and fixtures need pytest-asyncio configured; the loop scopes must match what the fixtures below assume:

# pyproject.toml
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "module"
asyncio_default_test_loop_scope = "module"

Pytest fixtures live in taskq.testing.fixtures and are imported into tests/conftest.py so they are available to all test modules. They are not re-exported from taskq.testing.__init__ to avoid importing pytest and asyncpg at the top level.

Unit-test fixtures (no containers)¶

Fixture	Scope	Yields	Notes
`memory_jobs`	function	`InMemoryBackend`	Fresh backend with a `FakeClock` at `2025-01-01 UTC`. Default actors pre-registered.
`actor_runner`	function	`ActorRunnerCallable`	Callable that builds a synthetic `JobContext` and invokes the handler, omitting `ctx` when the handler (or `ActorRef`) declares none. Forwards declared `**deps` as DI kwargs.

async def test_with_memory_jobs(memory_jobs: InMemoryBackend) -> None:
    # the ActorRef, not a bare name
    memory_jobs.register_stub(my_actor, lambda p, ctx: {"ok": True})
    ...

(A bare-name stub like this still runs, but emits StubPayloadTypeWarning because the runner cannot see a declared payload model for it: pass the ActorRef or payload_type= as described in "Registering actor stubs" to validate payloads for real.)

Integration fixtures (Postgres + Redis containers)¶

Published by taskq.testing.fixtures (import them into your own conftest to use them):

Fixture	Scope	Yields	Notes
`jobs_app`	function	`JobsApp(deps, backend)`	Drops/migrates/seeds a per-test schema.
`module_pg_schema`	module	`ModulePgSchema`	Per-file schema (worker-qualified hashed name). Migrates + seeds once.
`module_pg_pool`	module	`asyncpg.Pool`	Shared pool on the module schema.
`module_jobs_app`	module	`JobsApp`	Shared `WorkerDeps` + `PostgresBackend`.
`clean_pg_conn`	function	`asyncpg.Connection`	Truncates + re-seeds within the module schema.
`clean_jobs_app`	function	`JobsApp`	Truncate + re-seed, then open `WorkerDeps` + backend.
`worker_with_running_job`	function	`(worker_id, job_id, conn)`	Pre-created worker + running job on `clean_pg_conn`.
`redis_container`	session	`RedisContainer`	Dragonfly v2.0.0 (Redis-compatible wire protocol), started with `--dbnum 1024 --proactor_threads 2 --maxmemory 512mb`.
`killable_redis_container`	function	`RedisContainer`	Own Dragonfly container per test, for chaos tests that stop/restart Redis. Never stop the session container.
`redis_url`	function	`str`	Per-test URL with a UNIQUE, never-reused logical DB (1-1023) allocated from the invocation's file-backed monotonic counter (unique across all xdist workers). DB 0 is reserved.
`module_redis_url`	module	`str`	Unique Redis DB (1-1023) per module. `FLUSHDB` at setup AND on teardown.
`clean_redis_url`	function	`str`	`FLUSHDB` before each test.
`clean_redis_client`	function	`redis.asyncio.Redis`	Fresh async client on the module DB.
`backend_pair`	function	`Backend`	Parametrised `["memory", "pg"]`. The `pg` branch skips unless `@pytest.mark.integration` is set.

redis_url_for(container, db) is a published helper (not a fixture) that builds a redis://host:port/{db} URL for any container.

The PG fixtures above request a pg_dsn fixture by name; it is part of the contract but is deliberately NOT published (how you provision Postgres is your call). Consumer contract for defining your own:

Name: must be pg_dsn, yielding an asyncpg-friendly DSN string.
Scope: module at minimum (per-module databases or schemas; a session-scoped DSN to a per-run cluster also works).
Exclusivity: the PG cluster must belong to this test run. Schema names are stable across runs (worker-qualified hashes of module path / node id) and fixture setup is DROP-first, so two concurrent runs sharing one cluster silently drop each other's schemas.

The async fixtures use pytest_asyncio.fixture explicitly, so they work under both asyncio_mode = "auto" and "strict". Keep pytest-asyncio's default loop scopes at module (asyncio_default_fixture_loop_scope / asyncio_default_test_loop_scope): asyncpg pools are bound to the loop that created them, so module-scoped fixtures and the tests consuming them must share one loop per module.

For reference, TaskQ's OWN suite (tests/conftest.py) defines these additional fixtures: they are repo-private, not published:

Fixture	Scope	Yields	Notes
`pg_container`	session	`PostgresContainer`	Postgres 18 Alpine by default (`TASKQ_TEST_PG_IMAGE` overrides, e.g. the CI matrix's 15/16/17 legs), `max_connections=1000`.
`pg_dsn`	module	`str`	Per-module database on the shared container (worker-qualified hashed name), dropped on module teardown.
`settings`	function	`TaskQSettings`	Per-test env via `monkeypatch`, then `TaskQSettings.load()`.
`pg_conn`	function	`asyncpg.Connection`	Drops schema before each test. Prefer `clean_pg_conn`.

import pytest


@pytest.mark.integration
async def test_pg_backend(clean_jobs_app: JobsApp) -> None:
    backend = clean_jobs_app.backend
    ...

JobsApp is a named tuple: access fields as jobs_app.deps and jobs_app.backend rather than unpacking.

Health-socket isolation under xdist¶

WorkerSettings.health_socket_path defaults to the shared path /tmp/taskq_health.sock. Two in-process workers racing to bind it (e.g. under pytest-xdist) lose with EADDRINUSE or silently steal each other's socket. Mint a unique path per test with taskq.testing.health.unique_health_sock_path and pass it at settings construction time:

from taskq.testing.health import unique_health_sock_path
from taskq.testing.settings import make_integration_settings

settings = make_integration_settings(
    pg_dsn, HEALTH_SOCKET_PATH=unique_health_sock_path("my_test_module")
)

OTel test utilities¶

taskq.testing.otel provides a self-contained OTel stack so individual tests do not need to set up providers, exporters, or patching themselves. These helpers require the [otel] extra (opentelemetry-sdk).

Test tracer¶

setup_tracer creates an in-process TracerProvider backed by ListSpanExporter and patches obs.get_tracer for the duration of the test:

import pytest
from opentelemetry import trace
from taskq.testing.otel import setup_tracer, ListSpanExporter


async def test_span_emitted(monkeypatch: pytest.MonkeyPatch) -> None:
    provider, exporter = setup_tracer(monkeypatch)

    # ... run enqueue or dispatch ...

    consumer = exporter.span_named("process my_actor")
    assert consumer is not None
    assert consumer.kind == trace.SpanKind.CONSUMER

ListSpanExporter query helpers:

Method	Returns
`span_named(name)`	First `ReadableSpan` with that name, or `None`
`spans_named(name)`	All `ReadableSpan` objects with that name
`events_on(span_name, event_name)`	All events named `event_name` on the first span named `span_name`
`spans_with_kind(kind)`	All spans with the given `SpanKind`

Test meter¶

setup_meter creates a per-test MeterProvider backed by InMemoryMetricReader and patches the four core dispatch-path instruments:

from taskq.testing.otel import setup_meter, counter_value, histogram_points


async def test_metrics_recorded(monkeypatch: pytest.MonkeyPatch) -> None:
    reader = setup_meter(monkeypatch)

    # ... run enqueue and dispatch ...

    assert counter_value(reader, "messaging.client.published.messages") >= 1
    assert counter_value(reader, "messaging.client.consumed.messages") >= 1
    assert len(histogram_points(reader, "messaging.process.duration")) >= 1

Metric query helpers:

Helper	Returns
`collect_metrics(reader)`	All `Metric` objects from the reader
`counter_value(reader, name)`	Summed integer value for a counter
`counter_data_points(reader, name)`	`list[NumberDataPoint]` for a counter
`histogram_points(reader, name)`	`list[HistogramDataPoint]` for a histogram

Autouse fixtures¶

taskq.testing.otel exports two autouse pytest fixtures imported into conftest.py:

_otel_enabled_guard: snapshots and restores _otel_enabled around each test.
_logging_configured_guard: resets structlog configuration and removes ProcessorFormatter handlers around each test.

These run automatically for any test that imports from taskq.testing.otel. See Observability: testing observability for the full trace-context-propagation pattern.

Assertions¶

taskq.testing.assertions provides behavioral assertions that query observable state (job rows, events, spans) rather than implementation details.

Job-status assertions¶

from taskq.testing.assertions import (
    assert_job_status,
    assert_job_terminal,
    assert_attempt,
    wait_for_job_status,
)

# Assert a row has the expected status and optional fields.
# Returns the row (non-None) for chained access.
row = assert_job_status(row, "failed", error_class="ValueError", attempt=2, finished=True)

# Assert a terminal status with finished_at set.
assert_job_terminal(row, "succeeded")

# Assert on the attempt row at a given index.
assert_attempt(attempts, 0, outcome="succeeded", attempt_num=1)

# Poll backend.get until the job reaches a status (with timeout).
row = await wait_for_job_status(backend, job_id, "succeeded", timeout=2.0)

Event assertions¶

from taskq.testing.assertions import (
    assert_has_event,
    assert_transition_sequence,
    parse_detail,
)

# Find at least one event matching kind and optional state filters.
assert_has_event(events, "state_change", from_state="running", to_state="succeeded")

# Assert the (from_state, to_state) sequence from state_change events matches.
# The sequence holds the transitions of record only: claims and promotion are
# scheduler bookkeeping and write no event rows, so a plain run to success is
# exactly its terminal transition.
assert_transition_sequence(
    events,
    expected=[("running", "succeeded")],
)

OTel span assertions¶

from taskq.testing.assertions import assert_has_span, assert_has_otel_event

# Find a span by name; assert kind/status if provided.
span = assert_has_span(exporter, "process my_actor", kind=trace.SpanKind.CONSUMER)

# Find an OTel event by span and event name; assert state attributes.
assert_has_otel_event(
    exporter,
    "process my_actor",
    "lifecycle.running",
    from_state="pending",
    to_state="running",
)

Async and PG helpers¶

Helper	Description
`wait_for(event, timeout=2.0)`	Wait for an `asyncio.Event` with test-failure semantics on timeout.
`wait_for_leader(deps, timeout=5.0)`	Wait for the leader event on `WorkerDeps`.
`pg_now(conn)`	Return PG's `clock_timestamp()`: use instead of `datetime.now(UTC)` for cutoffs compared against SQL-written rows.
`plain_cli_output(output)`	Strip ANSI escapes and collapse whitespace for stable CLI-output assertions.

Chaos testing¶

taskq.testing.asyncpg_chaos provides ChaosConnection and ChaosPool for simulating mid-transaction failures in integration tests. The wrapper raises ChaosException on the configured call number, allowing tests to verify that transaction rollback works correctly when a failure occurs between SQL statements inside a transaction.

ChaosConnection¶

from taskq.testing.asyncpg_chaos import ChaosConnection, ChaosException

# Wrap a real connection; raise on the 3rd query call.
chaos_conn = ChaosConnection(real_conn, fail_on_call=3)

async with chaos_conn.transaction():
    await chaos_conn.execute("INSERT ...")  # call 1
    await chaos_conn.execute("UPDATE ...")  # call 2
    await chaos_conn.execute("DELETE ...")  # call 3 → ChaosException
# transaction rolls back; calls 1 and 2 are undone

fail_on_call counts query methods (execute, fetchrow, fetch, fetchval) in execution order. The exception is raised before the query is sent to PG. Pass fail_with= to raise a different exception type (e.g. asyncpg.PostgresConnectionError).

transaction() delegates to the real connection so asyncpg's commit/rollback works correctly when ChaosException propagates through async with conn.transaction():.

ChaosPool¶

ChaosPool is a pool-like object that yields a ChaosConnection from acquire(). Temporarily replace backend._worker_pool to test mid-transaction failures in backend methods:

from taskq.testing.asyncpg_chaos import ChaosPool

chaos_conn = ChaosConnection(real_conn, fail_on_call=5)
saved_pool = backend._worker_pool
backend._worker_pool = ChaosPool(chaos_conn)
try:
    # ... exercise backend method that acquires from the pool ...
    ...
except ChaosException:
    pass  # expected
finally:
    backend._worker_pool = saved_pool

ChaosException carries the call_number for debugging. It does not swallow CancelledError, which propagates naturally from the wrapped connection.

Shortened timing for chaos tests¶

shorten_chaos_settings temporarily reduces heartbeat/lock-lease/grace timing on WorkerDeps so chaos scenarios trigger sweeps quickly:

from taskq.testing.settings import shorten_chaos_settings

with shorten_chaos_settings(deps_a, deps_b):
    # heartbeat→1s, lock_lease→4s, grace periods→0
    ...

Property-based testing¶

TaskQ uses Hypothesis extensively for invariant testing. Property tests run against InMemoryBackend with a FakeClock: Hypothesis controls backend lifecycle, and PG requires testcontainers so it cannot be reset between examples.

The pattern: build a strategy of operations, drive the backend, and assert a deterministic invariant holds for every generated input.

from datetime import UTC, datetime, timedelta
from hypothesis import assume, given, settings
from hypothesis import strategies as st

from taskq._ids import new_job_id
from taskq.backend import EnqueueArgs
from taskq.testing.clock import FakeClock
from taskq.testing.in_memory import InMemoryBackend

_START = datetime(2025, 1, 1, tzinfo=UTC)
_LOCK_LEASE = timedelta(seconds=60)


@settings(max_examples=200, deadline=None)
@given(
    delay_seconds=st.floats(min_value=1, max_value=3600, allow_nan=False),
    deadline_offset=st.one_of(
        st.none(),
        st.floats(min_value=0, max_value=7200, allow_nan=False),
    ),
)
async def test_snooze_deterministic_outcome(
    delay_seconds: float,
    deadline_offset: float | None,
) -> None:
    backend = InMemoryBackend(clock=FakeClock(_START))
    delay = timedelta(seconds=delay_seconds)
    schedule_to_close = (
        None if deadline_offset is None else _START + timedelta(seconds=deadline_offset)
    )

    # Use assume() to exclude the dead zone where the invariant is ambiguous.
    if schedule_to_close is not None:
        assume(schedule_to_close > _START)

    job_id = new_job_id()
    await backend.enqueue(
        EnqueueArgs(
            id=job_id,
            actor="test_actor",
            queue="default",
            payload={"key": "value"},
            max_attempts=5,
            retry_kind="transient",
            scheduled_at=_START,
            schedule_to_close=schedule_to_close,
        )
    )

    wid = backend._worker_id
    dispatched = await backend.dispatch_batch(wid, ["default"], limit=1, lock_lease=_LOCK_LEASE)
    assert len(dispatched) == 1

    result = await backend.mark_snoozed(job_id, wid, delay)
    deadline_exceeded = schedule_to_close is not None and _START + delay > schedule_to_close

    if deadline_exceeded:
        assert result == "failed"
    else:
        assert result == "scheduled"

Tips for TaskQ property tests:

Use @settings(max_examples=200, deadline=None): async tests and PG latency make Hypothesis's default deadline flaky.
Use allow_nan=False, allow_infinity=False on st.floats to avoid timedelta edge cases.
Use assume() to exclude ranges where the invariant is ambiguous (e.g. the "dead zone" between a snooze and re-dispatch).
Drive the backend directly (enqueue, dispatch_batch, mark_*) rather than through JobsClient so strategies map to primitive values.
FakeClock + backend.advance_clock_to() makes time-dependent invariants fully deterministic.

Integration tests¶

Integration tests use testcontainers to spin up real Postgres and Redis containers. They are marked @pytest.mark.integration so unit-only runs skip them:

# Run unit tests only (no containers)
uv run pytest -m "not integration"

# Run integration tests (boots containers)
uv run pytest -m integration

Testcontainers setup¶

The session-scoped pg_container and redis_container fixtures boot Postgres (18 by default; TASKQ_TEST_PG_IMAGE overrides the major — the CI version matrix rides this) and Dragonfly (Redis-compatible) once per pytest invocation, one pair shared by every xdist worker of that invocation, and visible to no other invocation (of this repo or any other):

import pytest


@pytest.mark.integration
async def test_real_pg(clean_jobs_app: JobsApp) -> None:
    backend = clean_jobs_app.backend
    # backend is a real PostgresBackend against a migrated, seeded schema
    ...

The clean_jobs_app fixture truncates and re-seeds the module's PG schema before each test, then opens WorkerDeps + PostgresBackend. For Redis:

@pytest.mark.integration
async def test_real_redis(clean_redis_client) -> None:
    # clean_redis_client is a fresh redis.asyncio.Redis on a clean DB
    ...

Fast integration-test settings¶

make_integration_settings constructs WorkerSettings with short intervals for bounded test timeouts (heartbeat 0.5s, lock-lease 2s, grace 0.5s). The schema name defaults to a per-call unique tq_<token> so concurrent callers on a shared database can never clobber each other; pass an explicit schema_name= when you need a deterministic name:

from taskq.testing.settings import make_integration_settings

settings = make_integration_settings(pg_dsn, schema_name="tq_test")

PG row helpers¶

taskq.testing.pg provides helpers for creating test fixtures directly in PG:

Helper	Description
`create_pending_job(conn, schema, ...)`	Insert a pending job row.
`create_running_job(conn, schema, ...)`	Insert a running job + worker.
`create_workered_running_job(conn, schema)`	Insert worker + running job, returns `(worker_id, job_id)`.
`create_worker(conn, schema, worker_id)`	Insert a worker row.
`seed_actors(conn, schema)`	Seed default `actor_config` rows.
`reset_schema(conn, schema)`	Truncate all tables (FK-safe CASCADE) + re-seed.
`truncate_schema(conn, schema)`	Truncate all tables.
`get_job_triple(conn, schema, job_id)`	Fetch job + actor config + worker.
`DEFAULT_ACTORS`	Pre-built actor configs for tests.

Two-pod / chaos scenarios¶

_open_two_pg_workers (in taskq.testing.fixtures) opens two independent WorkerDeps + PostgresBackend instances against the same schema for two-pod leader-election and chaos-kill tests. It does not start the leader loops; callers construct MaintenanceLeader themselves with different initial states.

xdist isolation¶

Integration tests are grouped by module via the pytest_collection_modifyitems hook in conftest.py, which assigns xdist_group(name=<module basename>) to every integration test without an explicit group (see pyproject.toml for the --dist=loadgroup setting). Grouping is defense-in-depth and efficiency, not a correctness requirement: every module-scoped name is already worker-qualified (schema names hash in the xdist worker id, per-module database names likewise, and Redis DBs come from a per-process counter), so a module accidentally split across workers would get distinct resources rather than clobbering. What grouping prevents is the duplicated create/migrate/drop work and doubled connection pressure of such a split.