temporalio.testing.WorkflowEnvironment

temporalio.testing.WorkflowEnvironment

class documentation

class WorkflowEnvironment: (source)

Known subclasses: temporalio.testing._workflow._HasAServer

Constructors: WorkflowEnvironment.from_client(client), WorkflowEnvironment.start_local(namespace, data_converter, interceptors, plugins, ...), WorkflowEnvironment.start_time_skipping(data_converter, interceptors, plugins, default_workflow_query_reject_condition, ...), WorkflowEnvironment.start_time_skipping_v2(ts_config, **kwargs), WorkflowEnvironment(client)

View In Hierarchy

Workflow environment for testing workflows.

Most developers will want to use one of these static methods:

This environment is an async context manager, so it can be used with async with to make sure it shuts down properly. Otherwise, shutdown can be manually called.

To use the environment, simply use the client on it.

Workflows invoked on the workflow environment are automatically configured to have assert failures fail the workflow with the assertion error.

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.
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 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_current_time Get the current time known to this environment.
Async Method get_time_skipping_info Fetch a workflow's TimeSkippingInfo via DescribeWorkflowExecution.
Async Method shutdown Shut down this environment.
Async Method sleep Sleep in this environment.
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_v1 True if this environment uses the V1 Java time-skipping server (which has limited server features).
Property supports_time_skipping_v2 True if this environment uses the V2 time-skipping server.
Instance Variable _client Undocumented
def from_client(cls, client: temporalio.client.Client) -> WorkflowEnvironment: (source)

Create a workflow environment from the given client.

supports_time_skipping_v1 and supports_time_skipping_v2 will both return False for this environment. sleep will sleep the actual amount of time and get_current_time will return the current time.

Parameters
client:temporalio.client.ClientThe client to use for the environment.
Returns
WorkflowEnvironmentThe workflow environment that runs against the given client.
async def start_local(cls, *, namespace: str = 'default', data_converter: temporalio.converter.DataConverter = temporalio.converter.DataConverter.default, interceptors: Sequence[temporalio.client.Interceptor] = [], plugins: Sequence[temporalio.client.Plugin] = [], default_workflow_query_reject_condition: None | temporalio.common.QueryRejectCondition = None, retry_config: temporalio.service.RetryConfig | None = None, rpc_metadata: Mapping[str, str | bytes] = {}, identity: str | None = None, tls: bool | temporalio.service.TLSConfig = False, ip: str = '127.0.0.1', port: int | None = None, download_dest_dir: str | None = None, ui: bool = False, runtime: temporalio.runtime.Runtime | None = None, search_attributes: Sequence[temporalio.common.SearchAttributeKey] = (), dev_server_existing_path: str | None = None, dev_server_database_filename: str | None = None, dev_server_log_format: str = 'pretty', dev_server_log_level: str | None = 'warn', dev_server_download_version: str = 'default', dev_server_extra_args: Sequence[str] = [], dev_server_download_ttl: timedelta | None = None, dev_server_startup_timeout: timedelta | None = None, ui_port: int | None = None) -> WorkflowEnvironment: (source)

Start a full Temporal server locally, downloading if necessary.

This environment is good for testing full server capabilities, but does not support time skipping like start_time_skipping or start_time_skipping_v2 do. supports_time_skipping_v1 and supports_time_skipping_v2 will both return False for this environment. sleep will sleep the actual amount of time and get_current_time will return the current time.

Internally, this uses the Temporal CLI dev server from https://github.com/temporalio/cli. This is a self-contained binary for Temporal using Sqlite persistence. This call will download the CLI to a temporary directory by default if it has not already been downloaded before and dev_server_existing_path is not set.

In the future, the dev server implementation may be changed to another implementation. Therefore, all dev_server_ prefixed parameters are dev-server specific and may not apply to newer versions.

