temporalio.testing.ActivityEnvironment

class documentation

class ActivityEnvironment: (source)

Constructor: ActivityEnvironment(client)

Activity environment for testing activities.

This environment is used for running activity code that can access the functions in the temporalio.activity module. Use run to run an activity function or any function within an activity context.

Static Method	`default_info`	Get the default activity info used for testing.
Method	`__init__`	Create an ActivityEnvironment for running activity code.
Method	`cancel`	Cancel the activity.
Method	`run`	Run the given callable in an activity context.
Method	`worker_shutdown`	Notify the activity that the worker is shutting down.
Instance Variable	`info`	The info that is returned from `temporalio.activity.info` function. To customize, use `default_info` with `dataclasses.replace` to modify fields.
Instance Variable	`metric_meter`	Metric meter set on the activity context. This must be set before `run`. Changes after the activity has started do not take effect. Default is noop.
Instance Variable	`on_heartbeat`	Function called on each heartbeat invocation by the activity.
Instance Variable	`payload_converter`	Payload converter set on the activity context. This must be set before `run`. Changes after the activity has started do not take effect.
Instance Variable	`_activities`	Undocumented
Instance Variable	`_cancellation_details`	Undocumented
Instance Variable	`_cancelled`	Undocumented
Instance Variable	`_client`	Undocumented
Instance Variable	`_worker_shutdown`	Undocumented

@staticmethod

def default_info() -> temporalio.activity.Info: (source) ¶

Get the default activity info used for testing.

Returns a new default Info instance that can be modified using dataclasses.replace before assigning to the info attribute.

def __init__(self, client: Client | None = None): (source) ¶

Create an ActivityEnvironment for running activity code.

def cancel(self, cancellation_details: temporalio.activity.ActivityCancellationDetails = temporalio.activity.ActivityCancellationDetails(cancel_requested=True)): (source) ¶

Cancel the activity.

This only has an effect on the first call.

Parameters
cancellation_details:`temporalio.activity.ActivityCancellationDetails`	details about the cancellation. These will be accessible through temporalio.activity.cancellation_details() in the activity after cancellation.

def run(self, fn: Callable[_Params, _Return], *args: _Params.args, **kwargs: _Params.kwargs) -> _Return: (source) ¶

Run the given callable in an activity context.

Parameters
fn:`Callable[_Params, _Return]`	The function/callable to run.
*args:`_Params.args`	All positional arguments to the callable.
**kwargs:`_Params.kwargs`	All keyword arguments to the callable.

Returns
`_Return`	The callable's result.

def worker_shutdown(self): (source) ¶

Notify the activity that the worker is shutting down.

This only has an effect on the first call.

info = (source) ¶

The info that is returned from temporalio.activity.info function. To customize, use default_info with dataclasses.replace to modify fields.

metric_meter = (source) ¶

Metric meter set on the activity context. This must be set before run. Changes after the activity has started do not take effect. Default is noop.

on_heartbeat: Callable[..., None] = (source) ¶

Function called on each heartbeat invocation by the activity.

payload_converter = (source) ¶

Payload converter set on the activity context. This must be set before run. Changes after the activity has started do not take effect.

_activities: set[_Activity] = (source) ¶

Undocumented

_cancellation_details = (source) ¶

Undocumented

_cancelled: bool = (source) ¶

Undocumented

_client = (source) ¶

Undocumented

_worker_shutdown: bool = (source) ¶

Undocumented