taskq.testing is the test-support namespace: deterministic fakes, pytest
fixtures, behavioral assertions, settings factories, and chaos tools. Every
symbol lives under taskq.testing, never in taskq.backend, so production
code never pulls in test doubles.
fromtaskq.testingimportFakeClock,InMemoryBackend
The Testing guide walks these tools in context; this
page is the namespace index: every public name, the module that defines it,
and the rendered reference for the core surfaces.
FakeClock's canonical home is taskq.clock — a production module that
imports only datetime. That placement (issue #625) is what lets
import taskq re-export FakeClock without pulling in the test-double
backend under taskq.testing. taskq.testing.clock is a backcompat shim
re-exporting the same class object, and the taskq package and the
taskq.testing namespace re-export it in turn. All four paths resolve to
one class object (is-identical, not copies):
Prefer from taskq.clock import FakeClock (or the shorter
from taskq import FakeClock) in new code; taskq.testing.clock exists for
backcompat and keeps working.
All 38 render under the package directive below. Three more surfaces are
also not in the package __all__ and have their own sections:
the pytest fixtures (taskq.testing.fixtures — keeps pytest/asyncpg off
the package import path), the OTel test utilities (taskq.testing.otel —
requires the [otel] extra), and the asyncpg chaos tools
(taskq.testing.asyncpg_chaos). The defining modules also carry public
module-level helpers that are not re-exported through the package; those are
indexed below.
Module-level publics outside the package contract¶
These names are public in their defining modules (most in the module's own
__all__) but deliberately not re-exported from taskq.testing — import
them from the defining module:
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).
Minimal backend recording method calls for assertions.
The terminal writes model the real backends' fencing, not just the
happy path: Postgres and the in-memory twin both fence on
(status='running', locked_by_worker, attempt), so once one
terminal write has moved a job row out of running every later
terminal write for that job matches nothing and reports False.
The double keeps the terminal-state half of that contract, the first
mark_succeeded/mark_cancelled for a job id lands, any
subsequent one returns False, so a test exercising a
cancel/success race cannot pass vacuously against a backend that let
both writes land. The worker/attempt conjuncts are subsumed by
first-writer-wins: the landing write IS the holder's, and on the real
backends a repeated write fails the status conjunct even from the
same worker and attempt.
def__init__(self,*,mark_snoozed_return:Literal["scheduled","failed","noop"]="scheduled",mark_retry_after_return:Literal["scheduled","failed:DeadlineExceeded","failed:MaxAttemptsExceeded","noop"]="scheduled",mark_interrupted_return:Literal["pending","scheduled","failed:DeadlineExceeded","noop"]="pending",)->None:self.mark_succeeded_calls:list[tuple[UUID,UUID,dict[str,object]|None,bytes|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_interrupted_calls:list[dict[str,object]]=[]self.mark_failed_or_retry_calls:list[dict[str,object]]=[]# The fence's terminal-state half: job ids a terminal write# already moved out of 'running', mapped to the outcome that won.self._terminal_outcomes:dict[UUID,str]={}self._mark_snoozed_return:Literal["scheduled","failed","noop"]=mark_snoozed_returnself._mark_retry_after_return:Literal["scheduled","failed:DeadlineExceeded","failed:MaxAttemptsExceeded","noop"]=mark_retry_after_returnself._mark_interrupted_return:Literal["pending","scheduled","failed:DeadlineExceeded","noop"]=mark_interrupted_return
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.
defmonotonic(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()
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.
def__init__(self,clock:Clock,cancellation_grace_period:timedelta=timedelta(seconds=30),cleanup_grace_period:timedelta=timedelta(seconds=30),*,actor_configs:Iterable[ActorConfig]|None=None,result_max_bytes:int=MAX_RESULT_BYTES,max_retry_backoff:timedelta=DEFAULT_MAX_RETRY_BACKOFF,)->None:self._clock=clockself._result_max_bytes=result_max_bytesself._cancellation_grace=cancellation_grace_periodself._cleanup_grace=cleanup_grace_period# The WorkerSettings.max_retry_backoff knob, carried the same way# the grace knobs above are: the reclaim sweep twin clamps the# reschedule delay at min(row cap, this ceiling), mirroring the# PG sweep's bound parameter.self._max_retry_backoff=max_retry_backoff# The claim cursor's knob, carried like the knobs above: the# dispatch twin reads it per round (0 disables the cursor - the# naive shape). The default mirrors WorkerSettings'# claim_cursor_reset_seconds' OPT-IN default (0 = disabled, the# A/B-measured posture), NOT the mechanism's 60s cadence constant# (backend/_claim_cursor.py CLAIM_CURSOR_RESET_SECONDS) - the# posture and the cadence are deliberately different numbers.self._claim_cursor_reset_seconds:float=0.0self._worker_id:UUID=new_uuid()self._jobs:_JobStore=_JobStore(clock)# The claim cursor: this backend's per-queue high-water mark of# successfully-claimed ids, the twin of PostgresBackend's (see# backend/_claim_cursor.py). Clocked by the backend's own clock so# tests drive the jitter reset like every other clocked seam.self._claim_cursor=ClaimCursor(clock=clock.monotonic)self._attempts:dict[JobId,list[AttemptRow]]={}self._events:list[EventRow]=[]# The event-prune watermark twin (PG: job_events_prune_state,# migration 01.00.20_02): the highest event id any deleter has# committed a delete below-or-at. Advanced by the prune# simulation (testing/_runner.archive_terminal_jobs, the mirror# of the PG cascade) so _poll_reclaim_events' gap gate sees the# same signal both backends raise.self._events_pruned_through:int=0self._idempotency_index:dict[tuple[str,str],JobId]={}self._event_seq:int=0self._cancel_observed_at:dict[JobId,datetime]={}self._cancel_events:dict[JobId,asyncio.Event]={}# The (job, task) of the attempt run_until_drained is currently# executing, keyed by job id so only the abandon of the job# actually in flight can act on it. tick_cancel_polling's# both-graces arm cancels it, the runner's mirror of production# phase 2's active.task.cancel(), so a non-cooperative attempt# cannot park the drain forever beside a row that already says# abandoned. None whenever no attempt is executing.self._inflight_attempt:tuple[JobId,asyncio.Task[object]]|None=Noneself._wake_subscribers:set[asyncio.Event]=set()self._wake_queues:dict[asyncio.Event,frozenset[str]|None]={}self._cancel_wake_subscribers:set[asyncio.Event]=set()# Leadership wake registry: parity with the PG backend's# subscribe_leader_wake keeps the differential tests honest about# the subscription contract. The in-memory backend never emits the# broadcast (there is no LISTEN wire); the event is set only by a# test that drives the wake by hand.self._leader_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]={}# Mirror of actor_config.last_claimed_at (the cross-round actor# rotation stamp the dispatch cut orders by):# PG stamps claiming actors with the statement's wall time, which# is strictly increasing across rounds; a frozen FakeClock cannot# supply that, so the twin stamps with a per-backend tick# incremented once per CLAIMING dispatch round. Same-round winners# share one tick and fall through to scheduled_at/id, exactly the# tie shape PG's single statement_timestamp() produces, and an# actor never claimed has no entry, mirroring NULL (sorts first).self._actor_claim_ticks:dict[str,int]={}self._claim_tick:int=0ifactor_configsisnotNone:forcfginactor_configs:self._actor_configs_meta[cfg.actor]=cfgself._slot_table:_SlotTable|None=Noneself._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]={}
The twin's event-prune watermark (PG:
job_events_prune_state, migration 01.00.20_02). Advanced by
the prune simulation's cascade arm
(testing._runner.archive_terminal_jobs); no sweep on this
backend deletes events any other way, so no other writer exists.
watch_reclaims' gap gate reads this through the same
getattr-probe it applies to the PG backend.
asyncdefevent_prune_watermark(self)->int:"""The twin's event-prune watermark (PG: ``job_events_prune_state``, migration 01.00.20_02). Advanced by the prune simulation's cascade arm (``testing._runner.archive_terminal_jobs``); no sweep on this backend deletes events any other way, so no other writer exists. watch_reclaims' gap gate reads this through the same getattr-probe it applies to the PG backend. """returnself._events_pruned_through
asyncdefextend_reservation_leases(self,worker_id:UUID,lock_lease:timedelta,*,disowned:Collection[UUID]=(),)->int:now=self._clock.now()excluded=set(disowned)count=0forjob_id,rowinlist(self._jobs.items()):if(row.status=="running"androw.locked_by_worker==worker_idandjob_idnotinexcludedandself._slot_tableisnotNone):# The count is SLOT rows renewed, the twin of PG's# UPDATE_RESERVATION_LEASES_SQL_TEMPLATE rowcount: dispatch# never creates a reservation_slots row (the reservation# registry's acquisition does), so a worker holding plain# running jobs extends nothing and answers 0. The mirror's# slot-table-less default models exactly that deployment -# a former `count += 1` fallback here counted every held# running job instead, reporting slot renewals production# never makes (pinned in# tests/test_heartbeat_seam_parity.py).count+=self._slot_table.extend_leases_for_job(job_id,now,lock_lease)returncount
asyncdefwrite_cancel_request(self,job_id:JobId,reason:str|None,)->bool:row=self._jobs.get(job_id)ifrowisNone:returnFalseifrow.status=="running"androw.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)foreventinself._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,)returnTrueifrow.statusin("pending","scheduled"):now=self._clock.now()prev_status=row.statusself._jobs[job_id]=replace(row,status="cancelled",finished_at=now,# Twin of cancel_pending_scheduled: the before-start origin# marker goes on the ROW only, the state_change event# detail keeps its {from_state, to_state} shape on both# backends (the differential suite pins it exactly).error_class=CANCEL_ORIGIN_PENDING,)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,)returnTruereturnFalse
Grouped pending+scheduled counts per queue, the admission count the
max_pending cap governs (the LIB-2 backpressure read's DEPTH half).
The in-memory twin of sql.count_pending_jobs_by_queue: same
statuses (pending + scheduled, exactly what
enqueue_max_pending_count counts), grouped per queue, absent
queues read 0 client-side.
asyncdefcount_pending_jobs_by_queue(self,queues:list[str])->dict[str,int]:"""Grouped pending+scheduled counts per queue, the admission count the max_pending cap governs (the LIB-2 backpressure read's DEPTH half). The in-memory twin of ``sql.count_pending_jobs_by_queue``: same statuses (pending + scheduled, exactly what ``enqueue_max_pending_count`` counts), grouped per queue, absent queues read 0 client-side. """ifnotqueues:return{}queue_set=set(queues)counts:dict[str,int]={}forrinself._jobs.values():ifr.queueinqueue_setandr.statusin("pending","scheduled"):counts[r.queue]=counts.get(r.queue,0)+1returncounts
The parent's pending children, exact and grouped by queue (the LIB-2
backpressure read's FAN-OUT half), the in-memory twin of
sql.count_pending_children_by_queue.
parent_id is a plain column, no FK: a purged or never-existed
parent is the same defined-empty result.
asyncdefcount_pending_children_by_queue(self,parent_id:JobId)->dict[str,int]:"""The parent's pending children, exact and grouped by queue (the LIB-2 backpressure read's FAN-OUT half), the in-memory twin of ``sql.count_pending_children_by_queue``. ``parent_id`` is a plain column, no FK: a purged or never-existed parent is the same defined-empty result. """counts:dict[str,int]={}forrinself._jobs.values():ifr.parent_id==parent_idandr.statusin("pending","scheduled"):counts[r.queue]=counts.get(r.queue,0)+1returncounts
Queue-scoped wake subscription, the PG backend's contract mirrored.
queues=None wakes on every enqueue; a queue set wakes only when
the enqueued row's queue is served by this subscriber. Parity with
the PG backend's payload filtering keeps the differential tests
honest about the wake contract.
defsubscribe_wake(self,queues:Iterable[str]|None=None)->AsyncContextManager[asyncio.Event]:"""Queue-scoped wake subscription, the PG backend's contract mirrored. ``queues=None`` wakes on every enqueue; a queue set wakes only when the enqueued row's queue is served by this subscriber. Parity with the PG backend's payload filtering keeps the differential tests honest about the wake contract. """event=asyncio.Event()# A bare str satisfies Iterable[str] and would char-split into a# nonsense set ('reports' -> {'r','e',...}), silently falling the# subscriber back to poll cadence: single strings are the whole set.queue_set=(frozenset({queues})ifisinstance(queues,str)else(frozenset(queues)ifqueueselseNone))return_SubscriberContext(event,self._wake_subscribers,queue_registry=self._wake_queues,queues=queue_set,)
Parity with the PG backend: subscribe to the leader's resign
broadcast. Nothing in-memory emits it (no LISTEN wire), so a
subscriber keeps the heartbeat-tick cadence, exactly the poll
fallback the hint semantics promise.
defsubscribe_leader_wake(self)->AsyncContextManager[asyncio.Event]:"""Parity with the PG backend: subscribe to the leader's resign broadcast. Nothing in-memory emits it (no LISTEN wire), so a subscriber keeps the heartbeat-tick cadence, exactly the poll fallback the hint semantics promise."""event=asyncio.Event()return_SubscriberContext(event,self._leader_wake_subscribers)
asyncdefcreate_schedule(self,args:ScheduleCreateArgs)->ScheduleRecord:forrecinself._schedules.values():ifrec.actor==args.actorandrec.name==args.name:raiseValueError(f"schedule for actor {args.actor!r} name {args.name!r} already exists")sid=new_uuid()# Same create-time provenance rule as backend/_schedules.py: an# operator-owned schedule created disabled carries operator intent# from birth; a code-owned one starts unowned (NULL).disabled_by:str|None=Noneifnotargs.enabledandargs.owner=="operator":disabled_by="operator"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,disabled_by=disabled_by,last_fired_at=None,last_fire_error=None,consecutive_failures=0,next_fire_at=args.next_fire_at,# PG serialises metadata into jsonb at INSERT time and reads it# back through loads, store the round-trip of the same# serialization, so a caller-held dict can never reach storage# by reference and values whose orjson encoding differs from# the Python object read back exactly as PG reads them.metadata=loads(dumps_jsonb_str(args.metadata)),)self._schedules[sid]=recordreturn_schedule_read_copy(record)
asyncdefupdate_schedule(self,schedule_id:UUID,args:ScheduleUpdateArgs,)->ScheduleRecord:rec=self._schedules.get(schedule_id)ifrecisNone:raiseKeyError(f"schedule {schedule_id} not found")updates:dict[str,object]={}ifargs.cron_exprisnotNone:updates["cron_expr"]=args.cron_exprifargs.next_fire_atisnotNone:updates["next_fire_at"]=args.next_fire_atifargs.enabledisnotNone:updates["enabled"]=args.enabledifargs.enabled:updates["consecutive_failures"]=0updates["last_fire_error"]=Noneupdates["disabled_by"]=Noneelse:# Mirror of backend/_schedules.update_schedule: a disable# through the backend API is operator-driven, the cron# loop's auto-disable stamps 'auto' in its own statement.updates["disabled_by"]="operator"ifargs.payload_factoryisnotNone:updates["payload_factory"]=args.payload_factoryelifargs.clear_payload_factory:updates["payload_factory"]=Noneifargs.metadataisnotNone:# Same storage contract as create_schedule: the round-trip of# the same serialization PG binds, no caller reference in# storage, PG's jsonb read-back values.updates["metadata"]=loads(dumps_jsonb_str(args.metadata))ifargs.consecutive_failuresisnotNone:updates["consecutive_failures"]=args.consecutive_failuresifargs.last_fire_errorisnotNone:updates["last_fire_error"]=args.last_fire_errorupdated=rec.model_copy(update=updates)self._schedules[schedule_id]=updatedreturn_schedule_read_copy(updated)
asyncdefcomplete_batch(self,batch_id:UUID,*,connection:object=None,)->bool:# The in-memory backend is single-threaded and its completion# decision reads the live member set at the write, so it never# gates out mid-drain and never owes a reissue._complete_batch(self,batch_id,connection)returnFalse
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).
Raise :class:asyncio.CancelledError when cancellation has
been requested, the production contract, so actors using the
raising style are exercisable through the harness.
defcheck_cancelled(self)->None:"""Raise :class:`asyncio.CancelledError` when cancellation has been requested, the production contract, so actors using the raising style are exercisable through the harness."""ifself.cancel_event.is_set():raiseasyncio.CancelledError
Record a progress report on the context (see the
progress_reports field). Signature and seq monotonicity
mirror the production :meth:taskq.context.JobContext.progress;
the harness never blocks on the network because it never
publishes.
asyncdefprogress(self,*,step:int|None=None,percent:float|None=None,detail:str|None=None,data:dict[str,object]|None=None,)->None:"""Record a progress report on the context (see the ``progress_reports`` field). Signature and ``seq`` monotonicity mirror the production :meth:`taskq.context.JobContext.progress`; the harness never blocks on the network because it never publishes."""self.progress_reports.append({"seq":len(self.progress_reports)+1,"step":step,"percent":percent,"detail":detail,"data":data,})
defassert_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*."""ifindex<0orindex>=len(attempts):raiseAssertionError(f"Attempt index {index} out of range (len={len(attempts)})")row=attempts[index]ifoutcomeisnotNone:actual=_get(row,"outcome")ifactual!=outcome:raiseAssertionError(f"attempts[{index}]: expected outcome {outcome!r}, got {actual!r}")iferror_classisnotNone:actual=_get(row,"error_class")ifactual!=error_class:raiseAssertionError(f"attempts[{index}]: expected error_class {error_class!r}, got {actual!r}")ifattempt_numisnotNone:actual=_get(row,"attempt")ifactual!=attempt_num:raiseAssertionError(f"attempts[{index}]: expected attempt {attempt_num}, got {actual}")returnrow
defassert_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."""forevinevents:if_get(ev,"kind")!=kind:continueiffrom_stateisnotNoneorto_stateisnotNone:detail=parse_detail(_get(ev,"detail"))iffrom_stateisnotNoneanddetail.get("from_state")!=from_state:continueifto_stateisnotNoneanddetail.get("to_state")!=to_state:continuereturnevavailable=[(i,_get(e,"kind"))fori,einenumerate(events)]msg=f"No event with kind={kind!r}"iffrom_stateisnotNoneorto_stateisnotNone:msg+=f" (from_state={from_state!r}, to_state={to_state!r})"msg+=f"; available events: {available}"raiseAssertionError(msg)
defassert_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)ifnotevents:raiseAssertionError(f"No OTel event {event_name!r} on span {span_name!r}")iffrom_stateisnotNoneorto_stateisnotNone:forevinevents:attrs=ev.attributesifattrsisNone:continueiffrom_stateisnotNoneandattrs.get("from_state")!=from_state:continueifto_stateisnotNoneandattrs.get("to_state")!=to_state:continuereturnevraiseAssertionError(f"No OTel event {event_name!r} on span {span_name!r} "f"with from_state={from_state!r}, to_state={to_state!r}")returnevents[0]
defassert_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)ifspanisNone:names=[s.nameforsinexporter.spans]raiseAssertionError(f"No span named {name!r}; available: {names}")ifkindisnotNoneandspan.kind!=kind:raiseAssertionError(f"Span {name!r}: expected kind={kind!r}, got {span.kind!r}")ifstatusisnotNoneandspan.status.status_code!=status:raiseAssertionError(f"Span {name!r}: expected status_code={status!r}, got {span.status.status_code!r}")returnspan
defassert_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. """assertrowisnotNone,"Expected a row but got None"actual_status=_get(row,"status")ifactual_status!=status:raiseAssertionError(f"Expected status {status!r}, got {actual_status!r}")iferror_classisnotNone:actual_ec=_get(row,"error_class")ifactual_ec!=error_class:raiseAssertionError(f"Expected error_class {error_class!r}, got {actual_ec!r}")ifattemptisnotNone:actual_attempt=_get(row,"attempt")ifactual_attempt!=attempt:raiseAssertionError(f"Expected attempt {attempt}, got {actual_attempt}")iffinishedisnotNone:finished_at=_get(row,"finished_at")iffinishedandfinished_atisNone:raiseAssertionError("Expected finished_at to be set, but it is None")ifnotfinishedandfinished_atisnotNone:raiseAssertionError(f"Expected finished_at to be None, got {finished_at!r}")returnrow
defassert_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."""returnassert_job_status(row,status,error_class=error_class,finished=True)
defparse_detail(detail:object)->dict[str,object]:"""Normalize a detail value (dict, JSON string, or other) to a dict."""ifisinstance(detail,dict):returndetail# 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].ifisinstance(detail,str):returndict(loads(detail))return{}
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.
asyncdefpg_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=awaitconn.fetchval("SELECT clock_timestamp()")returnvalue
Wait for an asyncio.Event with test-failure semantics on timeout.
description names the awaited thing in the failure message so a
timeout says what never happened ("<description> not set within
Ns"); without it the generic "Event not set" is used.
asyncdefwait_for(event:asyncio.Event,timeout:float=2.0,# noqa: ASYNC109 # Why: a `timeout` parameter is the point, callers pass per-call deadlines, not an enclosing asyncio.timeout scope; the repo's established wait-helper shape.*,description:str|None=None,)->None:"""Wait for an asyncio.Event with test-failure semantics on timeout. ``description`` names the awaited thing in the failure message so a timeout says what never happened (``"<description> not set within Ns"``); without it the generic ``"Event not set"`` is used. """try:awaitasyncio.wait_for(event.wait(),timeout=timeout)exceptTimeoutError:label=descriptionor"Event"raiseAssertionError(f"{label} not set within {timeout}s")fromNone
asyncdefwait_for_job_status(backend:_AssertBackend,job_id:JobId,status:str,*,timeout:float=2.0,# noqa: ASYNC109poll_interval:float=0.05,)->JobRow:"""Poll backend.get until the job reaches the expected status."""loop=asyncio.get_running_loop()deadline=loop.time()+timeoutwhileTrue:row=awaitbackend.get(job_id)ifrowisnotNoneand_get(row,"status")==status:returnrowremaining=deadline-loop.time()ifremaining<=0:actual=_get(row,"status")ifrowisnotNoneelseNoneraiseAssertionError(f"Job {job_id} did not reach status {status!r} within {timeout}s "f"(actual: {actual!r})")awaitasyncio.sleep(min(poll_interval,remaining))
asyncdefwait_for_leader(deps:_LeaderDeps,timeout:float=5.0)->None:# noqa: ASYNC109"""Wait for the leader event on WorkerDeps with test-failure semantics."""try:awaitasyncio.wait_for(deps.is_leader.wait(),timeout=timeout)exceptTimeoutError:raiseAssertionError(f"Leader event not set within {timeout}s")fromNone
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 \.
defunique_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"/"inmoduleor"\\"inmodule:raiseValueError(f"module label must not contain path separators, got {module!r} ""(it is embedded verbatim in the socket path)")returnf"/tmp/tq-{module}-{os.getpid()}-{new_base62()}.sock"# noqa: S108 # Why: test-only socket files; /tmp is the shortest safe prefix.
deferror_info(error_class:str="ValueError",error_message:str="boom",)->ErrorInfo:"""Shorthand for ErrorInfo with error_traceback=None."""returnErrorInfo(error_class=error_class,error_message=error_message,error_traceback=None,)
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.
retry_jitter defaults to the EnqueueArgs field's own default;
pass 0.0 to make a test's reclaim/retry timing assertions exact ,
jitter is fleet randomness, and the suite doubles randomness at the
boundary rather than sampling it.
defmake_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,retry_jitter:float|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. ``retry_jitter`` defaults to the ``EnqueueArgs`` field's own default; pass ``0.0`` to make a test's reclaim/retry timing assertions exact , jitter is fleet randomness, and the suite doubles randomness at the boundary rather than sampling it. """base=EnqueueArgs(id=new_job_id(),actor=actor,queue=queue,payload=payloador{"value":1},max_attempts=max_attempts,retry_kind=retry_kind,scheduled_at=scheduled_ator(datetime.now(UTC)-timedelta(seconds=1)),priority=priority,schedule_to_close=schedule_to_close,idempotency_key=IdempotencyKey(idempotency_key)ifidempotency_keyisnotNoneelseNone,idempotency_scope=idempotency_scope,identity_key=IdentityKey(identity_key)ifidentity_keyisnotNoneelseNone,metadata=metadataor{},tags=tagsiftagsisnotNoneelse(),)ifretry_jitterisNone:returnbasereturnreplace(base,retry_jitter=retry_jitter)
defmake_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:CancelPhaseifcancel_phaseisNone:phase=CancelPhase.NONEelifisinstance(cancel_phase,CancelPhase):phase=cancel_phaseelse:phase=CancelPhase(cancel_phase)locked_by=_WORKER_IDifstatus=="running"elseNonereturnJobRow(id=new_job_id(),actor=actor,queue=queue,identity_key=identity_key,fairness_key=None,payload=payloador{},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=_NOWifstatus=="running"elseNone,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=(),)
Seed one job row directly. scheduled_at defaults to the
application clock's now(), a stamp a claim CTE comparing against
the database's statement_timestamp() reads as not-yet-due whenever
the database clock lags the application clock (Docker VM pause and NTP
drift both cause it). Seed a past margin or an explicit scheduled_at
whenever the test then asserts the row is claimable.
asyncdefcreate_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:"""Seed one job row directly. ``scheduled_at`` defaults to the application clock's ``now()``, a stamp a claim CTE comparing against the database's ``statement_timestamp()`` reads as not-yet-due whenever the database clock lags the application clock (Docker VM pause and NTP drift both cause it). Seed a past margin or an explicit ``scheduled_at`` whenever the test then asserts the row is claimable. """ifnot_IDENT_RE.match(schema):raiseValueError(f"invalid schema name {schema!r}")job_id=job_idornew_uuid()stc=schedule_to_closeor(datetime.now(UTC)+timedelta(seconds=60))sa=scheduled_atordatetime.now(UTC)awaitconn.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: S608job_id,"test_actor","default",'{"key": "value"}',3,"transient",status,sa,stc,)returnjob_id
asyncdefcreate_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:ifnot_IDENT_RE.match(schema):raiseValueError(f"invalid schema name {schema!r}")job_id=job_idornew_uuid()expires_at=lock_expires_ator(datetime.now(UTC)+timedelta(seconds=60))now=datetime.now(UTC)awaitconn.execute(f"""INSERT INTO "{schema}".jobs ( id, actor, queue, payload, max_attempts, retry_kind, status, priority, attempt, claim_epoch, 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, $13, clock_timestamp(), $8, $9, clock_timestamp(), clock_timestamp(), $10, $11, $12 )""",# noqa: S608job_id,"test_actor","default",'{"key": "value"}',max_attempts,retry_kind,attempt,worker_id,expires_at,cancel_phase,cancel_requested_at,schedule_to_close,# The epoch seed models the refund-free history this helper# represents: a row that reached attempt N through N claims, each# of which stamped attempt and claim_epoch together, so the row's# epoch is N and the fences under test are satisfiable by# presenting the row's own values. attempt and claim_epoch are# distinct counters that DO diverge in production (the attempt# saturates at the smallint ceiling and non-terminal releases# refund it; the epoch only ever increments, so the row's epoch is# always >= its attempt); the equality fences treat a diverged row# identically, and a test that needs a diverged shape seeds the# two columns separately.attempt,)ifwith_events:detail=dumps_str({"from_state":"pending","to_state":"running","worker_id":str(worker_id)})awaitconn.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,)returnjob_id
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.
asyncdefcreate_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_idornew_uuid()await_create_worker(conn,schema,wid)jid=awaitcreate_running_job(conn,schema,wid,**job_kwargs)returnwid,jid
asyncdefget_job_triple(conn:_Conn,schema:str,job_id:UUID)->JobTriple:ifnot_IDENT_RE.match(schema):raiseValueError(f"invalid schema name {schema!r}")row=awaitconn.fetchrow(f'SELECT * FROM "{schema}".jobs WHERE id = $1',# noqa: S608job_id,)assertrowisnotNoneattempts=awaitconn.fetch(f'SELECT * FROM "{schema}".job_attempts WHERE job_id = $1',# noqa: S608job_id,)events=awaitconn.fetch(f'SELECT * FROM "{schema}".job_events WHERE job_id = $1 ORDER BY occurred_at',# noqa: S608job_id,)returnJobTriple(row=row,attempts=list(attempts),events=list(events))
asyncdefreset_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=[]``. """awaittruncate_schema(conn,schema)awaitseed_actors(conn,schema,actors=actors)
asyncdefseed_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. """ifnot_IDENT_RE.match(schema):raiseValueError(f"invalid schema name {schema!r}")target=actorsifactorsisnotNoneelseDEFAULT_ACTORSawaitconn.executemany(f'INSERT INTO "{schema}".actor_config (actor, queue) VALUES ($1, $2) ON CONFLICT (actor) DO NOTHING',# noqa: S608[(actor,"default")foractorintarget],)
asyncdefsetup_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`. """returnawaitcreate_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,)
asyncdeftruncate_schema(conn:_Conn,schema:str)->None:"""Truncate all dynamic tables in FK-safe order using CASCADE, and drop triggers a previous test added to them. Leaves ``schema_migrations`` intact. Safe to call repeatedly. """ifnot_IDENT_RE.match(schema):raiseValueError(f"invalid schema name {schema!r}")fortablein_TRUNCATE_TABLES:awaitconn.execute(f'TRUNCATE TABLE "{schema}"."{table}" CASCADE')await_reset_triggers(conn,schema)
defmake_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. """returnWorkerSettings.load_from_dict(_build_dict(pg_dsn,**overrides))
defmake_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)
Canonical home: taskq.clock. This module is production code — it
imports only datetime and nothing from taskq.testing — so
import taskq (which re-exports FakeClock from here) never pulls
in the test-double backend under taskq.testing.
Clock and SystemClock remain in taskq.backend.clock (production);
taskq.testing.clock re-exports FakeClock from here for backcompat.
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.
defmonotonic(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()
Backcompat shim: re-export FakeClock from its canonical home.
FakeClock lives in :mod:taskq.clock (production, dependency-free —
imports only datetime) so import taskq never pulls in
taskq.testing and its test-double backend. Import either way:
from taskq.clock import FakeClock # canonical
from taskq.testing.clock import FakeClock # backcompat shim
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.
defmonotonic(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()
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. The module defines 17 fixtures (sync via
pytest.fixture, async via pytest_asyncio.fixture) plus the
run_isolation_token helper and RUN_TOKEN_ENV_VAR used to isolate
per-run/per-worker database and schema names, and the RedisContainerLike
protocol that redis_url_for reads a running (or killable) container as:
Connect budget for every fixture client that dials a test broker.
redis-py 8 defaults socket_connect_timeout to 5s. On a loaded runner
that budget is a red band, not a safety net: the testcontainer's address
resolution (getaddrinfo, in the loop's executor) can out-run 5s and
the first command dies with Timeout connecting to server — the
rotating one-test-in-~12.7k red (PR #682's 3.13 leg). The connect phase
is setup weather, never a pin's subject — the pins' own bounds are the
tests' overall_timeouts and the suite-wide pytest-timeout — so every
fixture client dials with this budget instead: the tree's outside-system
bound (:data:~taskq.client._transport.POLL_FAILURE_BUDGET_SECS, the
reload factory timeout, the notify reconnect ceiling — all 30).
Protocol for the callable yielded by the actor_runner fixture.
actor_fn accepts a bare handler function or an
:class:~taskq.actor.ActorRef (the decorated handle); the runner
omits the ctx argument when the handler does not declare one,
mirroring the production dispatch decision
(worker/dispatch.py reads actor_ref.wants_ctx).
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.
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.
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).
@pytest_asyncio.fixtureasyncdefmemory_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)foractorinDEFAULT_ACTORS:backend.register_actor_config(actor=actor)yieldbackend
Yield a callable that constructs a synthetic JobContext and
calls actor_fn(payload, ctx), or actor_fn(payload) when the
handler declares no ctx parameter (a bare payload-only
handler, or an :class:~taskq.actor.ActorRef whose wants_ctx is
false; calling a no-ctx actor with a context raises TypeError in
production, so the fixture must mirror that contract).
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. Only the dependency parameters
the handler declared are injected as keyword arguments (the same
selectivity as the production DI pass); the full deps mapping is
also available to the actor as ctx.deps.
@pytest.fixturedefactor_runner()->ActorRunnerCallable:"""Yield a callable that constructs a synthetic ``JobContext`` and calls ``actor_fn(payload, ctx)``, or ``actor_fn(payload)`` when the handler declares no ``ctx`` parameter (a bare ``payload``-only handler, or an :class:`~taskq.actor.ActorRef` whose ``wants_ctx`` is false; calling a no-ctx actor with a context raises ``TypeError`` in production, so the fixture must mirror that contract). 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. Only the dependency parameters the handler declared are injected as keyword arguments (the same selectivity as the production DI pass); the full ``deps`` mapping is also available to the actor as ``ctx.deps``. """asyncdefrun_actor(actor_fn:Callable[...,object]|ActorRef[Any,Any],payload:BaseModel|dict[str,object]|object,*,backend:InMemoryBackend,job_id:JobId|UUID|None=None,attempt:int=1,# The same one-claim default attempt carries: a fresh dispatch# hands the actor attempt 1 and epoch 1, so the context's epoch# fence is satisfiable against a row the fixture's caller just# dispatched. A caller that overrides either value must keep the# pair consistent with the row it dispatched (fences compare the# epoch for equality; a stale one no-ops the terminal write).claim_epoch:int=1,snooze_count:int=0,cancel_event:asyncio.Event|None=None,actor:str="test_actor",queue:str="default",**deps:object,)->object:jid:JobId=JobId(job_id)ifjob_idisnotNoneelsenew_job_id()evt=cancel_eventorasyncio.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:BaseModelifisinstance(payload,BaseModel):ctx_payload=payloadelifisinstance(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,claim_epoch=claim_epoch,snooze_count=snooze_count,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# Why a marker for worker_pool: SubJobEnqueuer only None-checks# the pool (the gate on its autonomous fallback arm, which# writes through the backend and never dereferences it); the# in-memory backend has no asyncpg pool, and without the# marker every ctx.jobs.enqueue would raise "ctx.jobs is only# available inside an actor body".jobs=SubJobEnqueuer(loop_scope_resolved=None,worker_pool=object(),# type: ignore[arg-type] # Why: SubJobEnqueuer only None-checks the pool, it gates its autonomous fallback arm which writes through the backend and never dereferences it; the in-memory backend has no asyncpg pool.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=depsifdepselseNone,)wants_ctx,declared_deps=_handler_call_shape(actor_fn)call_deps=({name:deps[name]fornameindeclared_depsifnameindeps}ifdeclared_depsisnotNoneelse{})# An ActorRef's handler declares a typed payload model, so it is# handed the coerced BaseModel the production worker would pass;# a bare test function keeps the historic raw-payload shape.ref_like=isinstance(getattr(actor_fn,"wants_ctx",None),bool)# Why the cast: the earlier payload-shape narrowing leaves the# parameter's residual union partially unknown (dict[Unknown]);# the fixture only needs the top type here.payload_arg:object=cast("object",ctx_payloadifref_likeelsepayload)ifwants_ctx:# Why the ignore: an ActorRef's __call__ declares# JobContext[Any] for the ctx parameter and that generic is# invariant, so the fixture's JobContext[BaseModel] cannot be# proven assignable even though the runtime contract is exact.result:object=actor_fn(payload_arg,ctx,**call_deps)# type: ignore[arg-type] # Why: see comment above.else:result=actor_fn(payload_arg,**call_deps)ifisinstance(result,Awaitable):result=awaitcast(Awaitable[object],result)returnresultreturnrun_actor
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.
@pytest_asyncio.fixtureasyncdefjobs_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:yieldJobsApp(deps=deps,backend=backend)finally:awaitstack.aclose()
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.
@pytest_asyncio.fixture(params=["memory","pg"],ids=["memory","pg"])asyncdefbackend_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. """ifrequest.param=="memory":clock=FakeClock(start=datetime(2025,1,1,tzinfo=UTC))backend:Backend=InMemoryBackend(clock=clock)foractorinDEFAULT_ACTORS:backend.register_actor_config(actor=actor)yieldbackendelse:ifnotrequest.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:yieldpg_backendfinally:awaitstack.aclose()
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. Skips
with a reason (never errors) when the Docker daemon is unreachable.
@pytest.fixture(scope="session")defredis_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. Skips with a reason (never errors) when the Docker daemon is unreachable. """skip_test_without_docker()state_dir=invocation_state_dir(tmp_path_factory)withshared_service_pair(state_dir)asservices:yield_RedisContainerShim(host=services.redis_host,port=services.redis_port,state_dir=state_dir)
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.
Skips with a reason (never errors) when the Docker daemon is unreachable.
@pytest.fixturedefkillable_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. Skips with a reason (never errors) when the Docker daemon is unreachable. """skip_test_without_docker()fromtestcontainers.community.redisimportRedisContainerwith(RedisContainer(image=DRAGONFLY_IMAGE).with_command(DRAGONFLY_RESOURCE_FLAGS).with_kwargs(labels=creator_labels()))asrc:yieldrc
defredis_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)returnf"redis://{host}:{port}/{db}"
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.
Tenancy contract (what MAY run here, and what MUST NOT)¶
redis_url (and the SHARED Dragonfly behind it) is one broker for every
redis-marked test of the invocation across all xdist workers. A test may
keep it as its broker only if ALL of these hold:
every command is scoped to the test's own logical DB (get/set/eval/
subscribe on unique keys and channels) — no broker-GLOBAL command
(CLIENT PAUSE/CLIENT KILL/CONFIG/FLUSHALL/restart), which
hits every co-tenant's connection at once;
no assertion measures broker latency (a PING budget, a fanout deadline) —
a co-tenant's stall band lands on the shared broker and reds the
assertion without any defect in the code under test;
no assertion depends on wall-clock boundary timing measured across
broker round trips — the same stall band shifts the boundary.
Everything else (chaos kills/restarts, CLIENT PAUSE scenarios, ping
assertions, pub/sub fanout- and boundary-timing pins) takes
:func:private_redis_url — a per-test disposable Dragonfly, ~1s boot, the
same image and flags as the shared pair. The historical record: every
cross/extras-leg "weather" red of this class (the subscriber wake, the
ping pins, the attack478 CLIENT PAUSE) was the shared broker's
co-tenancy, and every cure that patched ONE test's budget found the NEXT
exposed test in the following run. The durable cure is tenancy
discipline, not budget inflation.
@pytest.fixturedefredis_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. Tenancy contract (what MAY run here, and what MUST NOT) ------------------------------------------------------- ``redis_url`` (and the SHARED Dragonfly behind it) is one broker for every redis-marked test of the invocation across all xdist workers. A test may keep it as its broker only if ALL of these hold: * every command is scoped to the test's own logical DB (get/set/eval/ subscribe on unique keys and channels) — no broker-GLOBAL command (``CLIENT PAUSE``/``CLIENT KILL``/``CONFIG``/``FLUSHALL``/restart), which hits every co-tenant's connection at once; * no assertion measures broker latency (a PING budget, a fanout deadline) — a co-tenant's stall band lands on the shared broker and reds the assertion without any defect in the code under test; * no assertion depends on wall-clock boundary timing measured across broker round trips — the same stall band shifts the boundary. Everything else (chaos kills/restarts, ``CLIENT PAUSE`` scenarios, ping assertions, pub/sub fanout- and boundary-timing pins) takes :func:`private_redis_url` — a per-test disposable Dragonfly, ~1s boot, the same image and flags as the shared pair. The historical record: every cross/extras-leg "weather" red of this class (the subscriber wake, the ping pins, the attack478 ``CLIENT PAUSE``) was the shared broker's co-tenancy, and every cure that patched ONE test's budget found the NEXT exposed test in the following run. The durable cure is tenancy discipline, not budget inflation. """returnredis_url_for(redis_container,db=_next_redis_db(redis_container))
Per-test PRIVATE Dragonfly URL — for every test the shared pair must not host.
This is the killable_redis_container + redis_url_for idiom as a
drop-in URL fixture: one function-scoped container (creator_labels()
ownership tags, same image/flags as the shared pair, ~1s boot), DB 0, and
no co-tenants — not even a sibling test on another xdist worker. A test on
private_redis_url may pause it, kill it, time its replies, and assert
fanout deadlines: whatever it does to this broker, it does to NOBODY else,
and a stall band on the shared pair cannot reach its assertions.
Consumers: the CLIENT PAUSE conservation attacks, the ping-budget
pins, the pub/sub fanout-timing pins, the skew-proxy crossings, and the
wall-clock boundary pins — see the tenancy contract on :func:redis_url
for the line between the two brokers.
@pytest.fixturedefprivate_redis_url(killable_redis_container:RedisContainer)->str:"""Per-test PRIVATE Dragonfly URL — for every test the shared pair must not host. This is the ``killable_redis_container`` + ``redis_url_for`` idiom as a drop-in URL fixture: one function-scoped container (``creator_labels()`` ownership tags, same image/flags as the shared pair, ~1s boot), DB 0, and no co-tenants — not even a sibling test on another xdist worker. A test on ``private_redis_url`` may pause it, kill it, time its replies, and assert fanout deadlines: whatever it does to this broker, it does to NOBODY else, and a stall band on the shared pair cannot reach its assertions. Consumers: the ``CLIENT PAUSE`` conservation attacks, the ping-budget pins, the pub/sub fanout-timing pins, the skew-proxy crossings, and the wall-clock boundary pins — see the tenancy contract on :func:`redis_url` for the line between the two brokers. """returnredis_url_for(killable_redis_container,db=0)
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 essential 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.
defrun_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 essential 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)iftoken:returntokenreturnos.environ.get("PYTEST_XDIST_WORKER","master")
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.
@pytest_asyncio.fixture(scope="module")asyncdefmodule_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. """importasyncpgfromtaskq.migrateimportapply_pendingschema_name=_schema_name_from_module(request)conn=awaitasyncpg.connect(pg_dsn)try:awaitconn.execute(f'DROP SCHEMA IF EXISTS "{schema_name}" CASCADE')awaitapply_pending(conn,schema=schema_name)awaitseed_actors(conn,schema_name)finally:awaitconn.close()yieldModulePgSchema(schema_name=schema_name,pg_dsn=pg_dsn)conn=awaitasyncpg.connect(pg_dsn)try:awaitconn.execute(f'DROP SCHEMA IF EXISTS "{schema_name}" CASCADE')finally:awaitconn.close()
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.
@pytest.fixture(scope="module")defmodule_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. """importredisasredis_syncurl=redis_url_for(redis_container,db=_next_redis_db(redis_container))# socket_timeout=None: the setup/teardown flushes open a FRESH# connection per module, and redis-py 8's 5s default read budget loses# the HELLO handshake race on a co-tenanted runner (the EEE setup-error# class on the cross legs). A wedged connection is the suite-wide# pytest-timeout's job, not a fixture budget.# socket_connect_timeout=TEST_REDIS_CONNECT_TIMEOUT_SECS: redis-py 8's# 5s CONNECT default is its own band - on a loaded runner the# testcontainer's address resolution (getaddrinfo in the loop's# executor) can out-run 5s and the first command dies with# ``Timeout connecting to server`` (the red 3.13 leg of PR #682, one# test in ~12.7k). The connect phase is setup weather, not a pin's# subject: bound it at the tree's outside-system budget.withredis_sync.from_url(url,decode_responses=False,socket_timeout=None,socket_connect_timeout=TEST_REDIS_CONNECT_TIMEOUT_SECS,)asclient:client.flushdb()yieldurl# Teardown: FLUSHDB the module's DB via sync client (safe in any context).withredis_sync.from_url(url,decode_responses=False,socket_timeout=None,socket_connect_timeout=TEST_REDIS_CONNECT_TIMEOUT_SECS,)asclient:client.flushdb()
@pytest_asyncio.fixture(scope="module")asyncdefmodule_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. """importasyncpgfromtaskq.connectionsimport(DEFAULT_MAX_CACHED_STATEMENT_LIFETIME,DEFAULT_STATEMENT_CACHE_SIZE,)pool=awaitasyncpg.create_pool(module_pg_schema.pg_dsn,min_size=1,max_size=4,statement_cache_size=DEFAULT_STATEMENT_CACHE_SIZE,max_cached_statement_lifetime=DEFAULT_MAX_CACHED_STATEMENT_LIFETIME,)try:yieldpoolfinally:awaitpool.close()
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.
@pytest_asyncio.fixture(scope="module")asyncdefmodule_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:yieldJobsApp(deps=deps,backend=backend)finally:awaitstack.aclose()
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.
@pytest_asyncio.fixtureasyncdefclean_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. """importasyncpgas_asyncpgconn=await_asyncpg.connect(module_pg_schema.pg_dsn)try:awaitreset_schema(conn,module_pg_schema.schema_name)yieldconnfinally:awaitconn.close()
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.
@pytest_asyncio.fixtureasyncdefworker_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=awaitcreate_workered_running_job(clean_pg_conn,module_pg_schema.schema_name)yieldwid,jid,clean_pg_conn
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.
@pytest_asyncio.fixtureasyncdefclean_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. """importasyncpgconn=awaitasyncpg.connect(module_pg_schema.pg_dsn)try:awaitreset_schema(conn,module_pg_schema.schema_name)finally:awaitconn.close()stack,deps,backend=await_open_pg_backend_on_schema(module_pg_schema.pg_dsn,module_pg_schema.schema_name,)try:yieldJobsApp(deps=deps,backend=backend)finally:awaitstack.aclose()
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.
@pytest.fixturedefclean_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. """importredisasredis_sync# socket_timeout=None: same fresh-connection HELLO race as# module_redis_url above; the per-test flush must not inherit the 5s# read budget on a co-tenanted runner.# socket_connect_timeout: the DNS/connect band (see module_redis_url).client=redis_sync.from_url(module_redis_url,decode_responses=False,socket_timeout=None,socket_connect_timeout=TEST_REDIS_CONNECT_TIMEOUT_SECS,)try:client.flushdb()finally:client.close()returnmodule_redis_url
@pytest_asyncio.fixtureasyncdefclean_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``. """fromredis.asyncioimportfrom_urlasredis_from_url# socket_timeout=None: the client handed to every test of a module# rides the shared co-tenanted broker for the test's whole life; the# suite-wide pytest-timeout owns wedged connections, not a 5s read# budget redis-py 8 started defaulting (the pre-8 behaviour).# socket_connect_timeout: the DNS/connect band (see module_redis_url).client=redis_from_url(clean_redis_url,decode_responses=False,socket_timeout=None,socket_connect_timeout=TEST_REDIS_CONNECT_TIMEOUT_SECS,)try:yieldclientfinally:awaitclient.aclose()
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).
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 \.
defunique_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"/"inmoduleor"\\"inmodule:raiseValueError(f"module label must not contain path separators, got {module!r} ""(it is embedded verbatim in the socket path)")returnf"/tmp/tq-{module}-{os.getpid()}-{new_base62()}.sock"# noqa: S108 # Why: test-only socket files; /tmp is the shortest safe prefix.
ChaosConnection wraps a real asyncpg connection and raises
ChaosException (or a fail_with exception type) just before the configured
call number — counting execute / fetchrow / fetch / fetchval in
execution order — while transaction() still delegates to the real
connection so asyncpg rolls the transaction back. ChaosPool wraps a pool
whose acquire() hands out chaos connections. Import from
taskq.testing.asyncpg_chaos directly:
Provides :class:ChaosConnection and :class:ChaosPool for simulating
mid-transaction failures in integration tests. The wrapper raises
:class:ChaosException on the configured call number, allowing tests
to verify that transaction rollback works correctly when a failure
occurs between SQL statements inside a transaction.
Contract:
:class:ChaosException is raised on the Nth query call (execute,
fetchrow, fetch, fetchval). Query calls are counted in execution
order regardless of method name.
Does not swallow CancelledError, it propagates naturally
from the wrapped connection.
:class:ChaosPool honours acquire(timeout=...) against its
acquire_delay, raising TimeoutError on expiry exactly as an
exhausted asyncpg.Pool does.
Defers all other semantics (transaction management, type codecs,
connection lifecycle) to the wrapped connection.
The transaction() method delegates to the real connection so
asyncpg's transaction management (commit / rollback) works correctly
when ChaosException is raised inside a transaction block.
Raised by :class:ChaosConnection on the configured call number.
Carries the call number for debugging. Does not represent a real
PG error, it simulates a failure between SQL statements inside a
transaction, causing the transaction to roll back.
Async wrapper around an asyncpg Connection that raises
:class:ChaosException on the configured call number.
fail_on_call counts query methods (execute, fetchrow,
fetch, fetchval) in execution order. When the counter
reaches fail_on_call, ChaosException is raised before
the query is sent to PG, simulating a mid-transaction failure.
Set fail_with to raise a different exception type (e.g.
:class:asyncpg.PostgresConnectionError or
:class:asyncpg.QueryCanceledError) instead of the default
:class:ChaosException.
transaction() delegates to the real connection so that asyncpg's
transaction context manager can roll back the real transaction when
ChaosException propagates through async with conn.transaction():.
asyncdefexecute(self,query:str,*args:object,timeout:float|None=None)->str:self._check()returnawaitself._conn.execute(# type: ignore[union-attr] # Why: _conn is typed as `object` at runtime; asyncpg is TYPE_CHECKING-only to avoid transitive importsquery,*args,timeout=timeout)
asyncdeffetchrow(self,query:str,*args:object,timeout:float|None=None)->object|None:self._check()returnawaitself._conn.fetchrow(# type: ignore[union-attr] # Why: _conn is typed as `object` at runtime; asyncpg is TYPE_CHECKING-only to avoid transitive importsquery,*args,timeout=timeout)
asyncdeffetch(self,query:str,*args:object,timeout:float|None=None)->list[object]:self._check()returnawaitself._conn.fetch(# type: ignore[union-attr] # Why: _conn is typed as `object` at runtime; asyncpg is TYPE_CHECKING-only to avoid transitive importsquery,*args,timeout=timeout)
asyncdeffetchval(self,query:str,*args:object,column:int=0,timeout:float|None=None,)->object|None:self._check()returnawaitself._conn.fetchval(# type: ignore[union-attr] # Why: _conn is typed as `object` at runtime; asyncpg is TYPE_CHECKING-only to avoid transitive importsquery,*args,column=column,timeout=timeout)
deftransaction(self,**kwargs:object)->object:returnself._conn.transaction(**kwargs)# type: ignore[union-attr] # Why: _conn is typed as `object` at runtime; asyncpg is TYPE_CHECKING-only to avoid transitive imports
asyncdefclose(self)->None:awaitself._conn.close()# type: ignore[union-attr] # Why: _conn is typed as `object` at runtime; asyncpg is TYPE_CHECKING-only to avoid transitive imports
Pool-like object that yields a :class:ChaosConnection from
acquire().
Used to inject a ChaosConnection into backend methods that acquire
connections from self._worker_pool. Temporarily replace
backend._worker_pool with a ChaosPool to test mid-transaction
failures.
acquire_delay simulates a pool with no free connection: acquire()
waits that many seconds before yielding. Combined with the caller's
timeout=, this is what makes the bounded-wait invariant testable
, a call site that forgets timeout= hangs here exactly as it would
against a wedged Postgres, and one that passes it raises
TimeoutError. Default None means a connection is immediately
available, so acquire() returns without waiting and timeout=
can never fire, matching a real pool that is not exhausted, and
leaving the mid-transaction-failure tests unaffected.
timeout raising (rather than being accepted and silently
discarded) is what makes an acquire bound observable to any test
built on this pool at all.
Deliberate divergences from asyncpg.Pool, all inherent to a
single-connection double and none of them silent: there is no holder
queue (so no contention between concurrent acquirers), no
release()/reset cycle, no closing/uninitialised state (a real
pool raises InterfaceError from acquire() for those before
it waits, regardless of timeout), and no bare-await acquire
form. Tests needing any of those want a real pool.
defacquire(self,*,timeout:float|None=None)->_ChaosAcquireCtx:# ``timeout`` mirrors asyncpg.Pool.acquire and is honoured against# ``acquire_delay``. ASYNC109 is suppressed file-wide via# per-file-ignores in pyproject.toml.return_ChaosAcquireCtx(self._conn,self._acquire_delay,timeout)
These require the [otel] extra (opentelemetry-sdk) and are not
re-exported from the package — import them from taskq.testing.otel
directly. The module's public surface: ListSpanExporter (records finished
spans; query with its span_named / spans_named / events_on methods),
setup_tracer, setup_meter, collect_metrics, counter_value,
counter_data_points, histogram_points, reset_otel_gauge_caches,
restore_otel_enabled, restore_logging_configured, and attribute_str.
The module also exports three autouse pytest fixtures —
_otel_enabled_guard, _otel_gauge_cache_guard, and
_logging_configured_guard — that activate automatically for any test
importing from it:
__all__ on the module is exactly that public surface: the autouse guards
are deliberately left private (the test conftest imports them directly, not
through the contract).
defspan_named(self,name:str)->ReadableSpan|None:"""Return the first finished span with *name*, or None."""forsinself.spans:ifs.name==name:returnsreturnNone
defevents_on(self,span_name:str,event_name:str)->list[Any]:"""Return all events named *event_name* on the first span named *span_name*. Returns an empty list if the span doesn't exist. """span=self.span_named(span_name)ifspanisNone:return[]return[evforevinspan.eventsifev.name==event_name]
defspans_with_kind(self,kind:trace.SpanKind)->list[ReadableSpan]:"""Return all finished spans with the given *kind*."""return[sforsinself.spansifs.kind==kind]
Narrow an OTel attribute value to the str callers assert on.
OTel 1.45.0 widened AttributeValue with Sequence[AnyValue] and
Mapping[str, AnyValue] arms, which are unhashable, so pyright
(correctly) refuses set-membership on a raw attribute lookup. Every
value the metric suites mint is a str; this says so at runtime
too, rather than a comprehension silently dropping a non-string.
defattribute_str(value:AttributeValue)->str:"""Narrow an OTel attribute value to the ``str`` callers assert on. OTel 1.45.0 widened ``AttributeValue`` with ``Sequence[AnyValue]`` and ``Mapping[str, AnyValue]`` arms, which are unhashable, so pyright (correctly) refuses set-membership on a raw attribute lookup. Every value the metric suites mint is a ``str``; this says so at runtime too, rather than a comprehension silently dropping a non-string. """assertisinstance(value,str),valuereturnvalue
defcollect_metrics(reader:InMemoryMetricReader)->list[Metric]:"""Flush the reader and return a flat list of all metrics."""md=reader.get_metrics_data()ifmdisNone:return[]results:list[Metric]=[]forrminmd.resource_metrics:forsminrm.scope_metrics:results.extend(sm.metrics)returnresults
defcounter_data_points(reader:InMemoryMetricReader,name:str)->list[NumberDataPoint]:"""Return NumberDataPoint list for the counter named *name*."""formincollect_metrics(reader):ifm.name==name:return[pforpinm.data.data_pointsifisinstance(p,NumberDataPoint)]return[]
defcounter_value(reader:InMemoryMetricReader,name:str)->int:"""Return the summed counter value for the metric named *name*. Returns 0 if the metric has not been recorded. """formincollect_metrics(reader):ifm.name==name:total=0forpinm.data.data_points:ifisinstance(p,NumberDataPoint):total+=int(p.value)returntotalreturn0
defhistogram_points(reader:InMemoryMetricReader,name:str)->list[HistogramDataPoint]:"""Return histogram data points for the metric named *name*."""formincollect_metrics(reader):ifm.name==name:returnlist(m.data.data_points)# type: ignore[return-value] # Why: histogram metric data_points are always HistogramDataPoint instances.return[]
Reset every process-global in :mod:taskq.obs._otel that a gauge
observer, or a test asserting on one, can read, to construction
state.
Rebind, never mutate in place: the gauge caches follow a copy-on-write
rebind discipline in production precisely because the OTel reader
thread iterates them, and an in-place dict.clear() landing while
any reader's iterator is open is the dictionary changed size during
iteration hazard that discipline exists to prevent. Rebinding is
also coherent for importers: every reader of these globals reads
through the module namespace at call time (worker/health.py reads
_otel._sweep_success_monotonic_cache / _otel._sweep_batch_size_cache
that way, the _observe_* callbacks read the module globals, and
the tests read otel_mod.<attr>), so no holder of a stale object
exists to desynchronize.
Covers the gauge-fed caches (queue depth, stranded jobs, reservation
slots, keyed-reclaim pending, sweep success/batch-size stamps, the
leader lease TTL, jobs-by-status, actor backlog and oldest-pending
age, oldest-due age, scheduled count, running-lease-expired count,
heartbeat consecutive failures, disabled schedules, the slot-pool
occupancy source) plus the two label-admission sets and the cron
failure-level ledger, emitter-side process globals the cardinality
tests were already forced to pin per test because a prior test's
admissions widen a later test's boundary. Deliberately NOT reset:
_lazy_counters / _lazy_histograms (memoized per meter
identity; a stale entry self-replaces on the first call after a meter
swap, and clearing would undo the memoization under test),
_library_tracer / _library_meter (resolve-once resolvers that
are safe across provider transitions; tests patch the accessor
functions, not these values), and _otel_enabled (restored by
_otel_enabled_guard).
defreset_otel_gauge_caches()->None:"""Reset every process-global in :mod:`taskq.obs._otel` that a gauge observer, or a test asserting on one, can read, to construction state. Rebind, never mutate in place: the gauge caches follow a copy-on-write rebind discipline in production precisely because the OTel reader thread iterates them, and an in-place ``dict.clear()`` landing while any reader's iterator is open is the ``dictionary changed size during iteration`` hazard that discipline exists to prevent. Rebinding is also coherent for importers: every reader of these globals reads through the module namespace at call time (``worker/health.py`` reads ``_otel._sweep_success_monotonic_cache`` / ``_otel._sweep_batch_size_cache`` that way, the ``_observe_*`` callbacks read the module globals, and the tests read ``otel_mod.<attr>``), so no holder of a stale object exists to desynchronize. Covers the gauge-fed caches (queue depth, stranded jobs, reservation slots, keyed-reclaim pending, sweep success/batch-size stamps, the leader lease TTL, jobs-by-status, actor backlog and oldest-pending age, oldest-due age, scheduled count, running-lease-expired count, heartbeat consecutive failures, disabled schedules, the slot-pool occupancy source) plus the two label-admission sets and the cron failure-level ledger, emitter-side process globals the cardinality tests were already forced to pin per test because a prior test's admissions widen a later test's boundary. Deliberately NOT reset: ``_lazy_counters`` / ``_lazy_histograms`` (memoized per meter identity; a stale entry self-replaces on the first call after a meter swap, and clearing would undo the memoization under test), ``_library_tracer`` / ``_library_meter`` (resolve-once resolvers that are safe across provider transitions; tests patch the accessor functions, not these values), and ``_otel_enabled`` (restored by ``_otel_enabled_guard``). """otel_mod._queue_depth_cache={}# pyright: ignore[reportPrivateUsage] # Why: test isolation seam, the suite's established pattern for module-global caches with no other reset surface.otel_mod._stranded_jobs_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._queue_live_workers_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._reservation_slots_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._keyed_reclaim_pending=0# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._sweep_success_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._sweep_success_monotonic_cache={}# pyright: ignore[reportPrivateUsage] # Why: the monotonic twin ledger maintenance_health reads; cleared with the wall one so no stamp leaks between tests.otel_mod._sweep_batch_size_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._sweep_batch_size_configured_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._leader_lease_expires_in_seconds_cache=None# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._jobs_by_status_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._actor_backlog_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._actor_oldest_pending_age_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._oldest_due_age_seconds=0.0# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._scheduled_count=0# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._running_lease_expired_count=0# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._heartbeat_consecutive_failures_count=0# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._disabled_schedules_count=0# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._cron_slots_behind_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above - the slots-behind gauge's per-actor depth cache is process state a cron-driving test must not leave behind.otel_mod._slot_pool_occupancy_source=None# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._worker_capacity_source=None# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._jobs_running_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._actor_oldest_running_age_cache={}# pyright: ignore[reportPrivateUsage] # Why: same seam as above.otel_mod._queue_label_values.clear()# pyright: ignore[reportPrivateUsage] # Why: same seam as above, the set is iterated only on the emitter thread, so in-place clear is the copy-on-write exemption's exact case.otel_mod._cron_actor_label_values.clear()# pyright: ignore[reportPrivateUsage] # Why: same seam and same exemption as above.otel_mod._cron_failure_levels.clear()# pyright: ignore[reportPrivateUsage] # Why: same seam as above.
defrestore_logging_configured(saved:bool)->None:"""Restore ``obs._structlog._logging_configured`` to a previously saved value."""structlog_mod._logging_configured=saved# type: ignore[reportPrivateUsage] # Why: testing utility restores private module flag from snapshot.
defrestore_otel_enabled(saved:bool)->None:"""Restore ``obs._otel._otel_enabled`` to a previously saved value."""otel_mod._otel_enabled=saved# type: ignore[reportPrivateUsage] # Why: testing utility restores private module flag from snapshot.
Create a test-scoped MeterProvider and patch obs.get_meter.
By default patches the instruments used by the dispatch and consumer
paths. Pass extra_instruments as {module_attr: instrument_type}
(where type is "counter" or "histogram") to patch additional
module-level instruments.
Returns the InMemoryMetricReader for metric assertions.
defsetup_meter(monkeypatch:pytest.MonkeyPatch,*,extra_instruments:dict[str,str]|None=None,)->InMemoryMetricReader:"""Create a test-scoped MeterProvider and patch ``obs.get_meter``. By default patches the instruments used by the dispatch and consumer paths. Pass *extra_instruments* as ``{module_attr: instrument_type}`` (where type is ``"counter"`` or ``"histogram"``) to patch additional module-level instruments. Returns the ``InMemoryMetricReader`` for metric assertions. """fromopentelemetry.sdk.metricsimportMeterProviderreader=InMemoryMetricReader()new_provider=MeterProvider(metric_readers=[reader])new_meter=new_provider.get_meter(obs_mod.INSTRUMENTATION_NAME,otel_mod._version())# type: ignore[reportPrivateUsage] # Why: testing utility reads private module state for test-scoped meter setup.monkeypatch.setattr(otel_mod,"_dispatch_duration",new_meter.create_histogram("taskq.dispatch.duration",unit="s"),)monkeypatch.setattr(otel_mod,"_dispatch_failures",new_meter.create_counter("taskq.dispatch.failures",unit="1"),)monkeypatch.setattr(otel_mod,"_pool_acquire_duration",new_meter.create_histogram("taskq.dispatch.pool_acquire_duration",unit="s"),)monkeypatch.setattr(otel_mod,"_consumed_messages",new_meter.create_counter("messaging.client.consumed.messages",unit="1"),)monkeypatch.setattr(otel_mod,"_process_duration",new_meter.create_histogram("messaging.process.duration",unit="s"),)monkeypatch.setattr(otel_mod,"_published_messages",new_meter.create_counter("messaging.client.published.messages",unit="1"),)monkeypatch.setattr(otel_mod,"get_meter",lambda:new_meter)monkeypatch.setattr(obs_mod,"get_meter",lambda:new_meter)ifextra_instruments:forattr,kindinextra_instruments.items():ifkind=="counter":monkeypatch.setattr(otel_mod,attr,new_meter.create_counter(attr,unit="1"))elifkind=="histogram":monkeypatch.setattr(otel_mod,attr,new_meter.create_histogram(attr,unit="s"))returnreader
Create a test-scoped TracerProvider and patch obs.get_tracer.
OTel's global TracerProvider has a _Once guard that prevents
trace.set_tracer_provider from being called more than once per
process. Patching obs_mod.get_tracer to return a tracer from
our per-test provider avoids the guard entirely.
Also sets otel_mod._otel_enabled = True so that
safe_start_span delegates to the real tracer.
defsetup_tracer(monkeypatch:pytest.MonkeyPatch,)->tuple[TracerProvider,ListSpanExporter]:"""Create a test-scoped TracerProvider and patch ``obs.get_tracer``. OTel's global TracerProvider has a ``_Once`` guard that prevents ``trace.set_tracer_provider`` from being called more than once per process. Patching ``obs_mod.get_tracer`` to return a tracer from our per-test provider avoids the guard entirely. Also sets ``otel_mod._otel_enabled = True`` so that ``safe_start_span`` delegates to the real tracer. """exporter=ListSpanExporter()provider=TracerProvider()provider.add_span_processor(SimpleSpanProcessor(exporter))test_tracer=provider.get_tracer(obs_mod.INSTRUMENTATION_NAME,otel_mod._version())# type: ignore[reportPrivateUsage] # Why: testing utility reads private module state for test-scoped tracer setup.monkeypatch.setattr(otel_mod,"get_tracer",lambda:test_tracer)monkeypatch.setattr(obs_mod,"get_tracer",lambda:test_tracer)otel_mod.set_otel_enabled(True)returnprovider,exporter