From 09acaf6e6ed15e233c6a5deb0bcbda15034fe888 Mon Sep 17 00:00:00 2001 From: Harry Nguyen Date: Tue, 25 Aug 2026 14:47:22 -0700 Subject: [PATCH] fix(testing): handle workflow failure states correctly --- .../execution.py | 4 +- .../executor.py | 12 ++-- .../tests/e2e/child_context_failure_test.py | 63 +++++++++++++++++++ .../tests/event_factory_test.py | 35 +++++++++++ .../tests/execution_test.py | 20 ++++++ .../tests/executor_test.py | 36 +++++++++++ 6 files changed, 162 insertions(+), 8 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/e2e/child_context_failure_test.py diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py index a4f54b53..e894caf1 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py @@ -406,7 +406,9 @@ def complete_success(self, result: str | None, now: datetime | None = None) -> N self.close_status = ExecutionStatus.SUCCEEDED self._end_execution(OperationStatus.SUCCEEDED, now) - def complete_fail(self, error: ErrorObject, now: datetime | None = None) -> None: + def complete_fail( + self, error: ErrorObject | None, now: datetime | None = None + ) -> None: """Complete execution with failure (DecisionType.FAIL_WORKFLOW_EXECUTION).""" self.result = DurableExecutionInvocationOutput( status=InvocationStatus.FAILED, error=error diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py index f11e39f6..3ace22b7 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py @@ -1285,9 +1285,7 @@ def _validate_invocation_response_and_store( ) raise InvalidParameterValueException(msg_failed_result) logger.info("[%s] Execution failed", execution_arn) - self._complete_workflow( - execution_arn, result=None, error=response.error - ) + self._fail_workflow(execution_arn, response.error) case InvocationStatus.SUCCEEDED: if response.error is not None: @@ -1591,7 +1589,7 @@ def _complete_workflow( else: self.complete_execution(execution_arn, result) - def _fail_workflow(self, execution_arn: str, error: ErrorObject): + def _fail_workflow(self, execution_arn: str, error: ErrorObject | None): """Fail workflow with terminal state validation.""" execution = self._store.load(execution_arn) @@ -1671,8 +1669,8 @@ def complete_execution(self, execution_arn: str, result: str | None = None) -> N raise IllegalStateException(msg) self._complete_events(execution_arn=execution_arn) - def fail_execution(self, execution_arn: str, error: ErrorObject) -> None: - """Fail execution with error (FAIL_WORKFLOW_EXECUTION decision).""" + def fail_execution(self, execution_arn: str, error: ErrorObject | None) -> None: + """Fail execution with optional error (FAIL_WORKFLOW_EXECUTION decision).""" logger.error("[%s] Completing execution with error: %s", execution_arn, error) execution: Execution = self._store.load(execution_arn=execution_arn) execution.complete_fail(error=error, now=self._clock.now()) @@ -1688,7 +1686,7 @@ def on_completed(self, execution_arn: str, result: str | None = None) -> None: """Complete execution successfully. Observer method triggered by notifier.""" self.complete_execution(execution_arn, result) - def on_failed(self, execution_arn: str, error: ErrorObject) -> None: + def on_failed(self, execution_arn: str, error: ErrorObject | None) -> None: """Fail execution. Observer method triggered by notifier.""" self.fail_execution(execution_arn, error) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/child_context_failure_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/child_context_failure_test.py new file mode 100644 index 00000000..44e4679b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/child_context_failure_test.py @@ -0,0 +1,63 @@ +"""End-to-end child context failure handling through the test runner.""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import StepConfig +from aws_durable_execution_sdk_python.context import ( + DurableContext, + durable_step, + durable_with_child_context, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.lambda_service import ( + InvocationStatus, + OperationStatus, +) +from aws_durable_execution_sdk_python.retries import RetryPresets +from aws_durable_execution_sdk_python.types import StepContext + +from aws_durable_execution_sdk_python_testing.runner import ( + ContextOperation, + DurableFunctionTestResult, + DurableFunctionTestRunner, +) + + +def test_caught_child_context_failure_does_not_fail_root_execution() -> None: + @durable_step + def failing_step(step_context: StepContext) -> str: # noqa: ARG001 + msg = "Child step failed" + raise RuntimeError(msg) + + @durable_with_child_context + def failing_child(ctx: DurableContext) -> str: + return ctx.step( + failing_step(), + config=StepConfig(retry_strategy=RetryPresets.none()), + ) + + @durable_step + def recovery_step(step_context: StepContext, value: str) -> str: # noqa: ARG001 + return value + + @durable_execution + def handler(event: Any, context: DurableContext) -> str: # noqa: ARG001 + try: + context.run_in_child_context(failing_child(), name="failing-child") + except Exception: + pass + + return context.step(recovery_step("handled")) + + with DurableFunctionTestRunner(handler=handler, execution_timeout=10) as runner: + result: DurableFunctionTestResult = runner.run(input="input str") + + assert result.status is InvocationStatus.SUCCEEDED + assert result.result == json.dumps("handled") + + child_op: ContextOperation = result.get_context("failing-child") + assert child_op.status == OperationStatus.FAILED + assert child_op.error is not None + assert child_op.error.message is not None + assert "Child step failed" in child_op.error.message diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py index 1f4238e3..6418304d 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py @@ -158,6 +158,41 @@ def test_create_execution_failed(): assert event.execution_failed_details.error.payload.message == "Execution failed" +def test_create_execution_failed_without_error_payload(): + from aws_durable_execution_sdk_python.execution import ( + DurableExecutionInvocationOutput, + InvocationStatus, + ) + + operation = create_mock_operation("op-1", status=OperationStatus.FAILED) + operation.end_timestamp = datetime.now(UTC) + + error_result = DurableExecutionInvocationOutput( + status=InvocationStatus.FAILED, + error=None, + ) + context = EventCreationContext.create( + operation=operation, + event_id=3, + durable_execution_arn="arn:test", + start_input=StartDurableExecutionInput( + account_id="123", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=300, + execution_retention_period_days=7, + ), + result=error_result, + include_execution_data=True, + ) + event = Event.create_execution_event(context) + + assert event.event_type == "ExecutionFailed" + assert event.execution_failed_details.error is not None + assert event.execution_failed_details.error.payload is None + + def test_create_execution_timed_out(): from aws_durable_execution_sdk_python.execution import ( DurableExecutionInvocationOutput, diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py index 7002d799..b6719230 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py @@ -507,6 +507,26 @@ def test_complete_fail(): assert execution.result.error == error +def test_complete_fail_without_error(): + """Test complete_fail preserves a missing error payload.""" + start_input = StartDurableExecutionInput( + account_id="123456789012", + function_name="test-function", + function_qualifier="$LATEST", + execution_name="test-execution", + execution_timeout_seconds=300, + execution_retention_period_days=7, + invocation_id="test-invocation-id", + ) + execution = Execution("test-arn", start_input, [Mock()]) + + execution.complete_fail(None) + + assert execution.is_complete is True + assert execution.result.status == InvocationStatus.FAILED + assert execution.result.error is None + + def test_find_operation_exists(): """Test find_operation method when operation exists.""" start_input = StartDurableExecutionInput( diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py index 36ec26bb..fde06ad7 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py @@ -310,6 +310,42 @@ def test_should_complete_workflow_with_error_when_invocation_fails( mock_fail.assert_called_once_with("test-arn", failed_response.error) +def test_validate_invocation_response_failed_without_error_still_fails(): + """FAILED without an error must fail (not succeed), preserving the null error.""" + + store = InMemoryExecutionStore() + executor = Executor(store, Mock(), Mock(), Mock()) + + start_input = StartDurableExecutionInput( + account_id="123456789012", + function_name="test-function", + function_qualifier="$LATEST", + execution_name="test-execution", + execution_timeout_seconds=300, + execution_retention_period_days=7, + invocation_id="test-invocation-id", + ) + execution = Execution.new(start_input) + execution.start() + store.save(execution) + + response = DurableExecutionInvocationOutput( + status=InvocationStatus.FAILED, error=None + ) + + executor._validate_invocation_response_and_store( # noqa: SLF001 + execution.durable_execution_arn, response, execution + ) + + stored = store.load(execution.durable_execution_arn) + assert stored.is_complete is True + assert stored.close_status is not None + assert stored.close_status.value == "FAILED" + assert stored.result is not None + assert stored.result.status == InvocationStatus.FAILED + assert stored.result.error is None + + def test_should_complete_workflow_with_result_when_invocation_succeeds( executor, mock_store, mock_scheduler, mock_invoker, start_input ):