Testing

InMemoryBackend, FakeClock, pytest fixtures, and test assertions.

Package surface (fakes, assertions, settings factories)

testing

Test-only helpers for TaskQ: FakeClock, InMemoryBackend, and stub utilities.

Every symbol here lives in taskq.testing — never in taskq.backend — so production code does not pull in test-only helpers.

run_until_drained, tick_cancel_polling, and register_cancel_event are methods on the re-exported :class:InMemoryBackend class. The runner logic lives in :mod:taskq.testing._runner.

Pytest fixtures are NOT re-exported here — they live in :mod:taskq.testing.fixtures and are imported by :mod:tests.conftest directly. This avoids importing pytest / asyncpg at the taskq.testing top level.

JobContext is re-exported for convenience; the eventual public JobContext will replace this test-scoped version.

OTel test utilities (ListSpanExporter, setup_tracer, setup_meter) are NOT re-exported here — import them from taskq.testing.otel directly. They require the [otel] extra (opentelemetry-sdk).

DEFAULT_ACTORS module-attribute

DEFAULT_ACTORS: tuple[str, ...] = (
    "actor_a",
    "actor_b",
    "actor_c",
    "A",
    "C",
    "X",
    "test_actor",
    "_progress_redis_hundred",
    "_progress_redis_single",
    "_progress_redis_three",
)

create_worker module-attribute

create_worker = _create_worker

__all__ module-attribute

__all__ = [
    "DEFAULT_ACTORS",
    "EmptyPayload",
    "FakeBackend",
    "FakeClock",
    "InMemoryBackend",
    "JobContext",
    "JobTriple",
    "StubActorConfig",
    "WarningSpy",
    "as_backend",
    "assert_attempt",
    "assert_has_event",
    "assert_has_otel_event",
    "assert_has_span",
    "assert_job_status",
    "assert_job_terminal",
    "assert_transition_sequence",
    "create_pending_job",
    "create_running_job",
    "create_worker",
    "create_workered_running_job",
    "default_actor_config",
    "error_info",
    "get_job_triple",
    "make_enqueue_args",
    "make_integration_settings",
    "make_integration_settings_dict",
    "make_job_row",
    "parse_detail",
    "pg_now",
    "reset_schema",
    "seed_actors",
    "setup_running_job",
    "truncate_schema",
    "unique_health_sock_path",
    "wait_for",
    "wait_for_job_status",
    "wait_for_leader",
]

EmptyPayload

Bases: BaseModel

FakeBackend

FakeBackend(
    *,
    mark_snoozed_return: Literal[
        "scheduled", "failed", "noop"
    ] = "scheduled",
    mark_retry_after_return: Literal[
        "scheduled",
        "failed:DeadlineExceeded",
        "failed:MaxAttemptsExceeded",
        "noop",
    ] = "scheduled",
)

Minimal backend recording method calls for assertions.

Source code in src/taskq/testing/actor.py

def __init__(
    self,
    *,
    mark_snoozed_return: Literal["scheduled", "failed", "noop"] = "scheduled",
    mark_retry_after_return: Literal[
        "scheduled", "failed:DeadlineExceeded", "failed:MaxAttemptsExceeded", "noop"
    ] = "scheduled",
) -> None:
    self.mark_succeeded_calls: list[tuple[UUID, UUID, dict[str, object] | None]] = []
    self.mark_cancelled_calls: list[dict[str, object]] = []
    self.mark_snoozed_calls: list[dict[str, object]] = []
    self.mark_retry_after_calls: list[dict[str, object]] = []
    self.mark_failed_or_retry_calls: list[dict[str, object]] = []
    self._mark_snoozed_return: Literal["scheduled", "failed", "noop"] = mark_snoozed_return
    self._mark_retry_after_return: Literal[
        "scheduled", "failed:DeadlineExceeded", "failed:MaxAttemptsExceeded", "noop"
    ] = mark_retry_after_return

BACKEND_PROTOCOL_VERSION class-attribute instance-attribute

BACKEND_PROTOCOL_VERSION: int = BACKEND_PROTOCOL_VERSION

supports_transactional_simulation class-attribute instance-attribute

supports_transactional_simulation: bool = False

mark_succeeded_calls instance-attribute

mark_succeeded_calls: list[
    tuple[UUID, UUID, dict[str, object] | None]
] = []

mark_cancelled_calls instance-attribute

mark_cancelled_calls: list[dict[str, object]] = []

mark_snoozed_calls instance-attribute

mark_snoozed_calls: list[dict[str, object]] = []

mark_retry_after_calls instance-attribute

mark_retry_after_calls: list[dict[str, object]] = []

mark_failed_or_retry_calls instance-attribute

mark_failed_or_retry_calls: list[dict[str, object]] = []

enqueue async

enqueue(args: EnqueueArgs) -> JobRow

Source code in src/taskq/testing/actor.py

async def enqueue(self, args: EnqueueArgs) -> JobRow:
    raise NotImplementedError

enqueue_with_conn async

enqueue_with_conn(
    conn: object, args: EnqueueArgs
) -> JobRow

Source code in src/taskq/testing/actor.py

async def enqueue_with_conn(self, conn: object, args: EnqueueArgs) -> JobRow:
    raise NotImplementedError

dispatch_batch async

dispatch_batch(
    worker_id: UUID,
    queues: list[str],
    limit: int,
    lock_lease: timedelta,
) -> list[JobRow]

Source code in src/taskq/testing/actor.py

async def dispatch_batch(
    self, worker_id: UUID, queues: list[str], limit: int, lock_lease: timedelta
) -> list[JobRow]:
    raise NotImplementedError

heartbeat_jobs async

heartbeat_jobs(
    worker_id: UUID, lock_lease: timedelta
) -> int

Source code in src/taskq/testing/actor.py

async def heartbeat_jobs(self, worker_id: UUID, lock_lease: timedelta) -> int:
    return 0

extend_reservation_leases async

extend_reservation_leases(
    worker_id: UUID, lock_lease: timedelta
) -> int

Source code in src/taskq/testing/actor.py

async def extend_reservation_leases(self, worker_id: UUID, lock_lease: timedelta) -> int:
    return 0

mark_succeeded async

mark_succeeded(
    job_id: UUID,
    worker_id: UUID,
    result: dict[str, object] | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    fallback_result_ttl: timedelta | None = None,
) -> bool

Source code in src/taskq/testing/actor.py

async def mark_succeeded(
    self,
    job_id: UUID,
    worker_id: UUID,
    result: dict[str, object] | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    fallback_result_ttl: timedelta | None = None,
) -> bool:
    self.mark_succeeded_calls.append((job_id, worker_id, result))
    return True

mark_succeeded_with_conn async

mark_succeeded_with_conn(
    conn: object,
    job_id: UUID,
    worker_id: UUID,
    result: dict[str, object] | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    fallback_result_ttl: timedelta | None = None,
) -> bool

Source code in src/taskq/testing/actor.py

async def mark_succeeded_with_conn(
    self,
    conn: object,
    job_id: UUID,
    worker_id: UUID,
    result: dict[str, object] | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    fallback_result_ttl: timedelta | None = None,
) -> bool:
    return await self.mark_succeeded(
        job_id, worker_id, result, progress_seq, progress_state, fallback_result_ttl
    )

mark_failed_or_retry async

