temporalio.testing._workflow._V1WorkflowEnvironment

class documentation

class _V1WorkflowEnvironment(_HasAServer): (source)

Constructors: _V1WorkflowEnvironment(client, server), _V1WorkflowEnvironment._create(data_converter, interceptors, plugins, default_workflow_query_reject_condition, ...)

View In Hierarchy

Java test-server env with global-clock time skipping.

Method __init__ Create a workflow environment from a client.
Method auto_time_skipping_disabled Disable V1's SDK-driven auto-unlock-on-result-await for the block.
Async Method get_current_time V1 test server's virtual clock.
Async Method sleep Virtual-clock sleep via the V1 test server's test_service.
Async Method time_skipping_unlocked Undocumented
Property supports_time_skipping_v1 True if this environment uses the V1 Java time-skipping server (which has limited server features).
Async Class Method _create Undocumented
Instance Variable _auto_time_skipping 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.
Async Method fast_forward Fast-forward this workflow's virtual clock by duration, and wait for it to complete.
Async Method get_time_skipping_info Fetch a workflow's TimeSkippingInfo via DescribeWorkflowExecution.
Method with_time_skipping_disabled Suspend V2 time-skipping config stamping on newly-started workflows within the block.
Property client Client to this environment.
Property supports_time_skipping Whether this environment supports either V1 or V2 time skipping.
Property supports_time_skipping_v2 True if this environment uses the V2 time-skipping server.
Instance Variable _client Undocumented
def __init__(self, client: temporalio.client.Client, server: temporalio.bridge.testing.EphemeralServer): (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

Disable V1's SDK-driven auto-unlock-on-result-await for the block.

async def get_current_time(self, handle: temporalio.client.WorkflowHandle[Any, Any] | None = None) -> datetime: (source)
overrides temporalio.testing.WorkflowEnvironment.get_current_time

V1 test server's virtual clock.

async def sleep(self, duration: timedelta | float): (source)
overrides temporalio.testing.WorkflowEnvironment.sleep

Virtual-clock sleep via the V1 test server's test_service.

async def time_skipping_unlocked(self) -> AsyncIterator[None]: (source)

Undocumented

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

True if this environment uses the V1 Java time-skipping server (which has limited server features).

async def _create(cls, *, data_converter: temporalio.converter.DataConverter, interceptors: Sequence[temporalio.client.Interceptor], plugins: Sequence[temporalio.client.Plugin], default_workflow_query_reject_condition: None | temporalio.common.QueryRejectCondition, retry_config: temporalio.service.RetryConfig | None, rpc_metadata: Mapping[str, str | bytes], identity: str | None, port: int | None, download_dest_dir: str | None, runtime: temporalio.runtime.Runtime | None, test_server_existing_path: str | None, test_server_download_version: str, test_server_extra_args: Sequence[str], test_server_download_ttl: timedelta | None) -> _V1WorkflowEnvironment: (source)

Undocumented

_auto_time_skipping: bool = (source)

Undocumented