temporalio.testing.WorkflowEnvironment

Workflow environment for testing workflows.

Most developers will want to use the static start_time_skipping to start a test server process that automatically skips time as needed. Alternatively, start_local may be used for a full, local Temporal server with more features. To use an existing server, use from_client.

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.

Create a workflow environment from the given client.

supports_time_skipping will always return False for this environment. sleep will sleep the actual amount of time and get_current_time will return the current time.

Returns
`Self`	The workflow environment that runs against the given client.

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 does. supports_time_skipping will always 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:`str`	Namespace name to use for this environment.
data_converter:`temporalio.converter.DataConverter`	See 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.QueryRejectCondition`	See parameter of the same name on `temporalio.client.Client.connect`.
retry_config:`temporalio.service.RetryConfig \| None`	See 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 \| None`	See parameter of the same name on `temporalio.client.Client.connect`.
tls:`bool \| temporalio.service.TLSConfig`	See parameter of the same name on `temporalio.client.Client.connect`.
ip:`str`	IP address to bind to, or 127.0.0.1 by default.
port:`int \| None`	Port number to bind to, or an OS-provided port by default.
download_dest_dir:`str \| None`	Directory to download binary to if a download is needed. If unset, this is the system's temporary directory.
ui:`bool`	If `True`, will start a UI in the dev server.
runtime:`temporalio.runtime.Runtime \| None`	Specific 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 \| None`	Existing path to the CLI binary. If present, no download will be attempted to fetch the binary.
dev_server_database_filename:`str \| None`	Path to the Sqlite database to use for the dev server. Unset default means only in-memory Sqlite will be used.
dev_server_log_format:`str`	Log format for the dev server.
dev_server_log_level:`str \| None`	Log 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:`str`	Specific 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 \| None`	TTL for the downloaded CLI binary. If unset, it will be cached indefinitely.
ui_port:`int \| None`	UI port to use if UI is enabled.

Start a time skipping workflow environment.

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.DataConverter`	See 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.QueryRejectCondition`	See parameter of the same name on `temporalio.client.Client.connect`.
retry_config:`temporalio.service.RetryConfig \| None`	See 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 \| None`	See parameter of the same name on `temporalio.client.Client.connect`.
port:`int \| None`	Port number to bind to, or an OS-provided port by default.
download_dest_dir:`str \| None`	Directory to download binary to if a download is needed. If unset, this is the system's temporary directory.
runtime:`temporalio.runtime.Runtime \| None`	Specific runtime to use or default if unset.
test_server_existing_path:`str \| None`	Existing path to the test server binary. If present, no download will be attempted to fetch the binary.
test_server_download_version:`str`	Specific 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 \| None`	TTL for the downloaded test server binary. If unset, it will be cached indefinitely.

Noop for async with support.

For async with support to just call shutdown.

Create a workflow environment from a client.

Most users would use a factory methods instead.

Disable any automatic time skipping if this is a time-skipping environment.

This is a context manager for use via with. Usually in time-skipping environments, waiting on a workflow result causes time to automatically skip until the next event. This can disable that. However, this only applies to results awaited inside this context. This will not disable automatic time skipping on previous results.

This has no effect on non-time-skipping environments.

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

Parameters
endpoint_name:`str`	The name of the Nexus endpoint to create.
task_queue:`str`	The task queue to associate with the endpoint.

Get the current time known to this environment.

For non-time-skipping environments this is simply the system time. For time-skipping environments this is whatever time has been skipped to.

Shut down this environment.

Sleep in this environment.

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

Parameters
duration:`timedelta \| float`	Amount of time to sleep.

Client to this environment.

Whether this environment supports time skipping.