mark_failed_or_retry(
    job_id: UUID,
    worker_id: UUID,
    error_info: ErrorInfo,
    retry_delay: timedelta | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> JobRow

Source code in src/taskq/testing/actor.py

async def mark_failed_or_retry(
    self,
    job_id: UUID,
    worker_id: UUID,
    error_info: ErrorInfo,
    retry_delay: timedelta | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> JobRow:
    self.mark_failed_or_retry_calls.append(
        {
            "job_id": job_id,
            "worker_id": worker_id,
            "error_info": error_info,
            "retry_delay": retry_delay,
        }
    )
    return _make_job_row()

mark_cancelled async

mark_cancelled(
    job_id: UUID,
    worker_id: UUID,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> bool

Source code in src/taskq/testing/actor.py

async def mark_cancelled(
    self,
    job_id: UUID,
    worker_id: UUID,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> bool:
    self.mark_cancelled_calls.append(
        {
            "job_id": job_id,
            "worker_id": worker_id,
            "progress_seq": progress_seq,
            "progress_state": progress_state,
        }
    )
    return True

write_cancel_escalation async

write_cancel_escalation(
    job_id: UUID, worker_id: UUID, phase: Literal[2]
) -> bool

Source code in src/taskq/testing/actor.py

async def write_cancel_escalation(
    self, job_id: UUID, worker_id: UUID, phase: Literal[2]
) -> bool:
    return False

mark_abandoned async

mark_abandoned(
    job_id: UUID,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> bool

Source code in src/taskq/testing/actor.py

async def mark_abandoned(
    self,
    job_id: UUID,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> bool:
    return False

mark_snoozed async

mark_snoozed(
    job_id: UUID,
    worker_id: UUID,
    delay: timedelta,
    *,
    metadata_update: dict[str, object] | None = None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    outcome: AttemptOutcome = "snoozed",
) -> Literal["scheduled", "failed", "noop"]

Source code in src/taskq/testing/actor.py

async def mark_snoozed(
    self,
    job_id: UUID,
    worker_id: UUID,
    delay: timedelta,
    *,
    metadata_update: dict[str, object] | None = None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    outcome: AttemptOutcome = "snoozed",
) -> Literal["scheduled", "failed", "noop"]:
    self.mark_snoozed_calls.append(
        {
            "job_id": job_id,
            "worker_id": worker_id,
            "delay": delay,
            "metadata_update": metadata_update,
            "progress_seq": progress_seq,
            "progress_state": progress_state,
            "outcome": outcome,
        }
    )
    return self._mark_snoozed_return

mark_retry_after async

mark_retry_after(
    job_id: UUID,
    worker_id: UUID,
    delay: timedelta,
    *,
    consume_budget: bool = True,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> Literal[
    "scheduled",
    "failed:DeadlineExceeded",
    "failed:MaxAttemptsExceeded",
    "noop",
]

Source code in src/taskq/testing/actor.py

async def mark_retry_after(
    self,
    job_id: UUID,
    worker_id: UUID,
    delay: timedelta,
    *,
    consume_budget: bool = True,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> Literal["scheduled", "failed:DeadlineExceeded", "failed:MaxAttemptsExceeded", "noop"]:
    self.mark_retry_after_calls.append(
        {
            "job_id": job_id,
            "worker_id": worker_id,
            "delay": delay,
            "consume_budget": consume_budget,
            "progress_seq": progress_seq,
            "progress_state": progress_state,
        }
    )
    return self._mark_retry_after_return

write_attempt async

write_attempt(attempt: AttemptRow) -> None

Source code in src/taskq/testing/actor.py

async def write_attempt(self, attempt: AttemptRow) -> None:
    pass

get_attempts async

get_attempts(job_id: UUID) -> list[AttemptRow]

Source code in src/taskq/testing/actor.py

async def get_attempts(self, job_id: UUID) -> list[AttemptRow]:
    return []

get_events async

get_events(job_id: UUID) -> list[EventRow]

Source code in src/taskq/testing/actor.py

async def get_events(self, job_id: UUID) -> list[EventRow]:
    return []

write_cancel_request async

write_cancel_request(
    job_id: UUID, reason: str | None
) -> bool

Source code in src/taskq/testing/actor.py

async def write_cancel_request(self, job_id: UUID, reason: str | None) -> bool:
    return False

poll_cancel_flags async

poll_cancel_flags(worker_id: UUID) -> list[CancelFlag]

Source code in src/taskq/testing/actor.py

async def poll_cancel_flags(self, worker_id: UUID) -> list[CancelFlag]:
    return []

scheduled_to_pending async

scheduled_to_pending() -> int

Source code in src/taskq/testing/actor.py

async def scheduled_to_pending(self) -> int:
    return 0

deadline_sweep async

deadline_sweep() -> int

Source code in src/taskq/testing/actor.py

async def deadline_sweep(self) -> int:
    return 0

reclaim_expired_locks async

reclaim_expired_locks(
    cancel_grace: timedelta, cleanup_grace: timedelta
) -> int

Source code in src/taskq/testing/actor.py

async def reclaim_expired_locks(self, cancel_grace: timedelta, cleanup_grace: timedelta) -> int:
    return 0

get async

get(job_id: UUID) -> JobRow | None

Source code in src/taskq/testing/actor.py

async def get(self, job_id: UUID) -> JobRow | None:
    return None

list_jobs async

list_jobs(filters: JobFilter) -> list[JobRow]

Source code in src/taskq/testing/actor.py

async def list_jobs(self, filters: JobFilter) -> list[JobRow]:
    return []

count_pending_jobs async

count_pending_jobs(actors: list[str]) -> dict[str, int]

Source code in src/taskq/testing/actor.py

async def count_pending_jobs(self, actors: list[str]) -> dict[str, int]:
    return {}

count_active_jobs async

count_active_jobs(queues: list[str]) -> int

Source code in src/taskq/testing/actor.py

async def count_active_jobs(self, queues: list[str]) -> int:
    return 0

get_actor_max_pending async

get_actor_max_pending() -> dict[str, int | None]

Source code in src/taskq/testing/actor.py

async def get_actor_max_pending(self) -> dict[str, int | None]:
    return {}

enqueue_batch async

enqueue_batch(
    args_list: list[EnqueueArgs],
    *,
    connection: object = None,
) -> list[JobRow]

Source code in src/taskq/testing/actor.py

async def enqueue_batch(
    self,
    args_list: list[EnqueueArgs],
    *,
    connection: object = None,
) -> list[JobRow]:
    raise NotImplementedError

subscribe_wake

subscribe_wake() -> AbstractAsyncContextManager[
    asyncio.Event
]

Source code in src/taskq/testing/actor.py

def subscribe_wake(self) -> AbstractAsyncContextManager[asyncio.Event]:
    raise NotImplementedError

create_schedule async

create_schedule(args: ScheduleCreateArgs) -> object

Source code in src/taskq/testing/actor.py

async def create_schedule(self, args: ScheduleCreateArgs) -> object:
    raise NotImplementedError

list_schedules async

list_schedules(
    *, actor: str | None = None, enabled: bool | None = None
) -> list[object]

Source code in src/taskq/testing/actor.py

async def list_schedules(
    self,
    *,
    actor: str | None = None,
    enabled: bool | None = None,
) -> list[object]:
    raise NotImplementedError

update_schedule async

update_schedule(
    schedule_id: UUID, args: ScheduleUpdateArgs
) -> object

Source code in src/taskq/testing/actor.py

async def update_schedule(self, schedule_id: UUID, args: ScheduleUpdateArgs) -> object:
    raise NotImplementedError

delete_schedule async

delete_schedule(schedule_id: UUID) -> None

Source code in src/taskq/testing/actor.py

async def delete_schedule(self, schedule_id: UUID) -> None:
    raise NotImplementedError

StubActorConfig dataclass

StubActorConfig(
    retry: RetryPolicy,
    non_retryable_exceptions: tuple[
        type[Exception], ...
    ] = (),
    retry_classifier: RetryClassifierHook | None = None,
    on_retry_exhausted: OnRetryExhausted | None = None,
    on_retry_exhausted_timeout: float = 3.0,
    on_success: OnSuccess | None = None,
    on_success_timeout: float = 3.0,
)

retry instance-attribute

retry: RetryPolicy

non_retryable_exceptions class-attribute instance-attribute

non_retryable_exceptions: tuple[type[Exception], ...] = ()

retry_classifier class-attribute instance-attribute

retry_classifier: RetryClassifierHook | None = None

on_retry_exhausted class-attribute instance-attribute

on_retry_exhausted: OnRetryExhausted | None = None

on_retry_exhausted_timeout class-attribute instance-attribute

on_retry_exhausted_timeout: float = 3.0

on_success class-attribute instance-attribute

on_success: OnSuccess | None = None

on_success_timeout class-attribute instance-attribute

on_success_timeout: float = 3.0

FakeClock

FakeClock(start: datetime)

Deterministic clock for tests.

Accepts a start datetime (typically datetime(2025, 1, 1, tzinfo=UTC)). now() returns the current internal time; move_to and advance let tests control the clock explicitly. monotonic() returns elapsed seconds from _EPOCH so that elapsed-time guards see a non-zero starting value.

Source code in src/taskq/testing/clock.py

def __init__(self, start: datetime) -> None:
    self._now = start

now

now() -> datetime

Source code in src/taskq/testing/clock.py

def now(self) -> datetime:
    return self._now

move_to

move_to(when: datetime) -> None

Set the clock to when.

Source code in src/taskq/testing/clock.py

def move_to(self, when: datetime) -> None:
    """Set the clock to *when*."""
    self._now = when

advance

advance(delta: timedelta) -> None

Add delta to the clock.

Source code in src/taskq/testing/clock.py

def advance(self, delta: timedelta) -> None:
    """Add *delta* to the clock."""
    self._now = self._now + delta

monotonic

monotonic() -> float

Elapsed seconds since _EPOCH — consistent with now().

Same wall-clock position always returns the same float; never decreases within a test.

Source code in src/taskq/testing/clock.py

def monotonic(self) -> float:
    """Elapsed seconds since ``_EPOCH`` — consistent with ``now()``.

    Same wall-clock position always returns the same float; never
    decreases within a test.
    """
    return (self._now - _EPOCH).total_seconds()

InMemoryBackend

InMemoryBackend(
    clock: Clock,
    cancellation_grace_period: timedelta = timedelta(
        seconds=30
    ),
    cleanup_grace_period: timedelta = timedelta(seconds=30),
    rng: Random | None = None,
    *,
    actor_configs: Iterable[ActorConfig] | None = None,
    result_max_bytes: int = MAX_RESULT_BYTES,
)

Deterministic, in-memory backend for unit tests.

All state is held as per-instance attributes: no module-level mutable state, no class-level caches. Two InMemoryBackend instances created in the same test session are fully isolated.

Single-threaded by contract — do not share across threads or event loops. Intra-coroutine re-entry within a single event loop is acceptable; cross-thread use is a caller bug.

Source code in src/taskq/testing/in_memory.py

def __init__(
    self,
    clock: Clock,
    cancellation_grace_period: timedelta = timedelta(seconds=30),
    cleanup_grace_period: timedelta = timedelta(seconds=30),
    rng: random.Random | None = None,
    *,
    actor_configs: Iterable[ActorConfig] | None = None,
    result_max_bytes: int = MAX_RESULT_BYTES,
) -> None:
    self._clock = clock
    self._result_max_bytes = result_max_bytes
    self._cancellation_grace = cancellation_grace_period
    self._cleanup_grace = cleanup_grace_period
    self._worker_id: UUID = new_uuid()
    self._rng = rng

    self._jobs: dict[JobId, JobRow] = {}
    self._attempts: dict[JobId, list[AttemptRow]] = {}
    self._events: list[EventRow] = []
    self._idempotency_index: dict[tuple[str, str], JobId] = {}
    self._event_seq: int = 0
    self._cancel_observed_at: dict[JobId, datetime] = {}
    self._cancel_events: dict[JobId, asyncio.Event] = {}
    self._wake_subscribers: set[asyncio.Event] = set()
    self._cancel_wake_subscribers: set[asyncio.Event] = set()
    self._actor_stubs: dict[str, StubFn] = {}
    self._actor_configs: dict[str, _InMemoryActorConfig] = {}
    self._actor_configs_meta: dict[str, ActorConfig] = {}
    if actor_configs is not None:
        for cfg in actor_configs:
            self._actor_configs_meta[cfg.actor] = cfg
    self._slot_table: _SlotTable | None = None
    self._archive: dict[JobId, _ArchivedJobRow] = {}
    self._archive_attempts: dict[JobId, list[AttemptRow]] = {}
    self._schedules: dict[UUID, ScheduleRecord] = {}
    self._queues: dict[str, QueueMode] = {}
    self._batches: dict[UUID, BatchRow] = {}

BACKEND_PROTOCOL_VERSION class-attribute

BACKEND_PROTOCOL_VERSION: int = BACKEND_PROTOCOL_VERSION

supports_transactional_simulation class-attribute

supports_transactional_simulation: bool = True

slot_table property

slot_table: _SlotTable

advance_clock_to

advance_clock_to(when: datetime) -> None

Source code in src/taskq/testing/in_memory.py

def advance_clock_to(self, when: datetime) -> None:
    _advance_clock_to(self, when)

register_stub

register_stub(
    actor_name: str,
    fn: StubFn,
    *,
    retry: RetryPolicy | None = None,
    non_retryable_exceptions: tuple[
        type[BaseException], ...
    ] = (),
    retry_classifier: RetryClassifierHook | None = None,
    on_retry_exhausted: OnRetryExhausted | None = None,
    on_retry_exhausted_timeout: float = 3.0,
    on_success: OnSuccess | None = None,
    on_success_timeout: float = 3.0,
    result_ttl: timedelta | None = None,
    payload_type: type[BaseModel] | None = None,
) -> None

Source code in src/taskq/testing/in_memory.py

def register_stub(
    self,
    actor_name: str,
    fn: StubFn,
    *,
    retry: RetryPolicy | None = None,
    non_retryable_exceptions: tuple[type[BaseException], ...] = (),
    retry_classifier: RetryClassifierHook | None = None,
    on_retry_exhausted: OnRetryExhausted | None = None,
    on_retry_exhausted_timeout: float = 3.0,
    on_success: OnSuccess | None = None,
    on_success_timeout: float = 3.0,
    result_ttl: timedelta | None = None,
    payload_type: type[BaseModel] | None = None,
) -> None:
    _register_stub(
        self,
        actor_name,
        fn,
        retry=retry,
        non_retryable_exceptions=non_retryable_exceptions,
        retry_classifier=retry_classifier,
        on_retry_exhausted=on_retry_exhausted,
        on_retry_exhausted_timeout=on_retry_exhausted_timeout,
        on_success=on_success,
        on_success_timeout=on_success_timeout,
        result_ttl=result_ttl,
        payload_type=payload_type,
    )

register_cancel_event

register_cancel_event(job_id: JobId, event: Event) -> None

Source code in src/taskq/testing/in_memory.py

def register_cancel_event(self, job_id: JobId, event: asyncio.Event) -> None:
    _register_cancel_event(self, job_id, event)

get_events async

get_events(job_id: JobId) -> list[EventRow]

Source code in src/taskq/testing/in_memory.py

async def get_events(self, job_id: JobId) -> list[EventRow]:
    return await _get_events(self, job_id)

poll_reclaim_events async

poll_reclaim_events(
    after_id: int,
    limit: int = DEFAULT_RECLAIM_POLL_LIMIT,
    *,
    visibility_delay: timedelta | None = None,
) -> list[EventRow]

Source code in src/taskq/testing/in_memory.py

async def poll_reclaim_events(
    self,
    after_id: int,
    limit: int = DEFAULT_RECLAIM_POLL_LIMIT,
    *,
    visibility_delay: timedelta | None = None,
) -> list[EventRow]:
    return await _poll_reclaim_events(self, after_id, limit, visibility_delay=visibility_delay)

enqueue async

enqueue(args: EnqueueArgs) -> JobRow

Source code in src/taskq/testing/in_memory.py

async def enqueue(self, args: EnqueueArgs) -> JobRow:
    return await _enqueue(self, args)

enqueue_with_conn async

enqueue_with_conn(
    conn: object, args: EnqueueArgs
) -> JobRow

Source code in src/taskq/testing/in_memory.py

async def enqueue_with_conn(
    self,
    conn: object,
    args: EnqueueArgs,
) -> JobRow:
    return await _enqueue_with_conn(self, conn, args)

enqueue_batch async

enqueue_batch(
    args_list: list[EnqueueArgs],
    *,
    connection: object = None,
) -> list[JobRow]

Source code in src/taskq/testing/in_memory.py

async def enqueue_batch(
    self,
    args_list: list[EnqueueArgs],
    *,
    connection: object = None,
) -> list[JobRow]:
    return await _enqueue_batch(self, args_list, connection=connection)

enqueue_batch_fast async

enqueue_batch_fast(
    args_list: list[EnqueueArgs],
    *,
    connection: object = None,
) -> int

Source code in src/taskq/testing/in_memory.py

async def enqueue_batch_fast(
    self,
    args_list: list[EnqueueArgs],
    *,
    connection: object = None,
) -> int:
    return await _enqueue_batch_fast(self, args_list, connection=connection)

register_actor_config

register_actor_config(
    *,
    actor: str,
    max_concurrent: int | None = None,
    max_pending: int | None = None,
    queue: str = "default",
    metadata: dict[str, object] | None = None,
) -> None

Source code in src/taskq/testing/in_memory.py

def register_actor_config(
    self,
    *,
    actor: str,
    max_concurrent: int | None = None,
    max_pending: int | None = None,
    queue: str = "default",
    metadata: dict[str, object] | None = None,
) -> None:
    _register_actor_config(
        self,
        actor=actor,
        max_concurrent=max_concurrent,
        max_pending=max_pending,
        queue=queue,
        metadata=metadata,
    )

register_actor_configs

register_actor_configs(
    configs: Iterable[ActorConfig],
) -> None

Source code in src/taskq/testing/in_memory.py

def register_actor_configs(self, configs: Iterable[ActorConfig]) -> None:
    _register_actor_configs(self, configs)

set_queue_mode

set_queue_mode(queue_name: str, mode: QueueMode) -> None

Source code in src/taskq/testing/in_memory.py

def set_queue_mode(self, queue_name: str, mode: QueueMode) -> None:
    _set_queue_mode(self, queue_name, mode)

dispatch_batch async

dispatch_batch(
    worker_id: UUID,
    queues: list[str],
    limit: int,
    lock_lease: timedelta,
) -> list[JobRow]

Source code in src/taskq/testing/in_memory.py

async def dispatch_batch(
    self,
    worker_id: UUID,
    queues: list[str],
    limit: int,
    lock_lease: timedelta,
) -> list[JobRow]:
    return await _dispatch_batch(self, worker_id, queues, limit, lock_lease)

heartbeat_jobs async

heartbeat_jobs(
    worker_id: UUID, lock_lease: timedelta
) -> int

Source code in src/taskq/testing/in_memory.py

async def heartbeat_jobs(
    self,
    worker_id: UUID,
    lock_lease: timedelta,
) -> int:
    now = self._clock.now()
    count = 0
    for job_id, row in list(self._jobs.items()):
        if row.status == "running" and row.locked_by_worker == worker_id:
            self._jobs[job_id] = replace(
                row,
                lock_expires_at=now + lock_lease,
                last_heartbeat_at=now,
            )
            count += 1
    return count

extend_reservation_leases async

extend_reservation_leases(
    worker_id: UUID, lock_lease: timedelta
) -> int

Source code in src/taskq/testing/in_memory.py

async def extend_reservation_leases(
    self,
    worker_id: UUID,
    lock_lease: timedelta,
) -> int:
    now = self._clock.now()
    count = 0
    for job_id, row in list(self._jobs.items()):
        if row.status == "running" and row.locked_by_worker == worker_id:
            if self._slot_table is not None:
                count += self._slot_table.extend_leases_for_job(job_id, now, lock_lease)
            else:
                count += 1
    return count

mark_succeeded async

mark_succeeded(
    job_id: JobId,
    worker_id: UUID,
    result: dict[str, object] | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    fallback_result_ttl: timedelta | None = None,
) -> bool

Source code in src/taskq/testing/in_memory.py

async def mark_succeeded(
    self,
    job_id: JobId,
    worker_id: UUID,
    result: dict[str, object] | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    fallback_result_ttl: timedelta | None = None,
) -> bool:
    return await _mark_succeeded(
        self, job_id, worker_id, result, progress_seq, progress_state, fallback_result_ttl
    )

mark_succeeded_with_conn async

mark_succeeded_with_conn(
    conn: object,
    job_id: JobId,
    worker_id: UUID,
    result: dict[str, object] | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    fallback_result_ttl: timedelta | None = None,
) -> bool

Source code in src/taskq/testing/in_memory.py

async def mark_succeeded_with_conn(
    self,
    conn: object,
    job_id: JobId,
    worker_id: UUID,
    result: dict[str, object] | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    fallback_result_ttl: timedelta | None = None,
) -> bool:
    return await _mark_succeeded_with_conn(
        self, conn, job_id, worker_id, result, progress_seq, progress_state, fallback_result_ttl
    )

mark_failed_or_retry async

mark_failed_or_retry(
    job_id: JobId,
    worker_id: UUID,
    error_info: ErrorInfo,
    retry_delay: timedelta | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> JobRow

Source code in src/taskq/testing/in_memory.py

async def mark_failed_or_retry(
    self,
    job_id: JobId,
    worker_id: UUID,
    error_info: ErrorInfo,
    retry_delay: timedelta | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> JobRow:
    return await _mark_failed_or_retry(
        self, job_id, worker_id, error_info, retry_delay, progress_seq, progress_state
    )

mark_cancelled async

mark_cancelled(
    job_id: JobId,
    worker_id: UUID,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> bool

Source code in src/taskq/testing/in_memory.py

async def mark_cancelled(
    self,
    job_id: JobId,
    worker_id: UUID,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> bool:
    return await _mark_cancelled(self, job_id, worker_id, progress_seq, progress_state)

write_cancel_escalation async

write_cancel_escalation(
    job_id: JobId, worker_id: UUID, phase: Literal[2]
) -> bool

Source code in src/taskq/testing/in_memory.py

async def write_cancel_escalation(
    self,
    job_id: JobId,
    worker_id: UUID,
    phase: Literal[2],
) -> bool:
    return await _write_cancel_escalation(self, job_id, worker_id, phase)

mark_abandoned async

mark_abandoned(
    job_id: JobId,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> bool

Source code in src/taskq/testing/in_memory.py

async def mark_abandoned(
    self,
    job_id: JobId,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> bool:
    return await _mark_abandoned(self, job_id, progress_seq, progress_state)

mark_snoozed async

mark_snoozed(
    job_id: JobId,
    worker_id: UUID,
    delay: timedelta,
    *,
    metadata_update: dict[str, object] | None = None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    outcome: AttemptOutcome = "snoozed",
) -> Literal["scheduled", "failed", "noop"]

Source code in src/taskq/testing/in_memory.py

async def mark_snoozed(
    self,
    job_id: JobId,
    worker_id: UUID,
    delay: timedelta,
    *,
    metadata_update: dict[str, object] | None = None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
    outcome: AttemptOutcome = "snoozed",
) -> Literal["scheduled", "failed", "noop"]:
    return await _mark_snoozed(
        self,
        job_id,
        worker_id,
        delay,
        metadata_update=metadata_update,
        progress_seq=progress_seq,
        progress_state=progress_state,
        outcome=outcome,
    )

mark_retry_after async

mark_retry_after(
    job_id: JobId,
    worker_id: UUID,
    delay: timedelta,
    *,
    consume_budget: bool = True,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> Literal[
    "scheduled",
    "failed:DeadlineExceeded",
    "failed:MaxAttemptsExceeded",
    "noop",
]

Source code in src/taskq/testing/in_memory.py

async def mark_retry_after(
    self,
    job_id: JobId,
    worker_id: UUID,
    delay: timedelta,
    *,
    consume_budget: bool = True,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> Literal["scheduled", "failed:DeadlineExceeded", "failed:MaxAttemptsExceeded", "noop"]:
    return await _mark_retry_after(
        self,
        job_id,
        worker_id,
        delay,
        consume_budget=consume_budget,
        progress_seq=progress_seq,
        progress_state=progress_state,
    )

write_attempt async

write_attempt(attempt: AttemptRow) -> None

Source code in src/taskq/testing/in_memory.py

async def write_attempt(self, attempt: AttemptRow) -> None:
    await _write_attempt(self, attempt)

get_attempts async

get_attempts(job_id: JobId) -> list[AttemptRow]

Source code in src/taskq/testing/in_memory.py

async def get_attempts(self, job_id: JobId) -> list[AttemptRow]:
    return await _get_attempts(self, job_id)

write_cancel_request async

write_cancel_request(
    job_id: JobId, reason: str | None
) -> bool

Source code in src/taskq/testing/in_memory.py

async def write_cancel_request(
    self,
    job_id: JobId,
    reason: str | None,
) -> bool:
    row = self._jobs.get(job_id)
    if row is None:
        return False

    if row.status == "running" and row.cancel_phase == 0:
        now = self._clock.now()
        self._jobs[job_id] = replace(
            row,
            cancel_requested_at=now,
            cancel_phase=1,
        )
        self._append_cancel_request_event(job_id, now, reason)
        for event in self._cancel_wake_subscribers:
            event.set()
        logger.debug(
            "cancel_requested",
            kind="state_change",
            from_state="running",
            to_state="running",
            job_id=job_id,
            cancel_phase=1,
        )
        return True

    if row.status in ("pending", "scheduled"):
        now = self._clock.now()
        prev_status = row.status
        self._jobs[job_id] = replace(
            row,
            status="cancelled",
            finished_at=now,
        )
        self._append_state_change_event(
            job_id=job_id,
            from_state=prev_status,
            to_state="cancelled",
            now=now,
        )
        self._append_cancel_request_event(job_id, now, reason)
        logger.debug(
            "state-change",
            kind="state_change",
            from_state=prev_status,
            to_state="cancelled",
            job_id=job_id,
        )
        return True

    return False

poll_cancel_flags async

poll_cancel_flags(worker_id: UUID) -> list[CancelFlag]

Source code in src/taskq/testing/in_memory.py

async def poll_cancel_flags(
    self,
    worker_id: UUID,
) -> list[CancelFlag]:
    return [
        CancelFlag(job_id=row.id, cancel_phase=row.cancel_phase)
        for row in self._jobs.values()
        if row.cancel_requested_at is not None
        and row.status == "running"
        and row.locked_by_worker == worker_id
    ]

cancel_where async

cancel_where(
    filter: JobFilter, reason: str | None
) -> BulkCancelResult

Source code in src/taskq/testing/in_memory.py

async def cancel_where(
    self,
    filter: JobFilter,
    reason: str | None,
) -> BulkCancelResult:
    return await _cancel_where(self, filter, reason)

retry_job async

retry_job(job_id: JobId) -> bool

Source code in src/taskq/testing/in_memory.py

async def retry_job(self, job_id: JobId) -> bool:
    row = self._jobs.get(job_id)
    if row is None or row.status not in ("failed", "crashed", "cancelled"):
        return False
    self._jobs[job_id] = replace(
        row,
        status="pending",
        attempt=0,
        cancel_phase=CancelPhase.NONE,
        error_class=None,
        error_message=None,
        error_traceback=None,
        scheduled_at=self._clock.now(),
        finished_at=None,
        result=None,
        result_size_bytes=None,
        result_expires_at=None,
    )
    for event in self._wake_subscribers:
        event.set()
    return True

scheduled_to_pending async

scheduled_to_pending() -> int

Source code in src/taskq/testing/in_memory.py

async def scheduled_to_pending(self) -> int:
    return await _scheduled_to_pending(self)

deadline_sweep async

deadline_sweep() -> int

Source code in src/taskq/testing/in_memory.py

async def deadline_sweep(self) -> int:
    return await _deadline_sweep(self)

reclaim_expired_locks async

reclaim_expired_locks(
    cancel_grace: timedelta, cleanup_grace: timedelta
) -> int

Source code in src/taskq/testing/in_memory.py

async def reclaim_expired_locks(
    self,
    cancel_grace: timedelta,
    cleanup_grace: timedelta,
) -> int:
    return await _reclaim_expired_locks(self, cancel_grace, cleanup_grace)

archive_terminal_jobs

archive_terminal_jobs(
    retention: timedelta,
    archive_retention: timedelta,
    *,
    statuses: frozenset[str] | None = None,
) -> PruneResult

Source code in src/taskq/testing/in_memory.py

def archive_terminal_jobs(
    self,
    retention: timedelta,
    archive_retention: timedelta,
    *,
    statuses: frozenset[str] | None = None,
) -> "PruneResult":
    return _archive_terminal_jobs(self, retention, archive_retention, statuses=statuses)

expire_archived_jobs

expire_archived_jobs() -> ArchiveExpiryResult

Source code in src/taskq/testing/in_memory.py

def expire_archived_jobs(self) -> "ArchiveExpiryResult":
    return _expire_archived_jobs(self)

get_archived async

get_archived(job_id: JobId) -> _ArchivedJobRow | None

Source code in src/taskq/testing/in_memory.py

async def get_archived(self, job_id: JobId) -> _ArchivedJobRow | None:
    return await _get_archived(self, job_id)

get async

get(job_id: JobId) -> JobRow | None

Source code in src/taskq/testing/in_memory.py

async def get(self, job_id: JobId) -> JobRow | None:
    return await _get(self, job_id)

list_jobs async

list_jobs(filters: JobFilter) -> list[JobRow]

Source code in src/taskq/testing/in_memory.py

async def list_jobs(self, filters: JobFilter) -> list[JobRow]:
    return await _list_jobs(self, filters)

count_pending_jobs async

count_pending_jobs(actors: list[str]) -> dict[str, int]

Source code in src/taskq/testing/in_memory.py

async def count_pending_jobs(self, actors: list[str]) -> dict[str, int]:
    return await _count_pending_jobs(self, actors)

count_active_jobs async

count_active_jobs(queues: list[str]) -> int

Source code in src/taskq/testing/in_memory.py

async def count_active_jobs(self, queues: list[str]) -> int:
    if not queues:
        return 0
    queue_set = set(queues)
    return sum(
        1 for r in self._jobs.values() if r.queue in queue_set and r.status in ACTIVE_STATUSES
    )

get_actor_max_pending async

get_actor_max_pending() -> dict[str, int | None]

Source code in src/taskq/testing/in_memory.py

async def get_actor_max_pending(self) -> dict[str, int | None]:
    return await _get_actor_max_pending(self)

subscribe_wake

subscribe_wake() -> AsyncContextManager[asyncio.Event]

Source code in src/taskq/testing/in_memory.py

def subscribe_wake(self) -> AsyncContextManager[asyncio.Event]:
    event = asyncio.Event()
    return _SubscriberContext(event, self._wake_subscribers)

subscribe_cancel_wake

subscribe_cancel_wake() -> AsyncContextManager[
    asyncio.Event
]

Source code in src/taskq/testing/in_memory.py

def subscribe_cancel_wake(self) -> AsyncContextManager[asyncio.Event]:
    event = asyncio.Event()
    return _SubscriberContext(event, self._cancel_wake_subscribers)

tick_cancel_polling async

tick_cancel_polling() -> None

Source code in src/taskq/testing/in_memory.py

async def tick_cancel_polling(self) -> None:
    await _tick_cancel_polling(self)

run_until_drained async

run_until_drained() -> None

Source code in src/taskq/testing/in_memory.py

async def run_until_drained(self) -> None:
    await _run_until_drained(self)

create_schedule async

create_schedule(args: ScheduleCreateArgs) -> ScheduleRecord

Source code in src/taskq/testing/in_memory.py

async def create_schedule(self, args: ScheduleCreateArgs) -> ScheduleRecord:
    for rec in self._schedules.values():
        if rec.actor == args.actor and rec.name == args.name:
            raise ValueError(
                f"schedule for actor {args.actor!r} name {args.name!r} already exists"
            )
    sid = new_uuid()
    record = ScheduleRecord(
        id=sid,
        actor=args.actor,
        name=args.name,
        cron_expr=args.cron_expr,
        timezone=args.timezone,
        dst_strategy=args.dst_strategy,
        payload_factory=args.payload_factory,
        identity_key=args.identity_key,
        enabled=args.enabled,
        last_fired_at=None,
        last_fire_error=None,
        consecutive_failures=0,
        next_fire_at=args.next_fire_at,
        metadata=args.metadata,
    )
    self._schedules[sid] = record
    return record

list_schedules async

list_schedules(
    *, actor: str | None = None, enabled: bool | None = None
) -> list[ScheduleRecord]

Source code in src/taskq/testing/in_memory.py

async def list_schedules(
    self,
    *,
    actor: str | None = None,
    enabled: bool | None = None,
) -> list[ScheduleRecord]:
    results: list[ScheduleRecord] = []
    for rec in self._schedules.values():
        if actor is not None and rec.actor != actor:
            continue
        if enabled is not None and rec.enabled != enabled:
            continue
        results.append(rec)
    return results

update_schedule async

update_schedule(
    schedule_id: UUID, args: ScheduleUpdateArgs
) -> ScheduleRecord

Source code in src/taskq/testing/in_memory.py

async def update_schedule(
    self,
    schedule_id: UUID,
    args: ScheduleUpdateArgs,
) -> ScheduleRecord:
    rec = self._schedules.get(schedule_id)
    if rec is None:
        raise KeyError(f"schedule {schedule_id} not found")

    updates: dict[str, object] = {}
    if args.cron_expr is not None:
        updates["cron_expr"] = args.cron_expr
    if args.next_fire_at is not None:
        updates["next_fire_at"] = args.next_fire_at
    if args.enabled is not None:
        updates["enabled"] = args.enabled
        if args.enabled:
            updates["consecutive_failures"] = 0
            updates["last_fire_error"] = None
    if args.payload_factory is not None:
        updates["payload_factory"] = args.payload_factory
    elif args.clear_payload_factory:
        updates["payload_factory"] = None
    if args.metadata is not None:
        updates["metadata"] = args.metadata
    if args.consecutive_failures is not None:
        updates["consecutive_failures"] = args.consecutive_failures
    if args.last_fire_error is not None:
        updates["last_fire_error"] = args.last_fire_error

    updated = rec.model_copy(update=updates)
    self._schedules[schedule_id] = updated
    return updated

delete_schedule async

delete_schedule(schedule_id: UUID) -> None

Source code in src/taskq/testing/in_memory.py

async def delete_schedule(self, schedule_id: UUID) -> None:
    self._schedules.pop(schedule_id, None)

enqueue_batch_atomic async

enqueue_batch_atomic(
    items: Iterable[EnqueueArgs],
    *,
    batch_id: UUID,
    queue: str,
    batch_row: BatchRow | None,
    finalizer_args: EnqueueArgs | None,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> list[JobRow]

Source code in src/taskq/testing/in_memory.py

async def enqueue_batch_atomic(
    self,
    items: Iterable[EnqueueArgs],
    *,
    batch_id: UUID,
    queue: str,
    batch_row: BatchRow | None,
    finalizer_args: EnqueueArgs | None,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> list[JobRow]:
    return await _enqueue_batch_atomic(
        self,
        items,
        batch_id=batch_id,
        queue=queue,
        batch_row=batch_row,
        finalizer_args=finalizer_args,
        chunk_size=chunk_size,
    )

create_batch async

create_batch(
    batch_id: UUID,
    queue: str,
    expected_size: int,
    failure_threshold: int | None,
    finalizer_job_id: UUID | None,
    originating_actor: str | None,
    *,
    connection: object = None,
) -> None

Source code in src/taskq/testing/in_memory.py

async def create_batch(
    self,
    batch_id: UUID,
    queue: str,
    expected_size: int,
    failure_threshold: int | None,
    finalizer_job_id: UUID | None,
    originating_actor: str | None,
    *,
    connection: object = None,
) -> None:
    _create_batch(
        self,
        batch_id,
        queue,
        expected_size,
        failure_threshold,
        finalizer_job_id,
        originating_actor,
        connection,
    )

increment_batch_failures async

increment_batch_failures(
    batch_id: UUID, *, connection: object = None
) -> tuple[int, int | None, int]

Source code in src/taskq/testing/in_memory.py

async def increment_batch_failures(
    self,
    batch_id: UUID,
    *,
    connection: object = None,
) -> tuple[int, int | None, int]:
    return _increment_batch_failures(self, batch_id, connection)

reset_batch_failures async

reset_batch_failures(
    batch_id: UUID, *, connection: object = None
) -> int

Source code in src/taskq/testing/in_memory.py

async def reset_batch_failures(
    self,
    batch_id: UUID,
    *,
    connection: object = None,
) -> int:
    return _reset_batch_failures(self, batch_id, connection)

abort_batch async

abort_batch(
    batch_id: UUID, *, connection: object = None
) -> int

Source code in src/taskq/testing/in_memory.py

async def abort_batch(
    self,
    batch_id: UUID,
    *,
    connection: object = None,
) -> int:
    return _abort_batch(self, batch_id, connection)

complete_batch async

complete_batch(
    batch_id: UUID, *, connection: object = None
) -> None

Source code in src/taskq/testing/in_memory.py

async def complete_batch(
    self,
    batch_id: UUID,
    *,
    connection: object = None,
) -> None:
    _complete_batch(self, batch_id, connection)

get_batch async

get_batch(batch_id: UUID) -> BatchRow | None

Source code in src/taskq/testing/in_memory.py

async def get_batch(
    self,
    batch_id: UUID,
) -> BatchRow | None:
    return _get_batch(self, batch_id)

list_batches async

list_batches(
    filter: BatchFilter,
) -> list[tuple[BatchRow, BatchCounts]]

Source code in src/taskq/testing/in_memory.py

async def list_batches(
    self,
    filter: BatchFilter,
) -> list[tuple[BatchRow, BatchCounts]]:
    return _list_batches(self, filter)

count_batch_non_terminal async

count_batch_non_terminal(
    batch_id: UUID, *, connection: object = None
) -> int

Source code in src/taskq/testing/in_memory.py

async def count_batch_non_terminal(
    self,
    batch_id: UUID,
    *,
    connection: object = None,
) -> int:
    return _count_batch_non_terminal(self, batch_id)

prune_old_batches async

prune_old_batches(cutoff: datetime) -> int

Source code in src/taskq/testing/in_memory.py

async def prune_old_batches(
    self,
    cutoff: datetime,
) -> int:
    return _prune_old_batches(self, cutoff)

JobContext dataclass

JobContext(
    job_id: JobId,
    actor: str,
    queue: str,
    attempt: int,
    payload: P,
    cancel_event: Event,
    worker_id: UUID,
    jobs: SubJobEnqueuer,
    log: BoundLogger,
    deps: dict[str, object] | None = None,
    abort_requested: Event = threading.Event(),
)

Test-scoped context with deps for fixture-injected dependencies.

Field shape mirrors :class:taskq.context.JobContext (the production class) and adds deps. The bound on P matches production — payload is always a :class:pydantic.BaseModel. Tests that pass raw dicts as payload should validate them through a wrapper model (:class:taskq.testing.in_memory._PassthroughPayload is the permissive default for register_stub callers).

job_id instance-attribute

job_id: JobId

actor instance-attribute

actor: str

queue instance-attribute

queue: str

attempt instance-attribute

attempt: int

payload instance-attribute

payload: P

cancel_event instance-attribute

cancel_event: Event

worker_id instance-attribute

worker_id: UUID

jobs instance-attribute

jobs: SubJobEnqueuer

log instance-attribute

log: BoundLogger

deps class-attribute instance-attribute

deps: dict[str, object] | None = field(default=None)

abort_requested class-attribute instance-attribute

abort_requested: Event = field(
    default_factory=threading.Event
)

cancellation_requested property

cancellation_requested: bool

True when the cancel event has been set.

should_abort

should_abort() -> bool

Synchronous cancellation check for sync actors (thread-safe).

Source code in src/taskq/testing/job_context.py

def should_abort(self) -> bool:
    """Synchronous cancellation check for sync actors (thread-safe)."""
    return self.abort_requested.is_set()

JobTriple

Bases: NamedTuple

row instance-attribute

row: Record

attempts instance-attribute

attempts: list[Record]

events instance-attribute

events: list[Record]

WarningSpy

WarningSpy()

Records how many times warning() was called, without sniffing args.

Source code in src/taskq/testing/spy.py

def __init__(self) -> None:
    self.warning_count = 0

warning_count instance-attribute

warning_count = 0

warning

warning(*_args: object, **_kwargs: object) -> None

Source code in src/taskq/testing/spy.py

def warning(self, *_args: object, **_kwargs: object) -> None:
    self.warning_count += 1

as_backend

as_backend(fb: FakeBackend) -> Backend

Source code in src/taskq/testing/actor.py

def as_backend(fb: FakeBackend) -> Backend:
    return cast(Backend, fb)

default_actor_config

default_actor_config() -> StubActorConfig

Source code in src/taskq/testing/actor.py

def default_actor_config() -> StubActorConfig:
    return StubActorConfig(retry=RetryPolicy(kind="transient", max_attempts=3, jitter=0.0))

assert_attempt

assert_attempt(
    attempts: Sequence[object],
    index: int,
    *,
    outcome: str | None = None,
    error_class: str | None = None,
    attempt_num: int | None = None,
) -> object

Assert on the attempt row at index.

Source code in src/taskq/testing/assertions.py

def assert_attempt(
    attempts: Sequence[object],
    index: int,
    *,
    outcome: str | None = None,
    error_class: str | None = None,
    attempt_num: int | None = None,
) -> object:
    """Assert on the attempt row at *index*."""
    if index < 0 or index >= len(attempts):
        raise AssertionError(f"Attempt index {index} out of range (len={len(attempts)})")
    row = attempts[index]
    if outcome is not None:
        actual = _get(row, "outcome")
        if actual != outcome:
            raise AssertionError(f"attempts[{index}]: expected outcome {outcome!r}, got {actual!r}")
    if error_class is not None:
        actual = _get(row, "error_class")
        if actual != error_class:
            raise AssertionError(
                f"attempts[{index}]: expected error_class {error_class!r}, got {actual!r}"
            )
    if attempt_num is not None:
        actual = _get(row, "attempt")
        if actual != attempt_num:
            raise AssertionError(f"attempts[{index}]: expected attempt {attempt_num}, got {actual}")
    return row

assert_has_event

assert_has_event(
    events: Sequence[Record],
    kind: str,
    *,
    from_state: str | None = None,
    to_state: str | None = None,
) -> asyncpg.Record

Find at least one event matching kind and optional state filters.

Source code in src/taskq/testing/assertions.py

def assert_has_event(
    events: Sequence[asyncpg.Record],
    kind: str,
    *,
    from_state: str | None = None,
    to_state: str | None = None,
) -> asyncpg.Record:
    """Find at least one event matching kind and optional state filters."""
    for ev in events:
        if _get(ev, "kind") != kind:
            continue
        if from_state is not None or to_state is not None:
            detail = parse_detail(_get(ev, "detail"))
            if from_state is not None and detail.get("from_state") != from_state:
                continue
            if to_state is not None and detail.get("to_state") != to_state:
                continue
        return ev
    available = [(i, _get(e, "kind")) for i, e in enumerate(events)]
    msg = f"No event with kind={kind!r}"
    if from_state is not None or to_state is not None:
        msg += f" (from_state={from_state!r}, to_state={to_state!r})"
    msg += f"; available events: {available}"
    raise AssertionError(msg)

assert_has_otel_event

assert_has_otel_event(
    exporter: _SpanExporter,
    span_name: str,
    event_name: str,
    *,
    from_state: str | None = None,
    to_state: str | None = None,
) -> object

Find an OTel event by span and event name; assert state attributes if provided.

Source code in src/taskq/testing/assertions.py

def assert_has_otel_event(
    exporter: _SpanExporter,
    span_name: str,
    event_name: str,
    *,
    from_state: str | None = None,
    to_state: str | None = None,
) -> object:
    """Find an OTel event by span and event name; assert state attributes if provided."""
    events = exporter.events_on(span_name, event_name)
    if not events:
        raise AssertionError(f"No OTel event {event_name!r} on span {span_name!r}")
    if from_state is not None or to_state is not None:
        for ev in events:
            attrs = ev.attributes
            if attrs is None:
                continue
            if from_state is not None and attrs.get("from_state") != from_state:
                continue
            if to_state is not None and attrs.get("to_state") != to_state:
                continue
            return ev
        raise AssertionError(
            f"No OTel event {event_name!r} on span {span_name!r} "
            f"with from_state={from_state!r}, to_state={to_state!r}"
        )
    return events[0]

assert_has_span

assert_has_span(
    exporter: _SpanExporter,
    name: str,
    *,
    kind: object = None,
    status: object = None,
) -> ReadableSpan

Find a span by name on the exporter; assert kind/status if provided.

Source code in src/taskq/testing/assertions.py

def assert_has_span(
    exporter: _SpanExporter,
    name: str,
    *,
    kind: object = None,
    status: object = None,
) -> ReadableSpan:
    """Find a span by name on the exporter; assert kind/status if provided."""
    span = exporter.span_named(name)
    if span is None:
        names = [s.name for s in exporter.spans]
        raise AssertionError(f"No span named {name!r}; available: {names}")
    if kind is not None and span.kind != kind:
        raise AssertionError(f"Span {name!r}: expected kind={kind!r}, got {span.kind!r}")
    if status is not None and span.status.status_code != status:
        raise AssertionError(
            f"Span {name!r}: expected status_code={status!r}, got {span.status.status_code!r}"
        )
    return span

assert_job_status

assert_job_status(
    row: Record | None,
    status: str,
    *,
    error_class: str | None = None,
    attempt: int | None = None,
    finished: bool | None = None,
) -> asyncpg.Record

Assert a job row has the expected status and optional fields.

Returns the row (guaranteed non-None on success) so callers can chain attribute/subscript access without pyright narrowing issues.

Source code in src/taskq/testing/assertions.py

def assert_job_status(
    row: asyncpg.Record | None,
    status: str,
    *,
    error_class: str | None = None,
    attempt: int | None = None,
    finished: bool | None = None,
) -> asyncpg.Record:
    """Assert a job row has the expected status and optional fields.

    Returns the row (guaranteed non-None on success) so callers can
    chain attribute/subscript access without pyright narrowing issues.
    """
    assert row is not None, "Expected a row but got None"
    actual_status = _get(row, "status")
    if actual_status != status:
        raise AssertionError(f"Expected status {status!r}, got {actual_status!r}")
    if error_class is not None:
        actual_ec = _get(row, "error_class")
        if actual_ec != error_class:
            raise AssertionError(f"Expected error_class {error_class!r}, got {actual_ec!r}")
    if attempt is not None:
        actual_attempt = _get(row, "attempt")
        if actual_attempt != attempt:
            raise AssertionError(f"Expected attempt {attempt}, got {actual_attempt}")
    if finished is not None:
        finished_at = _get(row, "finished_at")
        if finished and finished_at is None:
            raise AssertionError("Expected finished_at to be set, but it is None")
        if not finished and finished_at is not None:
            raise AssertionError(f"Expected finished_at to be None, got {finished_at!r}")
    return row

assert_job_terminal

assert_job_terminal(
    row: Record | None,
    status: str,
    *,
    error_class: str | None = None,
) -> asyncpg.Record

Assert a job is in a terminal status with finished_at set.

Source code in src/taskq/testing/assertions.py

def assert_job_terminal(
    row: asyncpg.Record | None,
    status: str,
    *,
    error_class: str | None = None,
) -> asyncpg.Record:
    """Assert a job is in a terminal status with finished_at set."""
    return assert_job_status(row, status, error_class=error_class, finished=True)

assert_transition_sequence

assert_transition_sequence(
    events: Sequence[object],
    expected: Sequence[tuple[str | None, str | None]],
) -> None

Assert the (from_state, to_state) sequence from state_change events matches expected.

Source code in src/taskq/testing/assertions.py

def assert_transition_sequence(
    events: Sequence[object],
    expected: Sequence[tuple[str | None, str | None]],
) -> None:
    """Assert the (from_state, to_state) sequence from state_change events matches expected."""
    transitions: list[tuple[object, object]] = []
    for ev in events:
        if _get(ev, "kind") != "state_change":
            continue
        detail = parse_detail(_get(ev, "detail"))
        transitions.append((detail.get("from_state"), detail.get("to_state")))
    if transitions != list(expected):
        raise AssertionError(f"Expected transition sequence {list(expected)}, got {transitions}")

parse_detail

parse_detail(detail: object) -> dict[str, object]

Normalize a detail value (dict, JSON string, or other) to a dict.

Source code in src/taskq/testing/assertions.py

def parse_detail(detail: object) -> dict[str, object]:
    """Normalize a detail value (dict, JSON string, or other) to a dict."""
    if isinstance(detail, dict):
        return detail  # type: ignore[return-value]  # Why: isinstance(detail, dict) guarantees a dict at runtime; the value type is object so pyright cannot narrow dict[unknown, unknown] to dict[str, object].
    if isinstance(detail, str):
        return dict(loads(detail))
    return {}

pg_now async

pg_now(conn: Connection) -> datetime

Return PG's clock_timestamp() — the realtime clock the server uses.

Use this instead of datetime.now(UTC) when a test needs to compute cutoffs/margins that are compared against rows written via SQL: the Python wall clock and PG's realtime clock can diverge enough under parallel load to make Python-computed margins flaky.

Source code in src/taskq/testing/assertions.py

async def pg_now(conn: asyncpg.Connection) -> datetime:
    """Return PG's ``clock_timestamp()`` — the realtime clock the server uses.

    Use this instead of ``datetime.now(UTC)`` when a test needs to compute
    cutoffs/margins that are compared against rows written via SQL: the
    Python wall clock and PG's realtime clock can diverge enough under
    parallel load to make Python-computed margins flaky.
    """
    value: datetime = await conn.fetchval("SELECT clock_timestamp()")
    return value

wait_for async

wait_for(event: Event, timeout: float = 2.0) -> None

Wait for an asyncio.Event with test-failure semantics on timeout.

Source code in src/taskq/testing/assertions.py

async def wait_for(event: asyncio.Event, timeout: float = 2.0) -> None:  # noqa: ASYNC109
    """Wait for an asyncio.Event with test-failure semantics on timeout."""
    try:
        await asyncio.wait_for(event.wait(), timeout=timeout)
    except TimeoutError:
        raise AssertionError(f"Event not set within {timeout}s") from None

wait_for_job_status async

wait_for_job_status(
    backend: _AssertBackend,
    job_id: JobId,
    status: str,
    *,
    timeout: float = 2.0,
    poll_interval: float = 0.05,
) -> JobRow

Poll backend.get until the job reaches the expected status.

Source code in src/taskq/testing/assertions.py

async def wait_for_job_status(
    backend: _AssertBackend,
    job_id: JobId,
    status: str,
    *,
    timeout: float = 2.0,  # noqa: ASYNC109
    poll_interval: float = 0.05,
) -> JobRow:
    """Poll backend.get until the job reaches the expected status."""
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    while True:
        row = await backend.get(job_id)
        if row is not None and _get(row, "status") == status:
            return row
        remaining = deadline - loop.time()
        if remaining <= 0:
            actual = _get(row, "status") if row is not None else None
            raise AssertionError(
                f"Job {job_id} did not reach status {status!r} within {timeout}s "
                f"(actual: {actual!r})"
            )
        await asyncio.sleep(min(poll_interval, remaining))

wait_for_leader async

wait_for_leader(
    deps: _LeaderDeps, timeout: float = 5.0
) -> None

Wait for the leader event on WorkerDeps with test-failure semantics.

Source code in src/taskq/testing/assertions.py

async def wait_for_leader(deps: _LeaderDeps, timeout: float = 5.0) -> None:  # noqa: ASYNC109
    """Wait for the leader event on WorkerDeps with test-failure semantics."""
    try:
        await asyncio.wait_for(deps.is_leader.wait(), timeout=timeout)
    except TimeoutError:
        raise AssertionError(f"Leader event not set within {timeout}s") from None

unique_health_sock_path

unique_health_sock_path(module: str) -> str

Return a unique unix-socket path for one test's health server.

/tmp/tq-<module>-<pid>-<token>.sock: the pid scopes across xdist workers, the random token across tests within a worker (stateless — no shared counter, so uniqueness survives any pytest import mode), and the module label identifies the owner when debugging stale files. The short /tmp/tq- prefix keeps paths well under the 104-char AF_UNIX sun_path limit on macOS.

Stale files are a resource-only leak: the per-run-unique token means a leftover path can never collide with a future run. HealthServer.stop unlinks on normal completion; suites wanting a crash backstop can sweep /tmp/tq-*-<pid>-*.sock scoped to their own pid (see TaskQ's tests/conftest.py::_sweep_health_sock_files).

:param module: label embedded in the path. Must not contain path separators — / would point the socket at a nonexistent directory and surface as a confusing ENOENT at bind time. :raises ValueError: if module contains / or \.

Source code in src/taskq/testing/health.py

def unique_health_sock_path(module: str) -> str:
    """Return a unique unix-socket path for one test's health server.

    ``/tmp/tq-<module>-<pid>-<token>.sock``: the pid scopes across xdist
    workers, the random token across tests within a worker (stateless — no
    shared counter, so uniqueness survives any pytest import mode), and the
    module label identifies the owner when debugging stale files. The short
    ``/tmp/tq-`` prefix keeps paths well under the 104-char AF_UNIX
    sun_path limit on macOS.

    Stale files are a resource-only leak: the per-run-unique token means a
    leftover path can never collide with a future run. ``HealthServer.stop``
    unlinks on normal completion; suites wanting a crash backstop can sweep
    ``/tmp/tq-*-<pid>-*.sock`` scoped to their own pid (see TaskQ's
    ``tests/conftest.py::_sweep_health_sock_files``).

    :param module: label embedded in the path. Must not contain path
        separators — ``/`` would point the socket at a nonexistent
        directory and surface as a confusing ENOENT at bind time.
    :raises ValueError: if *module* contains ``/`` or ``\\``.
    """
    if "/" in module or "\\" in module:
        raise ValueError(
            f"module label must not contain path separators, got {module!r} "
            "(it is embedded verbatim in the socket path)"
        )
    return f"/tmp/tq-{module}-{os.getpid()}-{new_base62()}.sock"  # noqa: S108  # Why: test-only socket files; /tmp is the shortest safe prefix.

error_info

error_info(
    error_class: str = "ValueError",
    error_message: str = "boom",
) -> ErrorInfo

Shorthand for ErrorInfo with error_traceback=None.

Source code in src/taskq/testing/jobs.py

def error_info(
    error_class: str = "ValueError",
    error_message: str = "boom",
) -> ErrorInfo:
    """Shorthand for ErrorInfo with error_traceback=None."""
    return ErrorInfo(
        error_class=error_class,
        error_message=error_message,
        error_traceback=None,
    )

make_enqueue_args

make_enqueue_args(
    *,
    actor: str = "test_actor",
    queue: str = "default",
    payload: dict[str, object] | None = None,
    idempotency_key: str | None = None,
    idempotency_scope: str = "",
    identity_key: str | None = None,
    scheduled_at: datetime | None = None,
    max_attempts: int = 3,
    retry_kind: RetryKind = "transient",
    priority: int = 0,
    schedule_to_close: datetime | None = None,
    metadata: dict[str, object] | None = None,
    tags: tuple[str, ...] | None = None,
) -> EnqueueArgs

Build EnqueueArgs with sensible defaults.

scheduled_at defaults to 1 second in the past (relative to the Python wall clock) so freshly-enqueued test jobs can never classify as a future-"scheduled" job: the enqueue SQL compares against PG's clock_timestamp(), and Python's monotonic/wall clock can diverge from PG's realtime clock enough under parallel load that a "now" computed here reads as still-future by the time the row lands.

Source code in src/taskq/testing/jobs.py

def make_enqueue_args(
    *,
    actor: str = "test_actor",
    queue: str = "default",
    payload: dict[str, object] | None = None,
    idempotency_key: str | None = None,
    idempotency_scope: str = "",
    identity_key: str | None = None,
    scheduled_at: datetime | None = None,
    max_attempts: int = 3,
    retry_kind: RetryKind = "transient",
    priority: int = 0,
    schedule_to_close: datetime | None = None,
    metadata: dict[str, object] | None = None,
    tags: tuple[str, ...] | None = None,
) -> EnqueueArgs:
    """Build EnqueueArgs with sensible defaults.

    ``scheduled_at`` defaults to 1 second in the past (relative to the
    Python wall clock) so freshly-enqueued test jobs can never classify
    as a future-"scheduled" job: the enqueue SQL compares against PG's
    ``clock_timestamp()``, and Python's monotonic/wall clock can diverge
    from PG's realtime clock enough under parallel load that a "now"
    computed here reads as still-future by the time the row lands.
    """
    return EnqueueArgs(
        id=new_job_id(),
        actor=actor,
        queue=queue,
        payload=payload or {"value": 1},
        max_attempts=max_attempts,
        retry_kind=retry_kind,
        scheduled_at=scheduled_at or (datetime.now(UTC) - timedelta(seconds=1)),
        priority=priority,
        schedule_to_close=schedule_to_close,
        idempotency_key=IdempotencyKey(idempotency_key) if idempotency_key is not None else None,
        idempotency_scope=idempotency_scope,
        identity_key=IdentityKey(identity_key) if identity_key is not None else None,
        metadata=metadata or {},
        tags=tags if tags is not None else (),
    )

make_job_row

make_job_row(
    *,
    attempt: int = 1,
    max_attempts: int = 3,
    retry_kind: RetryKind = "transient",
    schedule_to_close: datetime | None = None,
    identity_key: IdentityKey | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
    cancel_phase: int | CancelPhase | None = None,
    status: JobStatus = "running",
    priority: int = 0,
    error_class: str | None = None,
    error_message: str | None = None,
    payload: dict[str, object] | None = None,
    queue: str = "default",
    actor: str = "test_actor",
    progress_seq: int = 0,
) -> JobRow

Build a JobRow with sensible defaults.

Source code in src/taskq/testing/jobs.py

def make_job_row(
    *,
    attempt: int = 1,
    max_attempts: int = 3,
    retry_kind: RetryKind = "transient",
    schedule_to_close: datetime | None = None,
    identity_key: IdentityKey | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    start_to_close: timedelta | None = None,
    heartbeat_timeout: timedelta | None = None,
    cancel_phase: int | CancelPhase | None = None,
    status: JobStatus = "running",
    priority: int = 0,
    error_class: str | None = None,
    error_message: str | None = None,
    payload: dict[str, object] | None = None,
    queue: str = "default",
    actor: str = "test_actor",
    progress_seq: int = 0,
) -> JobRow:
    """Build a JobRow with sensible defaults."""
    phase: CancelPhase
    if cancel_phase is None:
        phase = CancelPhase.NONE
    elif isinstance(cancel_phase, CancelPhase):
        phase = cancel_phase
    else:
        phase = CancelPhase(cancel_phase)

    locked_by = _WORKER_ID if status == "running" else None

    return JobRow(
        id=new_job_id(),
        actor=actor,
        queue=queue,
        identity_key=identity_key,
        fairness_key=None,
        payload=payload or {},
        payload_schema_ver=1,
        status=status,
        priority=priority,
        attempt=attempt,
        max_attempts=max_attempts,
        retry_kind=retry_kind,
        schedule_to_close=schedule_to_close,
        start_to_close=start_to_close,
        heartbeat_timeout=heartbeat_timeout,
        created_at=_NOW,
        scheduled_at=_NOW,
        started_at=_NOW if status == "running" else None,
        finished_at=None,
        last_heartbeat_at=None,
        locked_by_worker=locked_by,
        lock_expires_at=None,
        cancel_requested_at=None,
        cancel_phase=phase,
        error_class=error_class,
        error_message=error_message,
        error_traceback=None,
        progress_state={},
        progress_seq=progress_seq,
        result=None,
        result_size_bytes=None,
        result_expires_at=None,
        idempotency_key=None,
        idempotency_scope="",
        trace_id=trace_id,
        span_id=span_id,
        metadata={},
        tags=(),
    )

create_pending_job async

create_pending_job(
    conn: _Conn,
    schema: str,
    job_id: UUID | None = None,
    *,
    schedule_to_close: datetime | None = None,
    status: str = "pending",
    scheduled_at: datetime | None = None,
) -> UUID

Source code in src/taskq/testing/pg.py

async def create_pending_job(
    conn: _Conn,
    schema: str,
    job_id: UUID | None = None,
    *,
    schedule_to_close: datetime | None = None,
    status: str = "pending",
    scheduled_at: datetime | None = None,
) -> UUID:
    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema name {schema!r}")
    job_id = job_id or new_uuid()
    stc = schedule_to_close or (datetime.now(UTC) + timedelta(seconds=60))
    sa = scheduled_at or datetime.now(UTC)
    await conn.execute(
        f"""INSERT INTO "{schema}".jobs (
            id, actor, queue, payload, max_attempts, retry_kind,
            status, priority, scheduled_at, schedule_to_close
        ) VALUES (
            $1, $2, $3, $4::jsonb, $5, $6,
            $7, 0, $8, $9
        )""",  # noqa: S608
        job_id,
        "test_actor",
        "default",
        '{"key": "value"}',
        3,
        "transient",
        status,
        sa,
        stc,
    )
    return job_id

create_running_job async

create_running_job(
    conn: _Conn,
    schema: str,
    worker_id: UUID,
    job_id: UUID | None = None,
    *,
    cancel_phase: int = 0,
    max_attempts: int = 3,
    retry_kind: str = "transient",
    attempt: int = 1,
    cancel_requested_at: datetime | None = None,
    lock_expires_at: datetime | None = None,
    schedule_to_close: datetime | None = None,
    with_events: bool = True,
) -> UUID

Source code in src/taskq/testing/pg.py

async def create_running_job(
    conn: _Conn,
    schema: str,
    worker_id: UUID,
    job_id: UUID | None = None,
    *,
    cancel_phase: int = 0,
    max_attempts: int = 3,
    retry_kind: str = "transient",
    attempt: int = 1,
    cancel_requested_at: datetime | None = None,
    lock_expires_at: datetime | None = None,
    schedule_to_close: datetime | None = None,
    with_events: bool = True,
) -> UUID:
    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema name {schema!r}")
    job_id = job_id or new_uuid()
    expires_at = lock_expires_at or (datetime.now(UTC) + timedelta(seconds=60))
    now = datetime.now(UTC)
    await conn.execute(
        f"""INSERT INTO "{schema}".jobs (
            id, actor, queue, payload, max_attempts, retry_kind,
            status, priority, attempt, scheduled_at,
            locked_by_worker, lock_expires_at, started_at, last_heartbeat_at,
            cancel_phase, cancel_requested_at, schedule_to_close
        ) VALUES (
            $1, $2, $3, $4::jsonb, $5, $6,
            'running', 0, $7, clock_timestamp(),
            $8, $9, clock_timestamp(), clock_timestamp(),
            $10, $11, $12
        )""",  # noqa: S608
        job_id,
        "test_actor",
        "default",
        '{"key": "value"}',
        max_attempts,
        retry_kind,
        attempt,
        worker_id,
        expires_at,
        cancel_phase,
        cancel_requested_at,
        schedule_to_close,
    )
    if with_events:
        detail = dumps_str(
            {"from_state": "pending", "to_state": "running", "worker_id": str(worker_id)}
        )
        await conn.execute(
            f'INSERT INTO "{schema}".job_events (job_id, occurred_at, kind, detail) '  # noqa: S608
            "VALUES ($1, $2, 'state_change', $3::jsonb)",
            job_id,
            now,
            detail,
        )
    return job_id

create_workered_running_job async

create_workered_running_job(
    conn: _Conn,
    schema: str,
    *,
    worker_id: UUID | None = None,
    **job_kwargs: Any,
) -> tuple[UUID, UUID]

Create a worker row and a running job row, returning (worker_id, job_id).

Passthrough wrapper: creates a worker (generating a UUID if none provided), then creates a running job belonging to that worker. All extra keyword arguments are forwarded to :func:create_running_job.

Source code in src/taskq/testing/pg.py

async def create_workered_running_job(
    conn: _Conn,
    schema: str,
    *,
    worker_id: UUID | None = None,
    **job_kwargs: Any,
) -> tuple[UUID, UUID]:
    """Create a worker row and a running job row, returning ``(worker_id, job_id)``.

    Passthrough wrapper: creates a worker (generating a UUID if none provided),
    then creates a running job belonging to that worker.  All extra keyword
    arguments are forwarded to :func:`create_running_job`.
    """
    wid = worker_id or new_uuid()
    await _create_worker(conn, schema, wid)
    jid = await create_running_job(conn, schema, wid, **job_kwargs)
    return wid, jid

get_job_triple async

get_job_triple(
    conn: _Conn, schema: str, job_id: UUID
) -> JobTriple

Source code in src/taskq/testing/pg.py

async def get_job_triple(conn: _Conn, schema: str, job_id: UUID) -> JobTriple:
    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema name {schema!r}")
    row = await conn.fetchrow(
        f'SELECT * FROM "{schema}".jobs WHERE id = $1',  # noqa: S608
        job_id,
    )
    assert row is not None
    attempts = await conn.fetch(
        f'SELECT * FROM "{schema}".job_attempts WHERE job_id = $1',  # noqa: S608
        job_id,
    )
    events = await conn.fetch(
        f'SELECT * FROM "{schema}".job_events WHERE job_id = $1 ORDER BY occurred_at',  # noqa: S608
        job_id,
    )
    return JobTriple(row=row, attempts=list(attempts), events=list(events))

reset_schema async

reset_schema(
    conn: _Conn,
    schema: str,
    *,
    actors: Sequence[str] | None = None,
) -> None

Truncate all dynamic tables then seed default actor_config rows.

Tests needing a custom actor set can pass actors=[...]; tests that need an empty actor_config can pass actors=[].

Source code in src/taskq/testing/pg.py

async def reset_schema(
    conn: _Conn,
    schema: str,
    *,
    actors: Sequence[str] | None = None,
) -> None:
    """Truncate all dynamic tables then seed default actor_config rows.

    Tests needing a custom actor set can pass ``actors=[...]``;
    tests that need an empty actor_config can pass ``actors=[]``.
    """
    await truncate_schema(conn, schema)
    await seed_actors(conn, schema, actors=actors)

seed_actors async

seed_actors(
    conn: _Conn,
    schema: str,
    *,
    actors: Sequence[str] | None = None,
) -> None

Insert actor_config rows for the given actors (or DEFAULT_ACTORS).

ON CONFLICT (actor) DO NOTHING makes this safe to call alongside custom seed data — it never overwrites existing rows.

Source code in src/taskq/testing/pg.py

async def seed_actors(
    conn: _Conn,
    schema: str,
    *,
    actors: Sequence[str] | None = None,
) -> None:
    """Insert actor_config rows for the given actors (or DEFAULT_ACTORS).

    ``ON CONFLICT (actor) DO NOTHING`` makes this safe to call
    alongside custom seed data — it never overwrites existing rows.
    """
    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema name {schema!r}")
    target = actors if actors is not None else DEFAULT_ACTORS
    await conn.executemany(
        f'INSERT INTO "{schema}".actor_config (actor, queue) VALUES ($1, $2) ON CONFLICT (actor) DO NOTHING',  # noqa: S608
        [(actor, "default") for actor in target],
    )

setup_running_job async

setup_running_job(
    conn: _Conn,
    schema: str,
    *,
    worker_id: UUID | None = None,
    job_id: UUID | None = None,
    attempt: int = 1,
    max_attempts: int = 3,
    retry_kind: str = "transient",
    cancel_phase: int = 0,
    cancel_requested_at: datetime | None = None,
    lock_expires_at: datetime | None = None,
    schedule_to_close: datetime | None = None,
    with_events: bool = True,
) -> tuple[UUID, UUID]

Create a worker row and a running job row in one call.

Returns (worker_id, job_id). Delegates to :func:create_workered_running_job.

Source code in src/taskq/testing/pg.py

async def setup_running_job(
    conn: _Conn,
    schema: str,
    *,
    worker_id: UUID | None = None,
    job_id: UUID | None = None,
    attempt: int = 1,
    max_attempts: int = 3,
    retry_kind: str = "transient",
    cancel_phase: int = 0,
    cancel_requested_at: datetime | None = None,
    lock_expires_at: datetime | None = None,
    schedule_to_close: datetime | None = None,
    with_events: bool = True,
) -> tuple[UUID, UUID]:
    """Create a worker row and a running job row in one call.

    Returns ``(worker_id, job_id)``.  Delegates to
    :func:`create_workered_running_job`.
    """
    return await create_workered_running_job(
        conn,
        schema,
        worker_id=worker_id,
        job_id=job_id,
        cancel_phase=cancel_phase,
        max_attempts=max_attempts,
        retry_kind=retry_kind,
        attempt=attempt,
        cancel_requested_at=cancel_requested_at,
        lock_expires_at=lock_expires_at,
        schedule_to_close=schedule_to_close,
        with_events=with_events,
    )

truncate_schema async

truncate_schema(conn: _Conn, schema: str) -> None

Truncate all dynamic tables in FK-safe order using CASCADE.

Leaves schema_migrations intact. Safe to call repeatedly.

Source code in src/taskq/testing/pg.py

async def truncate_schema(conn: _Conn, schema: str) -> None:
    """Truncate all dynamic tables in FK-safe order using CASCADE.

    Leaves ``schema_migrations`` intact.  Safe to call repeatedly.
    """
    if not _IDENT_RE.match(schema):
        raise ValueError(f"invalid schema name {schema!r}")
    for table in _TRUNCATE_TABLES:
        await conn.execute(f'TRUNCATE TABLE "{schema}"."{table}" CASCADE')

make_integration_settings

make_integration_settings(
    pg_dsn: str, **overrides: str
) -> WorkerSettings

Construct WorkerSettings with fast intervals for integration tests.

The schema defaults to a per-call unique tq_<token> name (see module docstring); pass schema_name="..." to pin it.

Source code in src/taskq/testing/settings.py

def make_integration_settings(pg_dsn: str, **overrides: str) -> WorkerSettings:
    """Construct WorkerSettings with fast intervals for integration tests.

    The schema defaults to a per-call unique ``tq_<token>`` name (see module
    docstring); pass ``schema_name="..."`` to pin it.
    """
    return WorkerSettings.load_from_dict(_build_dict(pg_dsn, **overrides))

make_integration_settings_dict

make_integration_settings_dict(
    pg_dsn: str, **overrides: str
) -> dict[str, str]

Return the raw dict passed to WorkerSettings.load_from_dict.

Same per-call unique schema default as :func:make_integration_settings.

Source code in src/taskq/testing/settings.py

def make_integration_settings_dict(pg_dsn: str, **overrides: str) -> dict[str, str]:
    """Return the raw dict passed to WorkerSettings.load_from_dict.

    Same per-call unique schema default as :func:`make_integration_settings`.
    """
    return _build_dict(pg_dsn, **overrides)

Pytest fixtures

The fixtures are not re-exported from taskq.testing.__init__ (importing pytest/asyncpg at the package top level is deliberately avoided), so they render from their defining module:

fixtures

__all__ module-attribute

__all__ = [
    "ActorRunnerCallable",
    "JobsApp",
    "ModulePgSchema",
    "_create_worker",
    "actor_runner",
    "backend_pair",
    "clean_jobs_app",
    "clean_pg_conn",
    "clean_redis_client",
    "clean_redis_url",
    "jobs_app",
    "killable_redis_container",
    "memory_jobs",
    "module_jobs_app",
    "module_pg_pool",
    "module_pg_schema",
    "module_redis_url",
    "redis_container",
    "redis_url",
    "redis_url_for",
    "worker_with_running_job",
]

RUN_TOKEN_ENV_VAR module-attribute

RUN_TOKEN_ENV_VAR = 'TASKQ_TEST_RUN_TOKEN'

JobsApp

Bases: NamedTuple

deps instance-attribute

deps: object

backend instance-attribute

backend: object

ModulePgSchema

Bases: NamedTuple

schema_name instance-attribute

schema_name: str

pg_dsn instance-attribute

pg_dsn: str

ActorRunnerCallable

Bases: Protocol

Protocol for the callable yielded by the actor_runner fixture.

payload is permissively typed: a :class:pydantic.BaseModel (typed actor payload) or a dict[str, object] / object that the runner wraps in a :class:PassthroughPayload model. The JobContext.payload handed to the actor is always a :class:pydantic.BaseModel per the locked architecture; the coercion happens here so test authors don't have to declare a model for ad-hoc payloads.

__call__ async

__call__(
    actor_fn: Callable[..., object],
    payload: BaseModel | dict[str, object] | object,
    *,
    backend: InMemoryBackend,
    job_id: JobId | UUID | None = ...,
    attempt: int = ...,
    cancel_event: Event | None = ...,
    actor: str = ...,
    queue: str = ...,
    **deps: object,
) -> object

Source code in src/taskq/testing/fixtures.py

async def __call__(
    self,
    actor_fn: Callable[..., object],
    payload: BaseModel | dict[str, object] | object,
    *,
    backend: InMemoryBackend,
    job_id: JobId | UUID | None = ...,
    attempt: int = ...,
    cancel_event: asyncio.Event | None = ...,
    actor: str = ...,
    queue: str = ...,
    **deps: object,
) -> object: ...

RedisContainerLike

Bases: Protocol

The RedisContainer surface redis_url_for needs — structural, so both the real testcontainers object (killable_redis_container) and the shared-pair shim (redis_container) satisfy it.

get_container_host_ip

get_container_host_ip() -> str

Source code in src/taskq/testing/fixtures.py

def get_container_host_ip(self) -> str: ...

get_exposed_port

get_exposed_port(port: int) -> int

Source code in src/taskq/testing/fixtures.py

def get_exposed_port(self, port: int) -> int: ...

memory_jobs async

memory_jobs() -> AsyncIterator[InMemoryBackend]

Yield a fresh InMemoryBackend with a FakeClock starting at datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC). Default cancellation and cleanup grace from InMemoryBackend.__init__ (30s each). No teardown beyond GC (fully isolated per fixture instance — each call constructs a new backend and clock; nothing is shared at module level).

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture
async def memory_jobs() -> AsyncIterator[InMemoryBackend]:
    """Yield a fresh ``InMemoryBackend`` with a ``FakeClock`` starting at
    ``datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC)``.  Default cancellation
    and cleanup grace from ``InMemoryBackend.__init__`` (30s each).
    No teardown beyond GC (fully isolated per fixture instance — each call
    constructs a new backend and clock; nothing is shared at module level).
    """
    clock = FakeClock(start=datetime(2025, 1, 1, tzinfo=UTC))
    backend = InMemoryBackend(clock=clock)
    for actor in DEFAULT_ACTORS:
        backend.register_actor_config(actor=actor)
    yield backend

actor_runner

actor_runner() -> ActorRunnerCallable

Yield a callable that constructs a synthetic JobContext and calls actor_fn(payload, ctx).

Accepts cancel_event to test cancellation paths and **deps to forward ad-hoc keyword-injected collaborators (e.g. stub HTTP clients or database sessions) directly to actor_fn without wiring the full DI scope hierarchy.

Source code in src/taskq/testing/fixtures.py

@pytest.fixture
def actor_runner() -> ActorRunnerCallable:
    """Yield a callable that constructs a synthetic ``JobContext`` and
    calls ``actor_fn(payload, ctx)``.

    Accepts ``cancel_event`` to test cancellation paths and ``**deps``
    to forward ad-hoc keyword-injected collaborators (e.g. stub HTTP
    clients or database sessions) directly to ``actor_fn`` without
    wiring the full DI scope hierarchy.
    """

    async def run_actor(
        actor_fn: Callable[..., object],
        payload: BaseModel | dict[str, object] | object,
        *,
        backend: InMemoryBackend,
        job_id: JobId | UUID | None = None,
        attempt: int = 1,
        cancel_event: asyncio.Event | None = None,
        actor: str = "test_actor",
        queue: str = "default",
        **deps: object,
    ) -> object:
        jid: JobId = JobId(job_id) if job_id is not None else new_job_id()
        evt = cancel_event or asyncio.Event()
        backend.register_cancel_event(jid, evt)

        # Coerce arbitrary payload shapes into BaseModel — the production
        # JobContext requires P: BaseModel. Real test payloads (BaseModel
        # instances) pass through; raw dicts and other shapes get wrapped
        # in the permissive PassthroughPayload (extra="allow").
        ctx_payload: BaseModel
        if isinstance(payload, BaseModel):
            ctx_payload = payload
        elif isinstance(payload, dict):
            ctx_payload = PassthroughPayload.model_validate(payload)
        else:
            ctx_payload = PassthroughPayload.model_validate({"value": payload})

        ctx: JobContext[BaseModel] = JobContext(
            job_id=jid,
            actor=actor,
            queue=queue,
            attempt=attempt,
            payload=ctx_payload,
            cancel_event=evt,
            worker_id=backend._worker_id,  # type: ignore[reportPrivateUsage]  # Why: fixture is an owned helper; _worker_id is private to InMemoryBackend but readable here for JobContext construction
            jobs=SubJobEnqueuer(
                loop_scope_resolved=None,
                worker_pool=None,
                backend=backend,
            ),
            log=bind_job_context(
                structlog.get_logger("taskq.testing.actor_runner"),
                job_id=jid,
                actor=actor,
                queue=queue,
                attempt=attempt,
                identity_key=None,
                trace_id="",
            ),
            deps=deps if deps else None,
        )
        result: object = actor_fn(payload, ctx)
        if isinstance(result, Awaitable):
            result = await cast(Awaitable[object], result)
        return result

    return run_actor

jobs_app async

jobs_app(
    pg_dsn: str, request: FixtureRequest
) -> AsyncIterator[JobsApp]

Yield a :class:JobsApp named tuple (deps, backend) against the database named by the consumer-provided pg_dsn fixture.

Access the fields as jobs_app.deps and jobs_app.backend instead of unpacking — the named-tuple interface is clearer and type-safe.

Per-test isolation: drops the schema CASCADE before each test (schema name is hashed from the test's own node id via :func:_schema_name_from_test, so distinct tests never share a schema), applies migrations, opens pools, constructs the backend. Teardown via AsyncExitStack unwind closes pools; the schema is dropped at the next invocation's setup for the same test — same pattern as pg_conn.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture
async def jobs_app(pg_dsn: str, request: pytest.FixtureRequest) -> AsyncIterator[JobsApp]:
    """Yield a :class:`JobsApp` named tuple ``(deps, backend)`` against the
    database named by the consumer-provided ``pg_dsn`` fixture.

    Access the fields as ``jobs_app.deps`` and ``jobs_app.backend`` instead
    of unpacking — the named-tuple interface is clearer and type-safe.

    Per-test isolation: drops the schema CASCADE before each test (schema
    name is hashed from the test's own node id via
    :func:`_schema_name_from_test`, so distinct tests never share a schema),
    applies migrations, opens pools, constructs the backend.  Teardown via
    ``AsyncExitStack`` unwind closes pools; the schema is dropped at the
    next invocation's setup for the same test — same pattern as ``pg_conn``.
    """
    stack, deps, backend = await _open_pg_backend(
        pg_dsn, schema_name=_schema_name_from_test(request)
    )
    try:
        yield JobsApp(deps=deps, backend=backend)
    finally:
        await stack.aclose()

backend_pair async

backend_pair(
    request: FixtureRequest,
) -> AsyncIterator[Backend]

Yield a single Backend instance per parametrize id.

• memory: same construction as memory_jobs.
• pg: requires a consumer-provided pg_dsn fixture (see the module docstring) plus migrations; reuses the same settings / migration sequence via :func:_open_pg_backend. Returns the PostgresBackend instance.

Tests using this fixture must be marked @pytest.mark.integration so the PG branch does not run in the unit tier. The guard below enforces this: the pg param is automatically skipped when the test lacks @pytest.mark.integration, preventing testcontainers from booting during unit-only runs.

pg_dsn is resolved lazily via request.getfixturevalue only inside the pg branch (after the integration-marker check) so the memory variant never triggers the pg_dsn fixture chain and stays container-free.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture(params=["memory", "pg"], ids=["memory", "pg"])
async def backend_pair(request: pytest.FixtureRequest) -> AsyncIterator[Backend]:
    """Yield a single ``Backend`` instance per parametrize id.

    - ``memory``: same construction as ``memory_jobs``.
    - ``pg``: requires a consumer-provided ``pg_dsn`` fixture (see the
      module docstring) plus migrations; reuses the same settings /
      migration sequence via :func:`_open_pg_backend`.  Returns the
      ``PostgresBackend`` instance.

    Tests using this fixture must be marked ``@pytest.mark.integration``
    so the PG branch does not run in the unit tier.  The guard below
    enforces this: the ``pg`` param is automatically skipped when the
    test lacks ``@pytest.mark.integration``, preventing testcontainers
    from booting during unit-only runs.

    ``pg_dsn`` is resolved lazily via ``request.getfixturevalue`` only
    inside the ``pg`` branch (after the integration-marker check) so
    the ``memory`` variant never triggers the ``pg_dsn`` fixture
    chain and stays container-free.
    """
    if request.param == "memory":
        clock = FakeClock(start=datetime(2025, 1, 1, tzinfo=UTC))
        backend: Backend = InMemoryBackend(clock=clock)
        for actor in DEFAULT_ACTORS:
            backend.register_actor_config(actor=actor)
        yield backend
    else:
        if not request.node.get_closest_marker("integration"):
            pytest.skip("PG backend requires @pytest.mark.integration")
        pg_dsn: str = request.getfixturevalue("pg_dsn")
        stack, _deps, pg_backend = await _open_pg_backend(
            pg_dsn, schema_name=_schema_name_from_test(request)
        )
        try:
            yield pg_backend
        finally:
            await stack.aclose()

redis_container

redis_container(
    tmp_path_factory: TempPathFactory,
) -> Iterator[_RedisContainerShim]

One shared Dragonfly (Redis-compatible) container per pytest invocation — every xdist worker of it, and no other invocation of any repo.

Backs onto :func:taskq.testing._shared_containers.shared_service_pair (file lock + per-invocation holder registry under the invocation's state dir — see :func:taskq.testing._shared_containers.invocation_state_dir): the first worker to take the lock boots the pair (--dbnum 1024 so every consumer — module or test function, across ALL workers of the invocation — gets its own logical DB; sharing would let one consumer's FLUSHDB wipe another's mid-run state); later workers reuse it; the last worker to finish removes it. The shim exposes get_container_host_ip / get_exposed_port so redis_url_for(redis_container, db) keeps working unchanged.

Source code in src/taskq/testing/fixtures.py

@pytest.fixture(scope="session")
def redis_container(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_RedisContainerShim]:
    """One shared Dragonfly (Redis-compatible) container per pytest invocation —
    every xdist worker of it, and no other invocation of any repo.

    Backs onto :func:`taskq.testing._shared_containers.shared_service_pair` (file
    lock + per-invocation holder registry under the invocation's state dir — see
    :func:`taskq.testing._shared_containers.invocation_state_dir`): the first
    worker to take the lock boots the pair (``--dbnum 1024`` so every consumer —
    module or test function, across ALL workers of the invocation — gets its own
    logical DB; sharing would let one consumer's FLUSHDB wipe another's mid-run
    state); later workers reuse it; the last worker to finish removes it. The
    shim exposes ``get_container_host_ip`` / ``get_exposed_port`` so
    ``redis_url_for(redis_container, db)`` keeps working unchanged.
    """
    state_dir = invocation_state_dir(tmp_path_factory)
    with shared_service_pair(state_dir) as services:
        yield _RedisContainerShim(
            host=services.redis_host, port=services.redis_port, state_dir=state_dir
        )

killable_redis_container

killable_redis_container() -> Iterator[RedisContainer]

Function-scoped disposable Dragonfly for chaos tests that stop/restart Redis.

Chaos tests must NEVER stop the shared container (redis_container) — every worker of the run shares it, and a slow restart leaks failures into unrelated tests. Each kill test gets its own container (~1s boot). The ownership labels (creator_labels) keep a crashed run's leftovers sweepable — Ryuk is disabled process-wide by the shared-container machinery (see :mod:taskq.testing._shared_containers), so labeling is what lets the next run's stale sweep remove these once their owner pids are dead.

Source code in src/taskq/testing/fixtures.py

@pytest.fixture
def killable_redis_container() -> Iterator[RedisContainer]:
    """Function-scoped disposable Dragonfly for chaos tests that stop/restart Redis.

    Chaos tests must NEVER stop the shared container (``redis_container``) —
    every worker of the run shares it, and a slow restart leaks failures into
    unrelated tests. Each kill test gets its own container (~1s boot). The
    ownership labels (``creator_labels``) keep a crashed run's leftovers
    sweepable — Ryuk is disabled process-wide by the shared-container machinery
    (see :mod:`taskq.testing._shared_containers`), so labeling is what lets the
    next run's stale sweep remove these once their owner pids are dead.
    """
    from testcontainers.community.redis import RedisContainer

    with (
        RedisContainer(image=DRAGONFLY_IMAGE)
        .with_command(DRAGONFLY_RESOURCE_FLAGS)
        .with_kwargs(labels=creator_labels())
    ) as rc:
        yield rc

redis_url_for

redis_url_for(
    container: RedisContainerLike, db: int = 0
) -> str

Build a redis://host:port/{db} URL for container.

Accepts anything with the RedisContainer host/port surface — the real testcontainers object or the shared-pair shim.

Source code in src/taskq/testing/fixtures.py

def redis_url_for(container: RedisContainerLike, db: int = 0) -> str:
    """Build a ``redis://host:port/{db}`` URL for *container*.

    Accepts anything with the ``RedisContainer`` host/port surface — the real
    testcontainers object or the shared-pair shim.
    """
    host = container.get_container_host_ip()
    port = container.get_exposed_port(6379)
    return f"redis://{host}:{port}/{db}"

redis_url

redis_url(redis_container: _RedisContainerShim) -> str

Per-test Redis URL with a UNIQUE logical DB — a guaranteed clean slate.

Each test function gets its own DB (never reused within the invocation), so no setup/teardown flushing is needed and no two tests can observe each other's keys. DB indices are drawn from the invocation's counter (see next_redis_logical_db) so consumers on different xdist workers can never collide inside the ONE shared Dragonfly.

Source code in src/taskq/testing/fixtures.py

@pytest.fixture
def redis_url(redis_container: _RedisContainerShim) -> str:
    """Per-test Redis URL with a UNIQUE logical DB — a guaranteed clean slate.

    Each test function gets its own DB (never reused within the invocation), so no
    setup/teardown flushing is needed and no two tests can observe each other's
    keys. DB indices are drawn from the invocation's counter (see
    ``next_redis_logical_db``) so consumers on different xdist workers can never
    collide inside the ONE shared Dragonfly.
    """
    return redis_url_for(redis_container, db=_next_redis_db(redis_container))

run_isolation_token

run_isolation_token() -> str

The token mixed into every hashed per-module/per-test database and schema name: the xdist worker id under xdist, the conftest-published per-invocation token in serial runs, master as a last-resort fallback (direct library use outside this repo's conftest).

The token's load-bearing job is WITHIN one invocation: the shared pair serves every xdist worker of it, and two workers hashing the same module or node id must not land the same database or schema on that one pair — each worker's DROP DATABASE ... WITH (FORCE) / DROP SCHEMA ... CASCADE would kill the other's live pools mid-test.

Invocation-uniqueness of the token (the conftest publishes the invocation-unique basetemp dir name, e.g. pytest-41) and the master fallback are defense-in-depth, not the isolation mechanism: the pair itself is per-invocation (:func:taskq.testing._shared_containers.invocation_state_dir), so two invocations' names can never land in one cluster. If that pair isolation ever regressed, per-invocation tokens would still keep the runs' names distinct on whatever they ended up sharing — mutual clobbering could not silently return.

Source code in src/taskq/testing/fixtures.py

def run_isolation_token() -> str:
    """The token mixed into every hashed per-module/per-test database and schema
    name: the xdist worker id under xdist, the conftest-published per-invocation
    token in serial runs, ``master`` as a last-resort fallback (direct library
    use outside this repo's conftest).

    The token's load-bearing job is WITHIN one invocation: the shared pair
    serves every xdist worker of it, and two workers hashing the same module
    or node id must not land the same database or schema on that one pair —
    each worker's ``DROP DATABASE ... WITH (FORCE)`` / ``DROP SCHEMA ... CASCADE``
    would kill the other's live pools mid-test.

    Invocation-uniqueness of the token (the conftest publishes the
    invocation-unique basetemp dir name, e.g. ``pytest-41``) and the
    ``master`` fallback are defense-in-depth, not the isolation mechanism:
    the pair itself is per-invocation
    (:func:`taskq.testing._shared_containers.invocation_state_dir`), so two
    invocations' names can never land in one cluster. If that pair isolation
    ever regressed, per-invocation tokens would still keep the runs' names
    distinct on whatever they ended up sharing — mutual clobbering could not
    silently return.
    """
    token = os.environ.get(RUN_TOKEN_ENV_VAR)
    if token:
        return token
    return os.environ.get("PYTEST_XDIST_WORKER", "master")

module_pg_schema async

module_pg_schema(
    request: FixtureRequest, pg_dsn: str
) -> AsyncIterator[ModulePgSchema]

Module-scoped PG schema: create once per test file, truncate per test.

Derives a unique schema name from the test module, applies migrations, seeds default actor_config rows. Drops the schema CASCADE on module teardown.

All tests in the module share the same schema; per-function isolation is provided by :func:clean_pg_conn and :func:clean_jobs_app which truncate every table before each test.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture(scope="module")
async def module_pg_schema(
    request: pytest.FixtureRequest,
    pg_dsn: str,
) -> AsyncIterator[ModulePgSchema]:
    """Module-scoped PG schema: create once per test file, truncate per test.

    Derives a unique schema name from the test module, applies migrations,
    seeds default actor_config rows.  Drops the schema CASCADE on module
    teardown.

    All tests in the module share the same schema; per-function isolation
    is provided by :func:`clean_pg_conn` and :func:`clean_jobs_app` which
    truncate every table before each test.
    """
    import asyncpg

    from taskq.migrate import apply_pending

    schema_name = _schema_name_from_module(request)
    conn = await asyncpg.connect(pg_dsn)
    try:
        await conn.execute(f'DROP SCHEMA IF EXISTS "{schema_name}" CASCADE')
        await apply_pending(conn, schema=schema_name)
        await seed_actors(conn, schema_name)
    finally:
        await conn.close()

    yield ModulePgSchema(schema_name=schema_name, pg_dsn=pg_dsn)

    conn = await asyncpg.connect(pg_dsn)
    try:
        await conn.execute(f'DROP SCHEMA IF EXISTS "{schema_name}" CASCADE')
    finally:
        await conn.close()

module_redis_url

module_redis_url(
    redis_container: _RedisContainerShim,
) -> Iterator[str]

Module-scoped Redis URL with a unique DB per test file.

Assigns a unique Redis DB id (1-1023, never reused, unique across ALL workers of the invocation — see _next_redis_db) and returns redis://host:port/{db}. FLUSHDB at setup (clean slate even after a crashed run) and again on teardown.

Source code in src/taskq/testing/fixtures.py

@pytest.fixture(scope="module")
def module_redis_url(
    redis_container: _RedisContainerShim,
) -> Iterator[str]:
    """Module-scoped Redis URL with a unique DB per test file.

    Assigns a unique Redis DB id (1-1023, never reused, unique across ALL
    workers of the invocation — see ``_next_redis_db``) and returns
    ``redis://host:port/{db}``. FLUSHDB at setup (clean slate even after a
    crashed run) and again on teardown.
    """
    import redis as redis_sync

    url = redis_url_for(redis_container, db=_next_redis_db(redis_container))
    with redis_sync.from_url(url, decode_responses=False) as client:
        client.flushdb()
    yield url

    # Teardown: FLUSHDB the module's DB via sync client (safe in any context).
    with redis_sync.from_url(url, decode_responses=False) as client:
        client.flushdb()

module_pg_pool async

module_pg_pool(
    module_pg_schema: ModulePgSchema,
) -> AsyncIterator[asyncpg.Pool]

Module-scoped asyncpg pool for the module's PG schema.

Pool is shared by all tests in a module. Per-test isolation is handled by clean_pg_conn/clean_jobs_app which truncate tables between tests.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture(scope="module")
async def module_pg_pool(module_pg_schema: ModulePgSchema) -> AsyncIterator[asyncpg.Pool]:
    """Module-scoped asyncpg pool for the module's PG schema.

    Pool is shared by all tests in a module. Per-test isolation is handled
    by clean_pg_conn/clean_jobs_app which truncate tables between tests.
    """
    import asyncpg

    pool = await asyncpg.create_pool(
        module_pg_schema.pg_dsn,
        min_size=1,
        max_size=4,
    )
    try:
        yield pool
    finally:
        await pool.close()

module_jobs_app async

module_jobs_app(
    module_pg_schema: ModulePgSchema,
) -> AsyncIterator[JobsApp]

Module-scoped JobsApp (WorkerDeps + PostgresBackend) on the module's PG schema.

Pools are opened once per test module. Per-test isolation is handled by clean_jobs_app which truncates tables between tests against the same schema. The backend instance is shared — callers should NOT cache or mutate backend state across test boundaries.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture(scope="module")
async def module_jobs_app(module_pg_schema: ModulePgSchema) -> AsyncIterator[JobsApp]:
    """Module-scoped JobsApp (WorkerDeps + PostgresBackend) on the module's PG schema.

    Pools are opened once per test module. Per-test isolation is handled by
    clean_jobs_app which truncates tables between tests against the same schema.
    The backend instance is shared — callers should NOT cache or mutate backend
    state across test boundaries.
    """
    stack, deps, backend = await _open_pg_backend_on_schema(
        module_pg_schema.pg_dsn,
        module_pg_schema.schema_name,
    )
    try:
        yield JobsApp(deps=deps, backend=backend)
    finally:
        await stack.aclose()

clean_pg_conn async

clean_pg_conn(
    module_pg_schema: ModulePgSchema,
) -> AsyncIterator[asyncpg.Connection]

Per-test clean asyncpg connection on the module's PG schema.

Truncates every dynamic table (FK-safe CASCADE) then re-seeds default actor_config rows, guaranteeing a blank slate for each test while keeping migration state intact.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture
async def clean_pg_conn(module_pg_schema: ModulePgSchema) -> AsyncIterator[asyncpg.Connection]:
    """Per-test clean asyncpg connection on the module's PG schema.

    Truncates every dynamic table (FK-safe CASCADE) then re-seeds default
    actor_config rows, guaranteeing a blank slate for each test while
    keeping migration state intact.
    """
    import asyncpg as _asyncpg

    conn = await _asyncpg.connect(module_pg_schema.pg_dsn)
    try:
        await reset_schema(conn, module_pg_schema.schema_name)
        yield conn
    finally:
        await conn.close()

worker_with_running_job async

worker_with_running_job(
    clean_pg_conn: Connection,
    module_pg_schema: ModulePgSchema,
) -> AsyncIterator[tuple[UUID, UUID, asyncpg.Connection]]

Per-test clean connection with a pre-created worker + running job.

Yields (worker_id, job_id, conn) where conn is the same asyncpg.Connection provided by :func:clean_pg_conn. The connection lifecycle is managed by clean_pg_conn — this fixture only inserts the worker and job rows.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture
async def worker_with_running_job(
    clean_pg_conn: asyncpg.Connection,
    module_pg_schema: ModulePgSchema,
) -> AsyncIterator[tuple[UUID, UUID, asyncpg.Connection]]:
    """Per-test clean connection with a pre-created worker + running job.

    Yields ``(worker_id, job_id, conn)`` where *conn* is the same
    ``asyncpg.Connection`` provided by :func:`clean_pg_conn`.  The
    connection lifecycle is managed by ``clean_pg_conn`` — this fixture
    only inserts the worker and job rows.
    """
    wid, jid = await create_workered_running_job(clean_pg_conn, module_pg_schema.schema_name)
    yield wid, jid, clean_pg_conn

clean_jobs_app async

clean_jobs_app(
    module_pg_schema: ModulePgSchema,
) -> AsyncIterator[JobsApp]

Per-test clean JobsApp (WorkerDeps + PostgresBackend) on the module's PG schema.

Truncates + re-seeds before each test, then opens WorkerDeps and constructs a PostgresBackend. Faster than the jobs_app fixture which drops/recreates the schema every test.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture
async def clean_jobs_app(module_pg_schema: ModulePgSchema) -> AsyncIterator[JobsApp]:
    """Per-test clean ``JobsApp`` (WorkerDeps + PostgresBackend) on the
    module's PG schema.

    Truncates + re-seeds before each test, then opens WorkerDeps and
    constructs a PostgresBackend.  Faster than the ``jobs_app`` fixture
    which drops/recreates the schema every test.
    """
    import asyncpg

    conn = await asyncpg.connect(module_pg_schema.pg_dsn)
    try:
        await reset_schema(conn, module_pg_schema.schema_name)
    finally:
        await conn.close()

    stack, deps, backend = await _open_pg_backend_on_schema(
        module_pg_schema.pg_dsn,
        module_pg_schema.schema_name,
    )
    try:
        yield JobsApp(deps=deps, backend=backend)
    finally:
        await stack.aclose()

clean_redis_url

clean_redis_url(module_redis_url: str) -> str

Per-test clean Redis URL.

FLUSHDB on the module's Redis DB before each test, guaranteeing no cross-test state within the module. Uses the synchronous redis.Redis client so the flush is reliable regardless of whether tests are sync or async.

Source code in src/taskq/testing/fixtures.py

@pytest.fixture
def clean_redis_url(module_redis_url: str) -> str:
    """Per-test clean Redis URL.

    FLUSHDB on the module's Redis DB before each test, guaranteeing
    no cross-test state within the module.  Uses the synchronous
    ``redis.Redis`` client so the flush is reliable regardless of
    whether tests are sync or async.
    """
    import redis as redis_sync

    client = redis_sync.from_url(module_redis_url, decode_responses=False)
    try:
        client.flushdb()
    finally:
        client.close()

    return module_redis_url

clean_redis_client async

clean_redis_client(
    clean_redis_url: str,
) -> AsyncIterator[object]

Per-test clean Redis async client against the module's DB.

Returns a fresh redis.asyncio.Redis client with decode_responses=False.

Source code in src/taskq/testing/fixtures.py

@pytest_asyncio.fixture
async def clean_redis_client(clean_redis_url: str) -> AsyncIterator[object]:
    """Per-test clean Redis async client against the module's DB.

    Returns a fresh ``redis.asyncio.Redis`` client with ``decode_responses=False``.
    """
    from redis.asyncio import from_url as redis_from_url

    client = redis_from_url(clean_redis_url, decode_responses=False)
    try:
        yield client
    finally:
        await client.aclose()

Health-socket helpers

health

Health-socket path helpers for test suites.

Solves the EADDRINUSE-under-xdist problem with the worker health server: WorkerSettings.health_socket_path defaults to the shared production path /tmp/taskq_health.sock, and worker._main starts a real HealthServer on it. Under pytest-xdist, two workers running _main in-process concurrently race on that one filesystem path — the loser gets EADDRINUSE (there is a TOCTOU window in create_unix_server's stale-file removal), or silently steals the socket from the live winner.

Recommended consumer pattern — mint a unique path per test 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")
)

Suites that build settings through the env cascade (WorkerSettings.load()) instead of a factory should additionally redirect HealthServer.start away from the shared default via an autouse monkeypatch — see tests/conftest.py::_isolate_health_server_socket in the TaskQ repo for the reference implementation (it is repo-specific and deliberately not published).

__all__ module-attribute

__all__ = ['unique_health_sock_path']

unique_health_sock_path

unique_health_sock_path(module: str) -> str

Return a unique unix-socket path for one test's health server.

/tmp/tq-<module>-<pid>-<token>.sock: the pid scopes across xdist workers, the random token across tests within a worker (stateless — no shared counter, so uniqueness survives any pytest import mode), and the module label identifies the owner when debugging stale files. The short /tmp/tq- prefix keeps paths well under the 104-char AF_UNIX sun_path limit on macOS.

Stale files are a resource-only leak: the per-run-unique token means a leftover path can never collide with a future run. HealthServer.stop unlinks on normal completion; suites wanting a crash backstop can sweep /tmp/tq-*-<pid>-*.sock scoped to their own pid (see TaskQ's tests/conftest.py::_sweep_health_sock_files).

:param module: label embedded in the path. Must not contain path separators — / would point the socket at a nonexistent directory and surface as a confusing ENOENT at bind time. :raises ValueError: if module contains / or \.

Source code in src/taskq/testing/health.py

def unique_health_sock_path(module: str) -> str:
    """Return a unique unix-socket path for one test's health server.

    ``/tmp/tq-<module>-<pid>-<token>.sock``: the pid scopes across xdist
    workers, the random token across tests within a worker (stateless — no
    shared counter, so uniqueness survives any pytest import mode), and the
    module label identifies the owner when debugging stale files. The short
    ``/tmp/tq-`` prefix keeps paths well under the 104-char AF_UNIX
    sun_path limit on macOS.

    Stale files are a resource-only leak: the per-run-unique token means a
    leftover path can never collide with a future run. ``HealthServer.stop``
    unlinks on normal completion; suites wanting a crash backstop can sweep
    ``/tmp/tq-*-<pid>-*.sock`` scoped to their own pid (see TaskQ's
    ``tests/conftest.py::_sweep_health_sock_files``).

    :param module: label embedded in the path. Must not contain path
        separators — ``/`` would point the socket at a nonexistent
        directory and surface as a confusing ENOENT at bind time.
    :raises ValueError: if *module* contains ``/`` or ``\\``.
    """
    if "/" in module or "\\" in module:
        raise ValueError(
            f"module label must not contain path separators, got {module!r} "
            "(it is embedded verbatim in the socket path)"
        )
    return f"/tmp/tq-{module}-{os.getpid()}-{new_base62()}.sock"  # noqa: S108  # Why: test-only socket files; /tmp is the shortest safe prefix.