temporalio.testing._workflow._HasAServer

class documentation

class _HasAServer(WorkflowEnvironment): (source)

Known subclasses: temporalio.testing._workflow._NonTsWorkflowEnvironment, temporalio.testing._workflow._V1WorkflowEnvironment, temporalio.testing._workflow._V2WorkflowEnvironment

Constructor: _HasAServer(client, server)

View In Hierarchy

Shared base for envs that own an EphemeralServer to shut down.

Method __init__ Create a workflow environment from a client.
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:

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 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 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 __init__(self, client: temporalio.client.Client, server: temporalio.bridge.testing.EphemeralServer): (source)
overrides temporalio.testing.WorkflowEnvironment.__init__
overridden in temporalio.testing._workflow._NonTsWorkflowEnvironment, temporalio.testing._workflow._V1WorkflowEnvironment, temporalio.testing._workflow._V2WorkflowEnvironment

Create a workflow environment from a client.

Most users would use a factory method instead.

async def shutdown(self): (source)
overrides temporalio.testing.WorkflowEnvironment.shutdown

Shut down this environment.

async def _bootstrap_dev_server(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, ui_port: int | None = None) -> tuple[temporalio.bridge.testing.EphemeralServer, temporalio.client.Client]: (source)

Start a CLI dev server and connect a client. Shared by non-ts and V2.

_server = (source)

Undocumented