Parameters
namespace:strNamespace name to use for this environment.
data_converter:temporalio.converter.DataConverterSee parameter of the same name on temporalio.client.Client.connect.
interceptors:Sequence[temporalio.client.Interceptor]See parameter of the same name on temporalio.client.Client.connect.
plugins:Sequence[temporalio.client.Plugin]Undocumented
default_workflow_query_reject_condition:None | temporalio.common.QueryRejectConditionSee parameter of the same name on temporalio.client.Client.connect.
retry_config:temporalio.service.RetryConfig | NoneSee parameter of the same name on temporalio.client.Client.connect.
rpc_metadata:Mapping[str, str | bytes]See parameter of the same name on temporalio.client.Client.connect.
identity:str | NoneSee parameter of the same name on temporalio.client.Client.connect.
tls:bool | temporalio.service.TLSConfigSee parameter of the same name on temporalio.client.Client.connect.
ip:strIP address to bind to, or 127.0.0.1 by default.
port:int | NonePort number to bind to, or an OS-provided port by default.
download_dest_dir:str | NoneDirectory to download binary to if a download is needed. If unset, this is the system's temporary directory.
ui:boolIf True, will start a UI in the dev server.
runtime:temporalio.runtime.Runtime | NoneSpecific runtime to use or default if unset.
search_attributes:Sequence[temporalio.common.SearchAttributeKey]Search attributes to register with the dev server.
dev_server_existing_path:str | NoneExisting path to the CLI binary. If present, no download will be attempted to fetch the binary.
dev_server_database_filename:str | NonePath to the Sqlite database to use for the dev server. Unset default means only in-memory Sqlite will be used.
dev_server_log_format:strLog format for the dev server.
dev_server_log_level:str | NoneLog level to use for the dev server. Default is warn, but if set to None this will translate the Python logger's level to a dev server log level.
dev_server_download_version:strSpecific CLI version to download. Defaults to default which downloads the version known to work best with this SDK.
dev_server_extra_args:Sequence[str]Extra arguments for the CLI binary.
dev_server_download_ttl:timedelta | NoneTTL for the downloaded CLI binary. If unset, it will be cached indefinitely.
dev_server_startup_timeout:timedelta | NoneMaximum time to wait for the dev server to become ready after spawning. Defaults to 5 seconds and does not include download time.
ui_port:int | NoneUI port to use if UI is enabled.
Returns
WorkflowEnvironmentThe started CLI dev server workflow environment.
async def start_time_skipping(cls, *, data_converter: temporalio.converter.DataConverter = temporalio.converter.DataConverter.default, interceptors: Sequence[temporalio.client.Interceptor] = [], plugins: Sequence[temporalio.client.Plugin] = [], default_workflow_query_reject_condition: None | temporalio.common.QueryRejectCondition = None, retry_config: temporalio.service.RetryConfig | None = None, rpc_metadata: Mapping[str, str | bytes] = {}, identity: str | None = None, port: int | None = None, download_dest_dir: str | None = None, runtime: temporalio.runtime.Runtime | None = None, test_server_existing_path: str | None = None, test_server_download_version: str = 'default', test_server_extra_args: Sequence[str] = [], test_server_download_ttl: timedelta | None = None, test_server_startup_timeout: timedelta | None = None) -> WorkflowEnvironment: (source)

Start a V1 time skipping workflow environment.

This environment uses a legacy Java Temporal server which does not have feature parity with the full temporal server, and which is no longer getting new features.

By default, this environment will automatically skip to the next events in time when a workflow's temporalio.client.WorkflowHandle.result is awaited on (which includes temporalio.client.Client.execute_workflow). Before the result is awaited on, time can be manually skipped forward using sleep. The currently known time can be obtained via get_current_time.

Internally, this environment lazily downloads a test-server binary for the current OS/arch into the temp directory if it is not already there. Then the executable is started and will be killed when shutdown is called (which is implicitly done if this is started via async with await WorkflowEnvironment.start_time_skipping()).

Users can reuse this environment for testing multiple independent workflows, but not concurrently. Time skipping, which is automatically done when awaiting a workflow result and manually done on sleep, is global to the environment, not to the workflow under test.

In the future, the test server implementation may be changed to another implementation. Therefore, all test_server_ prefixed parameters are test server specific and may not apply to newer versions.

Parameters
data_converter:temporalio.converter.DataConverterSee parameter of the same name on temporalio.client.Client.connect.
interceptors:Sequence[temporalio.client.Interceptor]See parameter of the same name on temporalio.client.Client.connect.
plugins:Sequence[temporalio.client.Plugin]Undocumented
default_workflow_query_reject_condition:None | temporalio.common.QueryRejectConditionSee parameter of the same name on temporalio.client.Client.connect.
retry_config:temporalio.service.RetryConfig | NoneSee parameter of the same name on temporalio.client.Client.connect.
rpc_metadata:Mapping[str, str | bytes]See parameter of the same name on temporalio.client.Client.connect.
identity:str | NoneSee parameter of the same name on temporalio.client.Client.connect.
port:int | NonePort number to bind to, or an OS-provided port by default.
download_dest_dir:str | NoneDirectory to download binary to if a download is needed. If unset, this is the system's temporary directory.
runtime:temporalio.runtime.Runtime | NoneSpecific runtime to use or default if unset.
test_server_existing_path:str | NoneExisting path to the test server binary. If present, no download will be attempted to fetch the binary.
test_server_download_version:strSpecific test server version to download. Defaults to default which downloads the version known to work best with this SDK.
test_server_extra_args:Sequence[str]Extra arguments for the test server binary.
test_server_download_ttl:timedelta | NoneTTL for the downloaded test server binary. If unset, it will be cached indefinitely.
test_server_startup_timeout:timedelta | NoneMaximum time to wait for the test server to become ready after spawning. Defaults to 5 seconds and does not include download time.
Returns
WorkflowEnvironmentThe started workflow environment with time skipping.
async def start_time_skipping_v2(cls, *, ts_config: TimeSkippingConfig = TimeSkippingConfig(), **kwargs: Any) -> WorkflowEnvironment: (source)

