temporalio.testing._workflow._EphemeralServerWorkflowEnvironment

class documentation

class _EphemeralServerWorkflowEnvironment(WorkflowEnvironment): (source)

Constructor: _EphemeralServerWorkflowEnvironment(client, server)

View In Hierarchy

Undocumented

Method __init__ Create a workflow environment from a client.
Method auto_time_skipping_disabled Disable any automatic time skipping if this is a time-skipping environment.
Async Method get_current_time Get the current time known to this environment.
Async Method shutdown Shut down this environment.
Async Method sleep Sleep in this environment.
Async Method time_skipping_unlocked Undocumented
Property supports_time_skipping Whether this environment supports time skipping.
Instance Variable _auto_time_skipping Undocumented
Instance Variable _server Undocumented
Instance Variable _supports_time_skipping Undocumented

Inherited from WorkflowEnvironment:

Static Method from_client Create a workflow environment from the given client.
Async Static Method start_local Start a full Temporal server locally, downloading if necessary.
Async Static Method start_time_skipping Start a time skipping workflow environment.
Async Method __aenter__ Noop for async with support.
Async Method __aexit__ For async with support to just call shutdown.
Property client Client to this environment.
Instance Variable _client Undocumented
def __init__(self, client: temporalio.client.Client, server: temporalio.bridge.testing.EphemeralServer): (source)
overrides temporalio.testing.WorkflowEnvironment.__init__

Create a workflow environment from a client.

Most users would use a static method instead.

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

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.

async def get_current_time(self) -> datetime: (source)
overrides temporalio.testing.WorkflowEnvironment.get_current_time

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.

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

Shut down this environment.

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

Sleep in this environment.

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

Parameters
duration:Union[timedelta, float]Amount of time to sleep.
@asynccontextmanager
async def time_skipping_unlocked(self) -> AsyncIterator[None]: (source)

Undocumented

@property
supports_time_skipping: bool = (source)
overrides temporalio.testing.WorkflowEnvironment.supports_time_skipping

Whether this environment supports time skipping.

_auto_time_skipping: bool = (source)

Undocumented

_server = (source)

Undocumented

_supports_time_skipping = (source)

Undocumented