Skip to content

Testing

InMemoryBackend, FakeClock, pytest fixtures, and test assertions.

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",
    "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 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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 = 2

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 
123
124
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 
126
127
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 
129
130
131
132
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 
134
135
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 
137
138
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,
) -> bool
Source code in src/taskq/testing/actor.py 
140
141
142
143
144
145
146
147
148
149
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,
) -> 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,
) -> bool
Source code in src/taskq/testing/actor.py 
151
152
153
154
155
156
157
158
159
160
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,
) -> bool:
    return await self.mark_succeeded(job_id, worker_id, result, progress_seq, progress_state)

mark_failed_or_retry async 

mark_failed_or_retry(
    job_id: UUID,
    worker_id: UUID,
    error_info: ErrorInfo,
    next_scheduled_at: datetime | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> JobRow
Source code in src/taskq/testing/actor.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
async def mark_failed_or_retry(
    self,
    job_id: UUID,
    worker_id: UUID,
    error_info: ErrorInfo,
    next_scheduled_at: datetime | 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,
            "next_scheduled_at": next_scheduled_at,
        }
    )
    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 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
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 
198
199
200
201
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 
203
204
205
206
207
208
209
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 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
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 
257
258
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 
260
261
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 
263
264
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 
266
267
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 
269
270
async def poll_cancel_flags(self, worker_id: UUID) -> list[CancelFlag]:
    return []

scheduled_to_pending async 

scheduled_to_pending(now: datetime) -> int
Source code in src/taskq/testing/actor.py 
272
273
async def scheduled_to_pending(self, now: datetime) -> int:
    return 0

deadline_sweep async 

deadline_sweep(now: datetime) -> int
Source code in src/taskq/testing/actor.py 
275
276
async def deadline_sweep(self, now: datetime) -> int:
    return 0

reclaim_expired_locks async 

reclaim_expired_locks(
    now: datetime,
    cancel_grace: timedelta,
    cleanup_grace: timedelta,
) -> int
Source code in src/taskq/testing/actor.py 
278
279
280
281
async def reclaim_expired_locks(
    self, now: datetime, 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 
283
284
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 
286
287
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 
289
290
async def count_pending_jobs(self, actors: list[str]) -> dict[str, int]:
    return {}

enqueue_batch async 

enqueue_batch(
    args_list: list[EnqueueArgs],
    *,
    connection: object = None,
) -> list[JobRow]
Source code in src/taskq/testing/actor.py 
292
293
294
295
296
297
298
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 
300
301
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 
303
304
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 
306
307
308
309
310
311
312
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 
314
315
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 
317
318
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 
29
30
def __init__(self, start: datetime) -> None:
    self._now = start

now 

now() -> datetime
Source code in src/taskq/testing/clock.py 
32
33
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 
35
36
37
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 
39
40
41
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 
43
44
45
46
47
48
49
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,
)

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 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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,
) -> None:
    self._clock = clock
    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[IdempotencyKey, 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] = {}

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 
211
212
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,
    payload_type: type[BaseModel] | None = None,
) -> None
Source code in src/taskq/testing/in_memory.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
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,
    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,
        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 
242
243
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 
247
248
async def get_events(self, job_id: JobId) -> list[EventRow]:
    return await _get_events(self, job_id)

enqueue async 

enqueue(args: EnqueueArgs) -> JobRow
Source code in src/taskq/testing/in_memory.py 
330
331
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 
333
334
335
336
337
338
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 
340
341
342
343
344
345
346
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 
348
349
350
351
352
353
354
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,
    queue: str = "default",
    metadata: dict[str, object] | None = None,
) -> None
Source code in src/taskq/testing/in_memory.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def register_actor_config(
    self,
    *,
    actor: str,
    max_concurrent: int | None = None,
    queue: str = "default",
    metadata: dict[str, object] | None = None,
) -> None:
    _register_actor_config(
        self,
        actor=actor,
        max_concurrent=max_concurrent,
        queue=queue,
        metadata=metadata,
    )

register_actor_configs 

register_actor_configs(
    configs: Iterable[ActorConfig],
) -> None
Source code in src/taskq/testing/in_memory.py 
374
375
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 
379
380
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 
382
383
384
385
386
387
388
389
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 
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
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 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
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,
) -> bool
Source code in src/taskq/testing/in_memory.py 
427
428
429
430
431
432
433
434
435
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,
) -> bool:
    return await _mark_succeeded(self, job_id, worker_id, result, progress_seq, progress_state)

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,
) -> bool
Source code in src/taskq/testing/in_memory.py 
437
438
439
440
441
442
443
444
445
446
447
448
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,
) -> bool:
    return await _mark_succeeded_with_conn(
        self, conn, job_id, worker_id, result, progress_seq, progress_state
    )