Start a local Temporal server with per-workflow time skipping enabled.

Warning

This API is experimental and unstable.

This environment is good for testing full server capabilities, and supports time skipping.

Equivalent to start_local plus a TimeSkipper wrap on the client, which stamps time_skipping_config on every workflow started via client and exposes fast_forward for driving time skipping on running workflows. Each workflow has its own virtual clock, unlike time-skipping V1.

See start_local for other keyword arguments.

async def __aenter__(self) -> WorkflowEnvironment: (source)

Noop for async with support.

async def __aexit__(self, *args: Any): (source)

For async with support to just call shutdown.

def __init__(self, client: temporalio.client.Client): (source)
overridden in temporalio.testing._workflow._HasAServer

Create a workflow environment from a client.

Most users would use a factory method instead.

@contextmanager
def auto_time_skipping_disabled(self) -> Iterator[None]: (source)
overridden in temporalio.testing._workflow._V1WorkflowEnvironment, temporalio.testing._workflow._V2WorkflowEnvironment

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

Only meaningful on time-skipping V1 envs. No-op on non-time-skipping envs. Unsupported on V2 envs — use with_time_skipping_disabled to suspend time-skipping config stamping on newly-started workflows instead.

async def connect_client(self, **kwargs: Any) -> temporalio.client.Client: (source)

Create another client connected to this environment.

Namespace and connection credentials from this environment's client are used by default. Keyword arguments are forwarded to temporalio.client.Client.connect and override those defaults.

async def create_nexus_endpoint(self, endpoint_name: str, task_queue: str) -> temporalio.api.nexus.v1.Endpoint: (source)

Create a Nexus endpoint with the given name and task queue.

Parameters
endpoint_name:strThe name of the Nexus endpoint to create.
task_queue:strThe task queue to associate with the endpoint.
Returns
temporalio.api.nexus.v1.EndpointThe created Nexus endpoint.
async def delete_nexus_endpoint(self, endpoint: temporalio.api.nexus.v1.Endpoint): (source)

Delete a Nexus endpoint.

Parameters
endpoint:temporalio.api.nexus.v1.EndpointThe Nexus endpoint to delete.
async def fast_forward(self, handle: temporalio.client.WorkflowHandle[Any, Any], duration: timedelta | float | None = None, /) -> bool: (source)
overridden in temporalio.testing._workflow._V2WorkflowEnvironment

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)
overridden in temporalio.testing._workflow._V1WorkflowEnvironment, temporalio.testing._workflow._V2WorkflowEnvironment

Get the current time known to this environment.

System time on non-time-skipping envs; the V1 test server's virtual clock on V1 envs. On V2 envs a handle is required — each workflow has its own virtual clock, read via TimeSkippingInfo.current_time.

Parameters
handle:temporalio.client.WorkflowHandle[Any, Any] | NoneOn V2 envs, the workflow whose virtual clock to read. Ignored on non-V2 envs.
Returns
datetimeUndocumented
async def get_time_skipping_info(self, handle: temporalio.client.WorkflowHandle[Any, Any]) -> temporalio.api.common.v1.TimeSkippingInfo | None: (source)
overridden in temporalio.testing._workflow._V2WorkflowEnvironment

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 shutdown(self): (source)
overridden in temporalio.testing._workflow._HasAServer

Shut down this environment.

async def sleep(self, duration: timedelta | float): (source)
overridden in temporalio.testing._workflow._V1WorkflowEnvironment, temporalio.testing._workflow._V2WorkflowEnvironment

Sleep in this environment.

This awaits a regular asyncio.sleep in regular environments, or manually skips time in time-skipping environments.

@contextmanager
def with_time_skipping_disabled(self) -> Iterator[None]: (source)
overridden in temporalio.testing._workflow._V2WorkflowEnvironment

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.

client: temporalio.client.Client = (source)

Client to this environment.

supports_time_skipping: bool = (source)

Whether this environment supports either V1 or V2 time skipping.

supports_time_skipping_v1: bool = (source)
overridden in temporalio.testing._workflow._V1WorkflowEnvironment

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

supports_time_skipping_v2: bool = (source)
overridden in temporalio.testing._workflow._V2WorkflowEnvironment

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

Warning

This API is experimental and unstable.

_client = (source)

Undocumented