From 4d90fc6d083eeb71f6eafac7d88ce42a0828fa93 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 13:09:44 -0600 Subject: [PATCH 1/6] Add durable entity unit testing support Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 643dfa24-59f4-4119-9377-0d9e488df141 --- azure-functions-durable/CHANGELOG.md | 4 + azure-functions-durable/README.md | 35 +++ .../durable_functions/testing/__init__.py | 18 +- .../azure/durable_functions/testing/entity.py | 208 ++++++++++++++++++ .../test_entity_testing_support.py | 159 +++++++++++++ 5 files changed, 420 insertions(+), 4 deletions(-) create mode 100644 azure-functions-durable/azure/durable_functions/testing/entity.py create mode 100644 tests/azure-functions-durable/test_entity_testing_support.py diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 5bea7251..bc5945db 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ADDED +- Added `azure.durable_functions.testing.execute_entity()` for unit testing +function-style and class-based entities without a Functions host or Durable Task +backend. The helper returns the operation result, resulting state, and typed +signal/orchestration actions. - Added `SyncDurableFunctionsClient`. `DFApp.durable_client_input()` now injects the synchronous client into synchronous functions and the asynchronous client into coroutine functions. Both clients support scheduled-task and history-export diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 7063e2db..7205e123 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -33,6 +33,41 @@ Key capabilities include durable orchestrations and sub-orchestrations, durable timers, external events, durable entities, retries, versioning, durable HTTP calls (`context.call_http(...)`), recurring scheduled tasks, and history export. +## Unit testing entities + +Use `execute_entity()` to run one entity operation in-process without a +Functions host or Durable Task backend. It supports v1-style entity functions, +durabletask-native entity functions, and `DurableEntity` subclasses: + +```python +from azure.durable_functions.testing import execute_entity +from durabletask.entities import DurableEntity + + +class Counter(DurableEntity): + def add(self, amount: int) -> int: + value = self.get_state(int, 0) + amount + self.set_state(value) + return value + + +outcome = execute_entity(Counter, "add", input=2, state=3) + +assert outcome.result == 5 +assert outcome.state == 5 +assert outcome.actions == () +``` + +For an `entity_trigger`-decorated function, pass the exposed entity function: + +```python +entity_function = counter.build().get_user_function().entity_function +outcome = execute_entity(entity_function, "add", input=2, state=3) +``` + +The returned `EntityTestResult` includes the operation result, resulting state, +and typed signal or orchestration-start actions scheduled by the operation. + ## Links - [Changelog](CHANGELOG.md) diff --git a/azure-functions-durable/azure/durable_functions/testing/__init__.py b/azure-functions-durable/azure/durable_functions/testing/__init__.py index cc1805b1..cdd77fa3 100644 --- a/azure-functions-durable/azure/durable_functions/testing/__init__.py +++ b/azure-functions-durable/azure/durable_functions/testing/__init__.py @@ -3,14 +3,24 @@ """Unit-testing utilities for Azure Durable Functions. -These helpers let you exercise orchestrator business logic in a plain unit test -without a running Functions host or a Durable Task backend. They mirror the -``azure.durable_functions.testing`` surface from the v1 SDK so existing tests -keep working against v2.x. +These helpers let you exercise orchestrator and entity business logic in a +plain unit test without a running Functions host or a Durable Task backend. """ +from .entity import ( + EntityAction, + EntitySignalAction, + EntityTestResult, + OrchestrationStartAction, + execute_entity, +) from .orchestrator_generator_wrapper import orchestrator_generator_wrapper __all__ = [ + "EntityAction", + "EntitySignalAction", + "EntityTestResult", + "OrchestrationStartAction", + "execute_entity", "orchestrator_generator_wrapper", ] diff --git a/azure-functions-durable/azure/durable_functions/testing/entity.py b/azure-functions-durable/azure/durable_functions/testing/entity.py new file mode 100644 index 00000000..5d137d02 --- /dev/null +++ b/azure-functions-durable/azure/durable_functions/testing/entity.py @@ -0,0 +1,208 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""In-process execution support for unit-testing durable entities.""" + +from __future__ import annotations + +import inspect +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Callable, TypeAlias + +from durabletask.entities import DurableEntity, EntityContext, EntityInstanceId +from durabletask.internal import type_discovery +from durabletask.internal.entity_state_shim import StateShim +import durabletask.internal.orchestrator_service_pb2 as pb + +from ..internal.compat.entity_context import wrap_entity +from ..internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER + + +@dataclass(frozen=True) +class EntitySignalAction: + """A signal scheduled by an entity operation.""" + + entity_id: EntityInstanceId + operation: str + input: Any = None + scheduled_time: datetime | None = None + + +@dataclass(frozen=True) +class OrchestrationStartAction: + """An orchestration start scheduled by an entity operation.""" + + name: str + instance_id: str + input: Any = None + + +EntityAction: TypeAlias = EntitySignalAction | OrchestrationStartAction + + +@dataclass(frozen=True) +class EntityTestResult: + """The observable outcome of executing one entity operation.""" + + result: Any + state: Any + actions: tuple[EntityAction, ...] + + +def execute_entity( + entity: Callable[..., Any], + operation: str, + input: Any = None, + state: Any = None, + *, + entity_name: str | None = None, + entity_key: str = "test", +) -> EntityTestResult: + """Execute one entity operation without a Functions host or backend. + + The ``entity`` argument may be a v1-style one-argument function, a + durabletask-native two-argument function, or a + :class:`~durabletask.entities.DurableEntity` subclass. Decorated entities can + be obtained from the function builder's exposed ``entity_function`` handle. + + Parameters + ---------- + entity : Callable[..., Any] + The entity function or ``DurableEntity`` subclass to execute. + operation : str + The entity operation name. + input : Any, optional + The operation input. + state : Any, optional + The state visible at the start of the operation. + entity_name : str, optional + The entity name exposed through its context. Defaults to the configured + durable entity name or the callable's ``__name__``. + entity_key : str, optional + The entity key exposed through its context. Defaults to ``"test"``. + + Returns + ------- + EntityTestResult + The operation result, resulting state, and scheduled actions. + + Raises + ------ + AttributeError + If a class-based entity does not define ``operation``. + TypeError + If the entity or operation is not callable. + Exception + Any exception raised by entity code or payload serialization. + """ + if not callable(entity): + raise TypeError("entity must be a callable or DurableEntity subclass") + + entity_callable = wrap_entity(entity) + resolved_name = ( + entity_name + or getattr(entity_callable, "__durable_entity_name__", None) + or getattr(entity_callable, "__name__", None) + ) + if not resolved_name: + raise ValueError( + "entity_name is required when the entity has no __name__") + + converter = DEFAULT_FUNCTIONS_DATA_CONVERTER + entity_id = EntityInstanceId(resolved_name, entity_key) + state_shim = StateShim(state, converter) + context = EntityContext( + str(entity_id), operation, state_shim, entity_id, converter) + + encoded_input = converter.serialize(input) + input_type = ( + type_discovery.entity_input_type( + entity_callable, operation, converter) + if encoded_input is not None + else None + ) + operation_input = converter.deserialize(encoded_input, input_type) + + try: + result = _invoke_entity( + entity_callable, operation, context, operation_input) + encoded_result = converter.serialize(result) + result = converter.deserialize(encoded_result) + state_shim.commit() + except Exception: + state_shim.rollback() + raise + + actions = tuple( + _decode_action(action) for action in state_shim.get_operation_actions()) + return EntityTestResult( + result=result, + state=state_shim.get_state(), + actions=actions, + ) + + +def _invoke_entity( + entity: Callable[..., Any], + operation: str, + context: EntityContext, + operation_input: Any, +) -> Any: + if isinstance(entity, type) and issubclass(entity, DurableEntity): + instance = entity() + if not hasattr(instance, operation): + raise AttributeError( + f"Entity '{context.entity_id}' does not have operation " + f"'{operation}'") + + method = getattr(instance, operation) + if not callable(method): + raise TypeError(f"Entity operation '{operation}' is not callable") + + instance._initialize_entity_context( # pyright: ignore[reportPrivateUsage] + context) + signature = inspect.signature(method) + has_required_parameter = any( + parameter.default is inspect.Parameter.empty + for parameter in signature.parameters.values() + if parameter.kind not in ( + inspect.Parameter.VAR_POSITIONAL, + inspect.Parameter.VAR_KEYWORD, + ) + ) + if has_required_parameter or operation_input is not None: + return method(operation_input) + return method() + + return entity(context, operation_input) + + +def _decode_action(action: pb.OperationAction) -> EntityAction: + converter = DEFAULT_FUNCTIONS_DATA_CONVERTER + if action.HasField("sendSignal"): + signal = action.sendSignal + scheduled_time = ( + signal.scheduledTime.ToDatetime(tzinfo=timezone.utc) + if signal.HasField("scheduledTime") + else None + ) + encoded_input = ( + signal.input.value if signal.HasField("input") else None) + return EntitySignalAction( + entity_id=EntityInstanceId.parse(signal.instanceId), + operation=signal.name, + input=converter.deserialize(encoded_input), + scheduled_time=scheduled_time, + ) + + if action.HasField("startNewOrchestration"): + start = action.startNewOrchestration + encoded_input = start.input.value if start.HasField("input") else None + return OrchestrationStartAction( + name=start.name, + instance_id=start.instanceId, + input=converter.deserialize(encoded_input), + ) + + raise ValueError("Unsupported entity action") diff --git a/tests/azure-functions-durable/test_entity_testing_support.py b/tests/azure-functions-durable/test_entity_testing_support.py new file mode 100644 index 00000000..1bd8fe29 --- /dev/null +++ b/tests/azure-functions-durable/test_entity_testing_support.py @@ -0,0 +1,159 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Unit tests for host-free durable entity execution.""" + +from datetime import datetime, timezone +from typing import Any + +import pytest + +import azure.durable_functions as df +from azure.durable_functions.testing import ( + EntitySignalAction, + OrchestrationStartAction, + execute_entity, +) +from durabletask.entities import DurableEntity, EntityContext, EntityInstanceId + + +def test_execute_v1_entity_returns_result_and_state(): + def counter(context: df.DurableEntityContext) -> None: + current = context.get_state(initializer=lambda: 0) + current += context.get_input() + context.set_state(current) + context.set_result(current) + + outcome = execute_entity( + counter, "add", input=5, state=10, entity_key="counter-1") + + assert outcome.result == 15 + assert outcome.state == 15 + assert outcome.actions == () + + +def test_execute_native_function_exposes_identity(): + def probe(context: EntityContext, input: Any = None) -> dict[str, str]: + return { + "entity": context.entity_id.entity, + "key": context.entity_id.key, + "operation": context.operation, + "input": input, + } + + outcome = execute_entity( + probe, + "describe", + input="value", + entity_name="CustomProbe", + entity_key="probe-1", + ) + + assert outcome.result == { + "entity": "customprobe", + "key": "probe-1", + "operation": "describe", + "input": "value", + } + assert outcome.state is None + + +def test_execute_class_entity_supports_state_and_inherited_delete(): + class Counter(DurableEntity): + def add(self, amount: Any = None) -> int: + value = self.get_state(int, 0) + amount + self.set_state(value) + return value + + added = execute_entity(Counter, "add", input=3, state=4) + deleted = execute_entity(Counter, "delete", state=added.state) + + assert added.result == 7 + assert added.state == 7 + assert deleted.result is None + assert deleted.state is None + + +def test_execute_class_entity_calls_optional_input_method_without_argument(): + class Counter(DurableEntity): + def reset(self) -> str: + self.set_state(0) + return "reset" + + outcome = execute_entity(Counter, "reset", state=5) + + assert outcome.result == "reset" + assert outcome.state == 0 + + +def test_execute_entity_returns_typed_actions(): + signal_time = datetime(2030, 1, 2, 3, 4, tzinfo=timezone.utc) + + class Relay(DurableEntity): + def dispatch(self, input: dict[str, Any]) -> str: + self.signal_entity( + EntityInstanceId("Counter", input["key"]), + "add", + input["amount"], + signal_time=signal_time, + ) + return self.schedule_new_orchestration( + "process", + input={"source": input["key"]}, + instance_id="orchestration-1", + ) + + outcome = execute_entity( + Relay, "dispatch", input={"key": "target", "amount": 2}) + + assert outcome.result == "orchestration-1" + assert outcome.actions == ( + EntitySignalAction( + entity_id=EntityInstanceId("counter", "target"), + operation="add", + input=2, + scheduled_time=signal_time, + ), + OrchestrationStartAction( + name="process", + instance_id="orchestration-1", + input={"source": "target"}, + ), + ) + + +def test_execute_entity_supports_decorated_function_handle(): + app = df.DFApp() + + @app.entity_trigger( # pyright: ignore[reportArgumentType] + context_name="context") + def accumulator(context: df.DurableEntityContext) -> None: + value = context.get_state(initializer=lambda: 0) + context.set_state(value + context.get_input()) + context.set_result(value + context.get_input()) + + entity_function = accumulator.build().get_user_function().entity_function # pyright: ignore[reportFunctionMemberAccess] + outcome = execute_entity(entity_function, "add", input=4, state=6) + + assert outcome.result == 10 + assert outcome.state == 10 + + +def test_execute_entity_propagates_operation_failure(): + class FailingEntity(DurableEntity): + def fail(self) -> None: + self.set_state("not persisted") + raise RuntimeError("operation failed") + + with pytest.raises(RuntimeError, match="operation failed"): + execute_entity(FailingEntity, "fail", state="original") + + +def test_execute_entity_rejects_missing_class_operation(): + class Counter(DurableEntity): + pass + + with pytest.raises( + AttributeError, + match="does not have operation 'missing'"): + execute_entity(Counter, "missing") From f2f20b40f7209de66cb4fc3ba6c55fc52257e19f Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 13:36:16 -0600 Subject: [PATCH 2/6] Make decorated entity example self-contained Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 643dfa24-59f4-4119-9377-0d9e488df141 --- azure-functions-durable/README.md | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 7205e123..b9b3ab2a 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -61,8 +61,26 @@ assert outcome.actions == () For an `entity_trigger`-decorated function, pass the exposed entity function: ```python +import azure.durable_functions as df +from azure.durable_functions.testing import execute_entity + + +app = df.DFApp() + + +@app.entity_trigger(context_name="context") +def counter(context: df.DurableEntityContext) -> None: + value = context.get_state(initializer=lambda: 0) + value += context.get_input() + context.set_state(value) + context.set_result(value) + + entity_function = counter.build().get_user_function().entity_function outcome = execute_entity(entity_function, "add", input=2, state=3) + +assert outcome.result == 5 +assert outcome.state == 5 ``` The returned `EntityTestResult` includes the operation result, resulting state, From 51304070d6358f66d8ff732e2622e3d6c09fdd29 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 21:52:34 -0600 Subject: [PATCH 3/6] Support strict entity payload testing Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 643dfa24-59f4-4119-9377-0d9e488df141 --- .../azure/durable_functions/testing/entity.py | 100 +++++++++++------- .../test_entity_testing_support.py | 57 ++++++++++ 2 files changed, 117 insertions(+), 40 deletions(-) diff --git a/azure-functions-durable/azure/durable_functions/testing/entity.py b/azure-functions-durable/azure/durable_functions/testing/entity.py index 5d137d02..361bdce9 100644 --- a/azure-functions-durable/azure/durable_functions/testing/entity.py +++ b/azure-functions-durable/azure/durable_functions/testing/entity.py @@ -7,13 +7,13 @@ import inspect from dataclasses import dataclass -from datetime import datetime, timezone +from datetime import datetime from typing import Any, Callable, TypeAlias from durabletask.entities import DurableEntity, EntityContext, EntityInstanceId from durabletask.internal import type_discovery from durabletask.internal.entity_state_shim import StateShim -import durabletask.internal.orchestrator_service_pb2 as pb +from durabletask.serialization import DataConverter from ..internal.compat.entity_context import wrap_entity from ..internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER @@ -112,8 +112,8 @@ def execute_entity( converter = DEFAULT_FUNCTIONS_DATA_CONVERTER entity_id = EntityInstanceId(resolved_name, entity_key) state_shim = StateShim(state, converter) - context = EntityContext( - str(entity_id), operation, state_shim, entity_id, converter) + context = _TestingEntityContext( + str(entity_id), operation, state_shim, entity_id, converter, state) encoded_input = converter.serialize(input) input_type = ( @@ -127,22 +127,72 @@ def execute_entity( try: result = _invoke_entity( entity_callable, operation, context, operation_input) - encoded_result = converter.serialize(result) - result = converter.deserialize(encoded_result) + converter.serialize(result) state_shim.commit() except Exception: state_shim.rollback() raise - actions = tuple( - _decode_action(action) for action in state_shim.get_operation_actions()) return EntityTestResult( result=result, - state=state_shim.get_state(), - actions=actions, + state=context.current_state, + actions=tuple(context.actions), ) +class _TestingEntityContext(EntityContext): + """Entity context that preserves live values for test assertions.""" + + def __init__( + self, + orchestration_id: str, + operation: str, + state: StateShim, + entity_id: EntityInstanceId, + data_converter: DataConverter, + current_state: Any, + ): + super().__init__( + orchestration_id, operation, state, entity_id, data_converter) + self.current_state = current_state + self.actions: list[EntityAction] = [] + + def set_state(self, new_state: Any) -> None: + super().set_state(new_state) + self.current_state = new_state + + def signal_entity( + self, + entity_instance_id: EntityInstanceId, + operation: str, + input: Any | None = None, + signal_time: datetime | None = None, + ) -> None: + super().signal_entity( + entity_instance_id, operation, input, signal_time) + self.actions.append(EntitySignalAction( + entity_id=entity_instance_id, + operation=operation, + input=input, + scheduled_time=signal_time, + )) + + def schedule_new_orchestration( + self, + orchestration_name: str, + input: Any | None = None, + instance_id: str | None = None, + ) -> str: + resolved_instance_id = super().schedule_new_orchestration( + orchestration_name, input, instance_id) + self.actions.append(OrchestrationStartAction( + name=orchestration_name, + instance_id=resolved_instance_id, + input=input, + )) + return resolved_instance_id + + def _invoke_entity( entity: Callable[..., Any], operation: str, @@ -176,33 +226,3 @@ def _invoke_entity( return method() return entity(context, operation_input) - - -def _decode_action(action: pb.OperationAction) -> EntityAction: - converter = DEFAULT_FUNCTIONS_DATA_CONVERTER - if action.HasField("sendSignal"): - signal = action.sendSignal - scheduled_time = ( - signal.scheduledTime.ToDatetime(tzinfo=timezone.utc) - if signal.HasField("scheduledTime") - else None - ) - encoded_input = ( - signal.input.value if signal.HasField("input") else None) - return EntitySignalAction( - entity_id=EntityInstanceId.parse(signal.instanceId), - operation=signal.name, - input=converter.deserialize(encoded_input), - scheduled_time=scheduled_time, - ) - - if action.HasField("startNewOrchestration"): - start = action.startNewOrchestration - encoded_input = start.input.value if start.HasField("input") else None - return OrchestrationStartAction( - name=start.name, - instance_id=start.instanceId, - input=converter.deserialize(encoded_input), - ) - - raise ValueError("Unsupported entity action") diff --git a/tests/azure-functions-durable/test_entity_testing_support.py b/tests/azure-functions-durable/test_entity_testing_support.py index 1bd8fe29..9a7b39bc 100644 --- a/tests/azure-functions-durable/test_entity_testing_support.py +++ b/tests/azure-functions-durable/test_entity_testing_support.py @@ -17,6 +17,24 @@ from durabletask.entities import DurableEntity, EntityContext, EntityInstanceId +class CustomPayload: + def __init__(self, value: int): + self.value = value + + def to_json(self) -> dict[str, int]: + return {"value": self.value} + + @classmethod + def from_json(cls, value: dict[str, int]) -> "CustomPayload": + return cls(value["value"]) + + def __eq__(self, other: object) -> bool: + return ( + isinstance(other, CustomPayload) + and self.value == other.value + ) + + def test_execute_v1_entity_returns_result_and_state(): def counter(context: df.DurableEntityContext) -> None: current = context.get_state(initializer=lambda: 0) @@ -122,6 +140,45 @@ def dispatch(self, input: dict[str, Any]) -> str: ) +def test_execute_entity_preserves_custom_payloads_in_strict_mode( + monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("AZURE_FUNCTIONS_DURABLE_STRICT_TYPING", "1") + + class PayloadEntity(DurableEntity): + def process(self, input: CustomPayload) -> CustomPayload: + state = CustomPayload(input.value + 1) + self.set_state(state) + self.signal_entity( + EntityInstanceId("target", "one"), + "accept", + CustomPayload(input.value + 2), + ) + self.schedule_new_orchestration( + "process-payload", + input=CustomPayload(input.value + 3), + instance_id="orchestration-1", + ) + return CustomPayload(input.value + 4) + + outcome = execute_entity( + PayloadEntity, "process", input=CustomPayload(1)) + + assert outcome.result == CustomPayload(5) + assert outcome.state == CustomPayload(2) + assert outcome.actions == ( + EntitySignalAction( + entity_id=EntityInstanceId("target", "one"), + operation="accept", + input=CustomPayload(3), + ), + OrchestrationStartAction( + name="process-payload", + instance_id="orchestration-1", + input=CustomPayload(4), + ), + ) + + def test_execute_entity_supports_decorated_function_handle(): app = df.DFApp() From c82f23bf85260fe838ca4a1d91c3d6369f16caee Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 23:04:38 -0600 Subject: [PATCH 4/6] Snapshot entity test outputs Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 643dfa24-59f4-4119-9377-0d9e488df141 --- .../azure/durable_functions/testing/entity.py | 86 ++++++++++++++++--- .../test_entity_testing_support.py | 45 ++++++++++ 2 files changed, 117 insertions(+), 14 deletions(-) diff --git a/azure-functions-durable/azure/durable_functions/testing/entity.py b/azure-functions-durable/azure/durable_functions/testing/entity.py index 361bdce9..db743fc2 100644 --- a/azure-functions-durable/azure/durable_functions/testing/entity.py +++ b/azure-functions-durable/azure/durable_functions/testing/entity.py @@ -7,12 +7,13 @@ import inspect from dataclasses import dataclass -from datetime import datetime +from datetime import datetime, timezone from typing import Any, Callable, TypeAlias from durabletask.entities import DurableEntity, EntityContext, EntityInstanceId from durabletask.internal import type_discovery from durabletask.internal.entity_state_shim import StateShim +import durabletask.internal.orchestrator_service_pb2 as pb from durabletask.serialization import DataConverter from ..internal.compat.entity_context import wrap_entity @@ -111,7 +112,7 @@ def execute_entity( converter = DEFAULT_FUNCTIONS_DATA_CONVERTER entity_id = EntityInstanceId(resolved_name, entity_key) - state_shim = StateShim(state, converter) + state_shim = _TestingStateShim(state, converter) context = _TestingEntityContext( str(entity_id), operation, state_shim, entity_id, converter, state) @@ -127,7 +128,8 @@ def execute_entity( try: result = _invoke_entity( entity_callable, operation, context, operation_input) - converter.serialize(result) + encoded_result = converter.serialize(result) + result = _restore_snapshot(encoded_result, result, converter) state_shim.commit() except Exception: state_shim.rollback() @@ -140,8 +142,24 @@ def execute_entity( ) +class _TestingStateShim(StateShim): + """State shim that exposes the exact action payload just serialized.""" + + def __init__( + self, + start_state: Any, + data_converter: DataConverter, + ): + super().__init__(start_state, data_converter) + self.latest_action: pb.OperationAction | None = None + + def add_operation_action(self, action: pb.OperationAction) -> None: + super().add_operation_action(action) + self.latest_action = action + + class _TestingEntityContext(EntityContext): - """Entity context that preserves live values for test assertions.""" + """Entity context that captures runtime-equivalent serialized snapshots.""" def __init__( self, @@ -150,16 +168,25 @@ def __init__( state: StateShim, entity_id: EntityInstanceId, data_converter: DataConverter, - current_state: Any, + initial_state: Any, ): super().__init__( orchestration_id, operation, state, entity_id, data_converter) - self.current_state = current_state + if not isinstance(state, _TestingStateShim): + raise TypeError("state must be a _TestingStateShim") + self._testing_state = state + self._testing_converter = data_converter + self.current_state = _restore_snapshot( + state.encode_state(), initial_state, data_converter) self.actions: list[EntityAction] = [] def set_state(self, new_state: Any) -> None: super().set_state(new_state) - self.current_state = new_state + self.current_state = _restore_snapshot( + self._testing_state.encode_state(), + new_state, + self._testing_converter, + ) def signal_entity( self, @@ -170,11 +197,21 @@ def signal_entity( ) -> None: super().signal_entity( entity_instance_id, operation, input, signal_time) + action = self._require_latest_action() + signal = action.sendSignal + encoded_input = ( + signal.input.value if signal.HasField("input") else None) + scheduled_time = ( + signal.scheduledTime.ToDatetime(tzinfo=timezone.utc) + if signal.HasField("scheduledTime") + else None + ) self.actions.append(EntitySignalAction( - entity_id=entity_instance_id, - operation=operation, - input=input, - scheduled_time=signal_time, + entity_id=EntityInstanceId.parse(signal.instanceId), + operation=signal.name, + input=_restore_snapshot( + encoded_input, input, self._testing_converter), + scheduled_time=scheduled_time, )) def schedule_new_orchestration( @@ -185,13 +222,34 @@ def schedule_new_orchestration( ) -> str: resolved_instance_id = super().schedule_new_orchestration( orchestration_name, input, instance_id) + action = self._require_latest_action() + start = action.startNewOrchestration + encoded_input = start.input.value if start.HasField("input") else None self.actions.append(OrchestrationStartAction( - name=orchestration_name, - instance_id=resolved_instance_id, - input=input, + name=start.name, + instance_id=start.instanceId, + input=_restore_snapshot( + encoded_input, input, self._testing_converter), )) return resolved_instance_id + def _require_latest_action(self) -> pb.OperationAction: + action = self._testing_state.latest_action + if action is None: + raise RuntimeError("Entity action was not recorded") + return action + + +def _restore_snapshot( + encoded_value: str | None, + value: Any, + converter: DataConverter, +) -> Any: + if value is None: + return None + value_type: type[Any] = type(value) + return converter.deserialize(encoded_value, value_type) + def _invoke_entity( entity: Callable[..., Any], diff --git a/tests/azure-functions-durable/test_entity_testing_support.py b/tests/azure-functions-durable/test_entity_testing_support.py index 9a7b39bc..b9a4a5e7 100644 --- a/tests/azure-functions-durable/test_entity_testing_support.py +++ b/tests/azure-functions-durable/test_entity_testing_support.py @@ -179,6 +179,51 @@ def process(self, input: CustomPayload) -> CustomPayload: ) +def test_execute_entity_snapshots_mutable_state_and_actions_in_strict_mode( + monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("AZURE_FUNCTIONS_DURABLE_STRICT_TYPING", "1") + + class MutatingEntity(DurableEntity): + def mutate(self) -> CustomPayload: + state = CustomPayload(1) + signal_input = CustomPayload(2) + orchestration_input = CustomPayload(3) + + self.set_state(state) + self.signal_entity( + EntityInstanceId("target", "one"), + "accept", + signal_input, + ) + self.schedule_new_orchestration( + "process-payload", + input=orchestration_input, + instance_id="orchestration-1", + ) + + state.value = 10 + signal_input.value = 20 + orchestration_input.value = 30 + return state + + outcome = execute_entity(MutatingEntity, "mutate") + + assert outcome.result == CustomPayload(10) + assert outcome.state == CustomPayload(1) + assert outcome.actions == ( + EntitySignalAction( + entity_id=EntityInstanceId("target", "one"), + operation="accept", + input=CustomPayload(2), + ), + OrchestrationStartAction( + name="process-payload", + instance_id="orchestration-1", + input=CustomPayload(3), + ), + ) + + def test_execute_entity_supports_decorated_function_handle(): app = df.DFApp() From e7317397af89fdf7e4192342cfc6180a9be43553 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 23:09:16 -0600 Subject: [PATCH 5/6] Fix entity snapshot restoration Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 643dfa24-59f4-4119-9377-0d9e488df141 --- .../azure/durable_functions/testing/entity.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/azure-functions-durable/azure/durable_functions/testing/entity.py b/azure-functions-durable/azure/durable_functions/testing/entity.py index db743fc2..d2bda13f 100644 --- a/azure-functions-durable/azure/durable_functions/testing/entity.py +++ b/azure-functions-durable/azure/durable_functions/testing/entity.py @@ -8,7 +8,7 @@ import inspect from dataclasses import dataclass from datetime import datetime, timezone -from typing import Any, Callable, TypeAlias +from typing import Any, Callable, TypeAlias, cast from durabletask.entities import DurableEntity, EntityContext, EntityInstanceId from durabletask.internal import type_discovery @@ -247,8 +247,8 @@ def _restore_snapshot( ) -> Any: if value is None: return None - value_type: type[Any] = type(value) - return converter.deserialize(encoded_value, value_type) + value_type = cast(type[Any], type(value)) + return converter.deserialize(encoded_value, value_type) def _invoke_entity( From a727c8205ca5151d86a2d41652462a6af85b9791 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Wed, 29 Jul 2026 09:57:20 -0600 Subject: [PATCH 6/6] Expose serialized entity test outcomes Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 643dfa24-59f4-4119-9377-0d9e488df141 --- azure-functions-durable/README.md | 17 +- .../azure/durable_functions/testing/entity.py | 215 +++++++++--------- .../test_entity_testing_support.py | 117 +++++----- 3 files changed, 182 insertions(+), 167 deletions(-) diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 4268f2db..1cd626c4 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -52,8 +52,8 @@ class Counter(DurableEntity): outcome = execute_entity(Counter, "add", input=2, state=3) -assert outcome.result == 5 -assert outcome.state == 5 +assert outcome.get_result() == 5 +assert outcome.get_state() == 5 assert outcome.actions == () ``` @@ -78,12 +78,17 @@ def counter(context: df.DurableEntityContext) -> None: entity_function = counter.build().get_user_function().entity_function outcome = execute_entity(entity_function, "add", input=2, state=3) -assert outcome.result == 5 -assert outcome.state == 5 +assert outcome.get_result() == 5 +assert outcome.get_state() == 5 ``` -The returned `EntityTestResult` includes the operation result, resulting state, -and typed signal or orchestration-start actions scheduled by the operation. +The returned `EntityTestResult` provides `get_result()` and `get_state()` +methods plus typed signal or orchestration-start actions scheduled by the +operation. Pass `expected_type` when reconstructing a custom payload: + +```python +assert outcome.get_state(expected_type=CounterState) == CounterState(value=5) +``` ## Links diff --git a/azure-functions-durable/azure/durable_functions/testing/entity.py b/azure-functions-durable/azure/durable_functions/testing/entity.py index d2bda13f..b3a2e9e6 100644 --- a/azure-functions-durable/azure/durable_functions/testing/entity.py +++ b/azure-functions-durable/azure/durable_functions/testing/entity.py @@ -6,9 +6,9 @@ from __future__ import annotations import inspect -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timezone -from typing import Any, Callable, TypeAlias, cast +from typing import Any, Callable, TypeAlias, TypeVar, overload from durabletask.entities import DurableEntity, EntityContext, EntityInstanceId from durabletask.internal import type_discovery @@ -19,6 +19,8 @@ from ..internal.compat.entity_context import wrap_entity from ..internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER +TValue = TypeVar("TValue") + @dataclass(frozen=True) class EntitySignalAction: @@ -26,8 +28,27 @@ class EntitySignalAction: entity_id: EntityInstanceId operation: str - input: Any = None + serialized_input: str | None = field(repr=False) scheduled_time: datetime | None = None + _data_converter: DataConverter = field( + repr=False, compare=False, + default=DEFAULT_FUNCTIONS_DATA_CONVERTER) + + @overload + def get_input(self, expected_type: type[TValue]) -> TValue | None: + ... + + @overload + def get_input(self, expected_type: None = None) -> Any: + ... + + def get_input( + self, + expected_type: type[TValue] | None = None, + ) -> TValue | Any | None: + """Deserialize the signal input, optionally as ``expected_type``.""" + return self._data_converter.deserialize( + self.serialized_input, expected_type) @dataclass(frozen=True) @@ -36,7 +57,26 @@ class OrchestrationStartAction: name: str instance_id: str - input: Any = None + serialized_input: str | None = field(repr=False) + _data_converter: DataConverter = field( + repr=False, compare=False, + default=DEFAULT_FUNCTIONS_DATA_CONVERTER) + + @overload + def get_input(self, expected_type: type[TValue]) -> TValue | None: + ... + + @overload + def get_input(self, expected_type: None = None) -> Any: + ... + + def get_input( + self, + expected_type: type[TValue] | None = None, + ) -> TValue | Any | None: + """Deserialize the orchestration input, optionally as ``expected_type``.""" + return self._data_converter.deserialize( + self.serialized_input, expected_type) EntityAction: TypeAlias = EntitySignalAction | OrchestrationStartAction @@ -46,9 +86,44 @@ class OrchestrationStartAction: class EntityTestResult: """The observable outcome of executing one entity operation.""" - result: Any - state: Any + serialized_result: str | None = field(repr=False) + serialized_state: str | None = field(repr=False) actions: tuple[EntityAction, ...] + _data_converter: DataConverter = field( + repr=False, compare=False, + default=DEFAULT_FUNCTIONS_DATA_CONVERTER) + + @overload + def get_result(self, expected_type: type[TValue]) -> TValue | None: + ... + + @overload + def get_result(self, expected_type: None = None) -> Any: + ... + + def get_result( + self, + expected_type: type[TValue] | None = None, + ) -> TValue | Any | None: + """Deserialize the operation result, optionally as ``expected_type``.""" + return self._data_converter.deserialize( + self.serialized_result, expected_type) + + @overload + def get_state(self, expected_type: type[TValue]) -> TValue | None: + ... + + @overload + def get_state(self, expected_type: None = None) -> Any: + ... + + def get_state( + self, + expected_type: type[TValue] | None = None, + ) -> TValue | Any | None: + """Deserialize the persisted state, optionally as ``expected_type``.""" + return self._data_converter.deserialize( + self.serialized_state, expected_type) def execute_entity( @@ -112,9 +187,9 @@ def execute_entity( converter = DEFAULT_FUNCTIONS_DATA_CONVERTER entity_id = EntityInstanceId(resolved_name, entity_key) - state_shim = _TestingStateShim(state, converter) - context = _TestingEntityContext( - str(entity_id), operation, state_shim, entity_id, converter, state) + state_shim = StateShim(state, converter) + context = EntityContext( + str(entity_id), operation, state_shim, entity_id, converter) encoded_input = converter.serialize(input) input_type = ( @@ -129,126 +204,54 @@ def execute_entity( result = _invoke_entity( entity_callable, operation, context, operation_input) encoded_result = converter.serialize(result) - result = _restore_snapshot(encoded_result, result, converter) state_shim.commit() except Exception: state_shim.rollback() raise + actions = tuple( + _decode_action(action, converter) + for action in state_shim.get_operation_actions() + ) return EntityTestResult( - result=result, - state=context.current_state, - actions=tuple(context.actions), + serialized_result=encoded_result, + serialized_state=state_shim.encode_state(), + actions=actions, + _data_converter=converter, ) -class _TestingStateShim(StateShim): - """State shim that exposes the exact action payload just serialized.""" - - def __init__( - self, - start_state: Any, - data_converter: DataConverter, - ): - super().__init__(start_state, data_converter) - self.latest_action: pb.OperationAction | None = None - - def add_operation_action(self, action: pb.OperationAction) -> None: - super().add_operation_action(action) - self.latest_action = action - - -class _TestingEntityContext(EntityContext): - """Entity context that captures runtime-equivalent serialized snapshots.""" - - def __init__( - self, - orchestration_id: str, - operation: str, - state: StateShim, - entity_id: EntityInstanceId, - data_converter: DataConverter, - initial_state: Any, - ): - super().__init__( - orchestration_id, operation, state, entity_id, data_converter) - if not isinstance(state, _TestingStateShim): - raise TypeError("state must be a _TestingStateShim") - self._testing_state = state - self._testing_converter = data_converter - self.current_state = _restore_snapshot( - state.encode_state(), initial_state, data_converter) - self.actions: list[EntityAction] = [] - - def set_state(self, new_state: Any) -> None: - super().set_state(new_state) - self.current_state = _restore_snapshot( - self._testing_state.encode_state(), - new_state, - self._testing_converter, - ) - - def signal_entity( - self, - entity_instance_id: EntityInstanceId, - operation: str, - input: Any | None = None, - signal_time: datetime | None = None, - ) -> None: - super().signal_entity( - entity_instance_id, operation, input, signal_time) - action = self._require_latest_action() +def _decode_action( + action: pb.OperationAction, + converter: DataConverter, +) -> EntityAction: + if action.HasField("sendSignal"): signal = action.sendSignal - encoded_input = ( - signal.input.value if signal.HasField("input") else None) scheduled_time = ( signal.scheduledTime.ToDatetime(tzinfo=timezone.utc) if signal.HasField("scheduledTime") else None ) - self.actions.append(EntitySignalAction( + return EntitySignalAction( entity_id=EntityInstanceId.parse(signal.instanceId), operation=signal.name, - input=_restore_snapshot( - encoded_input, input, self._testing_converter), + serialized_input=( + signal.input.value if signal.HasField("input") else None), scheduled_time=scheduled_time, - )) + _data_converter=converter, + ) - def schedule_new_orchestration( - self, - orchestration_name: str, - input: Any | None = None, - instance_id: str | None = None, - ) -> str: - resolved_instance_id = super().schedule_new_orchestration( - orchestration_name, input, instance_id) - action = self._require_latest_action() + if action.HasField("startNewOrchestration"): start = action.startNewOrchestration - encoded_input = start.input.value if start.HasField("input") else None - self.actions.append(OrchestrationStartAction( + return OrchestrationStartAction( name=start.name, instance_id=start.instanceId, - input=_restore_snapshot( - encoded_input, input, self._testing_converter), - )) - return resolved_instance_id - - def _require_latest_action(self) -> pb.OperationAction: - action = self._testing_state.latest_action - if action is None: - raise RuntimeError("Entity action was not recorded") - return action - + serialized_input=( + start.input.value if start.HasField("input") else None), + _data_converter=converter, + ) -def _restore_snapshot( - encoded_value: str | None, - value: Any, - converter: DataConverter, -) -> Any: - if value is None: - return None - value_type = cast(type[Any], type(value)) - return converter.deserialize(encoded_value, value_type) + raise ValueError("Unsupported entity action") def _invoke_entity( diff --git a/tests/azure-functions-durable/test_entity_testing_support.py b/tests/azure-functions-durable/test_entity_testing_support.py index b9a4a5e7..cc3b13f7 100644 --- a/tests/azure-functions-durable/test_entity_testing_support.py +++ b/tests/azure-functions-durable/test_entity_testing_support.py @@ -45,8 +45,8 @@ def counter(context: df.DurableEntityContext) -> None: outcome = execute_entity( counter, "add", input=5, state=10, entity_key="counter-1") - assert outcome.result == 15 - assert outcome.state == 15 + assert outcome.get_result() == 15 + assert outcome.get_state() == 15 assert outcome.actions == () @@ -67,13 +67,13 @@ def probe(context: EntityContext, input: Any = None) -> dict[str, str]: entity_key="probe-1", ) - assert outcome.result == { + assert outcome.get_result() == { "entity": "customprobe", "key": "probe-1", "operation": "describe", "input": "value", } - assert outcome.state is None + assert outcome.get_state() is None def test_execute_class_entity_supports_state_and_inherited_delete(): @@ -84,12 +84,12 @@ def add(self, amount: Any = None) -> int: return value added = execute_entity(Counter, "add", input=3, state=4) - deleted = execute_entity(Counter, "delete", state=added.state) + deleted = execute_entity(Counter, "delete", state=added.get_state()) - assert added.result == 7 - assert added.state == 7 - assert deleted.result is None - assert deleted.state is None + assert added.get_result() == 7 + assert added.get_state() == 7 + assert deleted.get_result() is None + assert deleted.get_state() is None def test_execute_class_entity_calls_optional_input_method_without_argument(): @@ -100,8 +100,8 @@ def reset(self) -> str: outcome = execute_entity(Counter, "reset", state=5) - assert outcome.result == "reset" - assert outcome.state == 0 + assert outcome.get_result() == "reset" + assert outcome.get_state() == 0 def test_execute_entity_returns_typed_actions(): @@ -124,20 +124,19 @@ def dispatch(self, input: dict[str, Any]) -> str: outcome = execute_entity( Relay, "dispatch", input={"key": "target", "amount": 2}) - assert outcome.result == "orchestration-1" - assert outcome.actions == ( - EntitySignalAction( - entity_id=EntityInstanceId("counter", "target"), - operation="add", - input=2, - scheduled_time=signal_time, - ), - OrchestrationStartAction( - name="process", - instance_id="orchestration-1", - input={"source": "target"}, - ), - ) + assert outcome.get_result() == "orchestration-1" + signal = outcome.actions[0] + assert isinstance(signal, EntitySignalAction) + assert signal.entity_id == EntityInstanceId("counter", "target") + assert signal.operation == "add" + assert signal.get_input() == 2 + assert signal.scheduled_time == signal_time + + start = outcome.actions[1] + assert isinstance(start, OrchestrationStartAction) + assert start.name == "process" + assert start.instance_id == "orchestration-1" + assert start.get_input() == {"source": "target"} def test_execute_entity_preserves_custom_payloads_in_strict_mode( @@ -163,20 +162,12 @@ def process(self, input: CustomPayload) -> CustomPayload: outcome = execute_entity( PayloadEntity, "process", input=CustomPayload(1)) - assert outcome.result == CustomPayload(5) - assert outcome.state == CustomPayload(2) - assert outcome.actions == ( - EntitySignalAction( - entity_id=EntityInstanceId("target", "one"), - operation="accept", - input=CustomPayload(3), - ), - OrchestrationStartAction( - name="process-payload", - instance_id="orchestration-1", - input=CustomPayload(4), - ), - ) + assert outcome.get_result(expected_type=CustomPayload) == CustomPayload(5) + assert outcome.get_state(expected_type=CustomPayload) == CustomPayload(2) + assert outcome.actions[0].get_input( + expected_type=CustomPayload) == CustomPayload(3) + assert outcome.actions[1].get_input( + expected_type=CustomPayload) == CustomPayload(4) def test_execute_entity_snapshots_mutable_state_and_actions_in_strict_mode( @@ -208,20 +199,36 @@ def mutate(self) -> CustomPayload: outcome = execute_entity(MutatingEntity, "mutate") - assert outcome.result == CustomPayload(10) - assert outcome.state == CustomPayload(1) - assert outcome.actions == ( - EntitySignalAction( - entity_id=EntityInstanceId("target", "one"), - operation="accept", - input=CustomPayload(2), - ), - OrchestrationStartAction( - name="process-payload", - instance_id="orchestration-1", - input=CustomPayload(3), - ), - ) + assert outcome.get_result( + expected_type=CustomPayload) == CustomPayload(10) + assert outcome.get_state( + expected_type=CustomPayload) == CustomPayload(1) + assert outcome.actions[0].get_input( + expected_type=CustomPayload) == CustomPayload(2) + assert outcome.actions[1].get_input( + expected_type=CustomPayload) == CustomPayload(3) + + +def test_execute_entity_returns_wire_shape_without_expected_type_in_strict_mode( + monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("AZURE_FUNCTIONS_DURABLE_STRICT_TYPING", "1") + + class TupleEntity(DurableEntity): + def process(self) -> tuple[int, int]: + value = (1, 2) + self.set_state(value) + self.signal_entity( + EntityInstanceId("target", "one"), "accept", value) + self.schedule_new_orchestration( + "process-tuple", input=value, instance_id="orchestration-1") + return value + + outcome = execute_entity(TupleEntity, "process") + + assert outcome.get_result() == [1, 2] + assert outcome.get_state() == [1, 2] + assert outcome.actions[0].get_input() == [1, 2] + assert outcome.actions[1].get_input() == [1, 2] def test_execute_entity_supports_decorated_function_handle(): @@ -237,8 +244,8 @@ def accumulator(context: df.DurableEntityContext) -> None: entity_function = accumulator.build().get_user_function().entity_function # pyright: ignore[reportFunctionMemberAccess] outcome = execute_entity(entity_function, "add", input=4, state=6) - assert outcome.result == 10 - assert outcome.state == 10 + assert outcome.get_result() == 10 + assert outcome.get_state() == 10 def test_execute_entity_propagates_operation_failure():