temporalio.testing.TimeSkipper

class documentation

class TimeSkipper: (source)

Constructor: TimeSkipper(client, config)

Client wrapper for per-workflow time skipping.

Warning

This API is experimental and unstable.

Wraps a client with an interceptor that stamps a TimeSkippingConfig on every workflow started through client. Use fast_forward to advance a running workflow's virtual clock (awaited the call to fast_forward waits for the fast-forward to complete and the transition event to fire).

In tests, the same functionality is available via WorkflowEnvironment.start_time_skipping_v2. Use TimeSkipper directly outside test environments.

Method	`__init__`	Create a time skipper.
Method	`config.setter`	Undocumented
Async Method	`fast_forward`	Issue a fast-forward on a workflow and wait for it to complete.
Async Method	`get_current_time`	Get a workflow's current virtual time.
Async Method	`get_time_skipping_info`	Fetch a workflow's `TimeSkippingInfo` via `DescribeWorkflowExecution`.
Method	`with_time_skipping_disabled`	Suspend time-skipping config stamping on newly-started workflows within the block.
Property	`client`	Client that stamps time-skipping config on every started workflow.
Property	`config`	Configuration applied to future start_workflow calls.
Async Method	`_poll_fast_forward_completion`	Long-poll for a bounded fast-forward to complete.
Async Method	`_update_time_skipping_config`	Replace the stored time-skipping config for a running workflow.
Async Method	`_wait_for_unbounded_fast_forward_completion`	Enable unbounded time skipping and wait for the workflow to terminate.
Instance Variable	`_client`	Undocumented
Instance Variable	`_config`	Undocumented
Instance Variable	`_ts_enabled`	Undocumented

def __init__(self, client: temporalio.client.Client, *, config: TimeSkippingConfig = TimeSkippingConfig()): (source) ¶

Create a time skipper.

Parameters
client:`temporalio.client.Client`	Client to wrap. A cloned client is created; the original is untouched.
config:`TimeSkippingConfig`	Default config stamped on every workflow started via `client`.

@config.setter

def config(self, value: TimeSkippingConfig): (source) ¶

Undocumented

async def fast_forward(self, handle: temporalio.client.WorkflowHandle[Any, Any], duration: timedelta | float | None = None, /) -> bool: (source) ¶

Issue a fast-forward on a workflow and wait for it to complete.

Sends an UpdateWorkflowExecutionOptions with the new TimeSkippingConfig. For a bounded duration, long-polls PollWorkflowExecutionTimeSkipping for completion, which follows the workflow across multiple runs.

Parameters
handle:`temporalio.client.WorkflowHandle[Any, Any]`	Target workflow execution.
duration:`timedelta \| float \| None`	One-shot advance by this amount (`timedelta` or seconds as a `float`). If `None`, enable unbounded skipping and wait for the workflow to terminate.

Returns
`For a bounded duration`	True if the fast-forward completes, False if the workflow chain terminates first. Raises `RuntimeError` if the fast-forward id is overridden by a later fast-forward. For `duration=None`: always False (there is no fast-forward completion to observe; the call waits until the workflow's terminal event).

async def get_current_time(self, handle: temporalio.client.WorkflowHandle[Any, Any]) -> datetime: (source) ¶

Get a workflow's current virtual time.

Reads TimeSkippingInfo.current_time via get_time_skipping_info. If time skipping has never been enabled on the workflow, falls back to wall-clock time (the two coincide when no skipping has ever happened).

async def get_time_skipping_info(self, handle: temporalio.client.WorkflowHandle[Any, Any]) -> temporalio.api.common.v1.TimeSkippingInfo | None: (source) ¶

Fetch a workflow's TimeSkippingInfo via DescribeWorkflowExecution.

Returns None if time skipping has never been enabled on the workflow.

@contextmanager

def with_time_skipping_disabled(self) -> Iterator[None]: (source) ¶

Suspend time-skipping config stamping on newly-started workflows within the block.

Workflows started via client inside the block do not get their time_skipping_config set; existing workflows are unaffected.

@property

client: temporalio.client.Client = (source) ¶

Client that stamps time-skipping config on every started workflow.

@property

config: TimeSkippingConfig = (source) ¶

Configuration applied to future start_workflow calls.

async def _poll_fast_forward_completion(self, handle: temporalio.client.WorkflowHandle[Any, Any], fast_forward_id: str) -> bool: (source) ¶

Long-poll for a bounded fast-forward to complete.

Passes run_id="" so the server resolves to the current run in the chain (retry / CAN / cron) and follows the FF across the chain boundary. Retries on server-side long-poll timeout until a terminal poll result arrives.

async def _update_time_skipping_config(self, handle: temporalio.client.WorkflowHandle[Any, Any], config: TimeSkippingConfig) -> temporalio.api.workflowservice.v1.UpdateWorkflowExecutionOptionsResponse: (source) ¶

Replace the stored time-skipping config for a running workflow.

The server accepts the whole TimeSkippingConfig field only, so callers must send a complete config. Public callers should use fast_forward.

async def _wait_for_unbounded_fast_forward_completion(self, handle: temporalio.client.WorkflowHandle[Any, Any]) -> bool: (source) ¶

Enable unbounded time skipping and wait for the workflow to terminate.

There is no fast-forward id to poll on, so we watch history events for a terminal event on the current run. Always returns False — unbounded skipping has no completion event of its own; the wait ends on the workflow's terminal event.

_client = (source) ¶

Undocumented

_config = (source) ¶

Undocumented

_ts_enabled = (source) ¶

Undocumented