temporalio.testing._workflow._V2WorkflowEnvironment

class documentation

class _V2WorkflowEnvironment(_HasAServer): (source)

Constructors: _V2WorkflowEnvironment(client, server, ts_config), _V2WorkflowEnvironment._create(ts_config, **kwargs)

View In Hierarchy

Dev-server env with per-workflow time skipping via TimeSkipper.

Method __init__ Create a workflow environment from a client.
Method auto_time_skipping_disabled Unsupported on V2 — use with_time_skipping_disabled instead.
Async Method fast_forward Fast-forward this workflow's virtual clock by duration, and wait for it to complete.
Async Method get_current_time A workflow's current virtual time. Requires a handle on V2.
Async Method get_time_skipping_info Fetch a workflow's TimeSkippingInfo via DescribeWorkflowExecution.
Async Method sleep Unsupported on V2 — use fast_forward instead.
Method with_time_skipping_disabled Suspend V2 time-skipping config stamping on newly-started workflows within the block.
Property supports_time_skipping_v2 True if this environment uses the V2 time-skipping server.
Async Class Method _create Undocumented
Instance Variable _ts_skipper Undocumented

Inherited from _HasAServer:

Async Method shutdown Shut down this environment.
Async Class Method _bootstrap_dev_server Start a CLI dev server and connect a client. Shared by non-ts and V2.
Instance Variable _server Undocumented

Inherited from WorkflowEnvironment (via _HasAServer):

Class Method from_client Create a workflow environment from the given client.
Async Class Method start_local Start a full Temporal server locally, downloading if necessary.
Async Class Method start_time_skipping Start a V1 time skipping workflow environment.
Async Class Method start_time_skipping_v2 Start a local Temporal server with per-workflow time skipping enabled.
Async Method __aenter__ Noop for async with support.
Async Method __aexit__ For async with support to just call shutdown.
Async Method connect_client Create another client connected to this environment.
Async Method create_nexus_endpoint Create a Nexus endpoint with the given name and task queue.
Async Method delete_nexus_endpoint Delete a Nexus endpoint.
Property client Client to this environment.
Property supports_time_skipping Whether this environment supports either V1 or V2 time skipping.
Property supports_time_skipping_v1 True if this environment uses the V1 Java time-skipping server (which has limited server features).
Instance Variable _client Undocumented
def __init__(self, client: temporalio.client.Client, server: temporalio.bridge.testing.EphemeralServer, ts_config: TimeSkippingConfig): (source)
overrides temporalio.testing._workflow._HasAServer.__init__

Create a workflow environment from a client.

Most users would use a factory method instead.

def auto_time_skipping_disabled(self) -> Iterator[None]: (source)
overrides temporalio.testing.WorkflowEnvironment.auto_time_skipping_disabled

Unsupported on V2 — use with_time_skipping_disabled instead.

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

Fast-forward this workflow's virtual clock by duration, and wait for it to complete.

Warning

This API is experimental and unstable.

Only supported on time-skipping V2 environments (created via start_time_skipping_v2).

Parameters
handle:temporalio.client.WorkflowHandle[Any, Any]Target workflow execution.
duration:timedelta | float | NoneOne-shot advance by this amount (timedelta or seconds as a float). If None, skip time until workflow completion.
Returns
boolTrue if the fast-forward transition is observed; False if the workflow terminates first.
Raises
RuntimeErrorIf called on a V1 or non-time-skipping environment.
async def get_current_time(self, handle: temporalio.client.WorkflowHandle[Any, Any] | None = None) -> datetime: (source)
overrides temporalio.testing.WorkflowEnvironment.get_current_time

A workflow's current virtual time. Requires a handle on V2.

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

Fetch a workflow's TimeSkippingInfo via DescribeWorkflowExecution.

Warning

This API is experimental and unstable.

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

Only supported on time-skipping V2 environments (created via start_time_skipping_v2).

Raises
RuntimeErrorIf called on a V1 or non-time-skipping environment.
async def sleep(self, duration: timedelta | float): (source)
overrides temporalio.testing.WorkflowEnvironment.sleep

Unsupported on V2 — use fast_forward instead.

def with_time_skipping_disabled(self) -> Iterator[None]: (source)
overrides temporalio.testing.WorkflowEnvironment.with_time_skipping_disabled

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

Warning

This API is experimental and unstable.

Workflows started via client during the block do not receive a time_skipping_config on their start request. Existing workflows are unaffected. No-op on non-V2 environments.

supports_time_skipping_v2: bool = (source)
overrides temporalio.testing.WorkflowEnvironment.supports_time_skipping_v2

True if this environment uses the V2 time-skipping server.

Warning

This API is experimental and unstable.

async def _create(cls, *, ts_config: TimeSkippingConfig, **kwargs: Any) -> _V2WorkflowEnvironment: (source)

Undocumented

_ts_skipper = (source)

Undocumented