mark_failed_or_retry async 

mark_failed_or_retry(
    job_id: JobId,
    worker_id: UUID,
    error_info: ErrorInfo,
    next_scheduled_at: datetime | None,
    progress_seq: int = 0,
    progress_state: dict[str, object] | None = None,
) -> JobRow
Source code in src/taskq/testing/in_memory.py 
450
451
452
453
454
455
456
457
458
459
460
461
async def mark_failed_or_retry(
    self,
    job_id: JobId,
    worker_id: UUID,
    error_info: ErrorInfo,
    next_scheduled_at: datetime | 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, next_scheduled_at, 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 
463
464
465
466
467
468
469
470
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 
472
473
474
475
476
477
478
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 
480
481
482
483
484
485
486
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 
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
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 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
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 
532
533
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 
535
536
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 
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
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 
595
596
597
598
599
600
601
602
603
604
605
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
    ]

retry_job async 

retry_job(job_id: JobId) -> bool
Source code in src/taskq/testing/in_memory.py 
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
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(now: datetime) -> int
Source code in src/taskq/testing/in_memory.py 
633
634
async def scheduled_to_pending(self, now: datetime) -> int:
    return await _scheduled_to_pending(self, now)

deadline_sweep async 

deadline_sweep(now: datetime) -> int
Source code in src/taskq/testing/in_memory.py 
636
637
async def deadline_sweep(self, now: datetime) -> int:
    return await _deadline_sweep(self, now)

reclaim_expired_locks async 

reclaim_expired_locks(
    now: datetime,
    cancel_grace: timedelta,
    cleanup_grace: timedelta,
) -> int
Source code in src/taskq/testing/in_memory.py 
639
640
641
642
643
644
645
async def reclaim_expired_locks(
    self,
    now: datetime,
    cancel_grace: timedelta,
    cleanup_grace: timedelta,
) -> int:
    return await _reclaim_expired_locks(self, now, 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 
649
650
651
652
653
654
655
656
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 
658
659
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 
661
662
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 
666
667
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 
669
670
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 
672
673
async def count_pending_jobs(self, actors: list[str]) -> dict[str, int]:
    return await _count_pending_jobs(self, actors)

subscribe_wake 

subscribe_wake() -> AsyncContextManager[asyncio.Event]
Source code in src/taskq/testing/in_memory.py 
677
678
679
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 
681
682
683
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 
687
688
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 
692
693
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 
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
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 
723
724
725
726
727
728
729
730
731
732
733
734
735
736
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 
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
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 
772
773
async def delete_schedule(self, schedule_id: UUID) -> None:
    self._schedules.pop(schedule_id, None)

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 
61
62
63
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 
11
12
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 
14
15
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 
321
322
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 
49
50
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 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
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 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
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 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
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 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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 
160
161
162
163
164
165
166
167
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 
196
197
198
199
200
201
202
203
204
205
206
207
208
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 
88
89
90
91
92
93
94
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 
302
303
304
305
306
307
308
309
310
311
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 
261
262
263
264
265
266
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 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
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 
294
295
296
297
298
299
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

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 
138
139
140
141
142
143
144
145
146
147
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,
    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 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def make_enqueue_args(
    *,
    actor: str = "test_actor",
    queue: str = "default",
    payload: dict[str, object] | None = None,
    idempotency_key: str | None = None,
    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,
        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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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,
        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 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
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 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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, now(),
            $8, $9, now(), now(),
            $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 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
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 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
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 
124
125
126
127
128
129
130
131
132
133
134
135
136
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 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
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 
 93
 94
 95
 96
 97
 98
 99
100
101
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.

Source code in src/taskq/testing/settings.py 
35
36
37
def make_integration_settings(pg_dsn: str, **overrides: str) -> WorkerSettings:
    """Construct WorkerSettings with fast intervals for integration tests."""
    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.

Source code in src/taskq/testing/settings.py 
40
41
42
def make_integration_settings_dict(pg_dsn: str, **overrides: str) -> dict[str, str]:
    """Return the raw dict passed to WorkerSettings.load_from_dict."""
    return _build_dict(pg_dsn, **overrides)