diff --git a/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md b/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md index c46f076a1..0a33e4ec1 100644 --- a/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md +++ b/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md @@ -430,8 +430,8 @@ retry budget cannot be bypassed. ### Retry budget -A handler exception or an output validation failure increases the -execution's `consecutive_failed_invocation_attempts`. At +A handler exception increases the execution's +`consecutive_failed_invocation_attempts`. At `MAX_CONSECUTIVE_FAILED_ATTEMPTS` the execution is failed with the last observed error. Below that, a retry is scheduled through `_invoke_execution` after `RETRY_BACKOFF_SECONDS`. Any clean invocation @@ -439,6 +439,15 @@ resets the counter. A clean invocation is one that succeeded or returned PENDING. So a transient failure followed by a success does not count against future budget. +An invalid output is handled as the service handles it. A FAILED output +with a Result, a SUCCEEDED output with an Error, a missing or unknown +Status, or an output that cannot be parsed fails the execution at once. +The error has type `InvalidParameterValueException` and the validation +message. The only invalid output that is retried is PENDING with no +pending operations, because an SDK race that resolves on replay can +produce it. It counts against the same counter, with a cap of +`MAX_CONSECUTIVE_INVALID_PENDING_ATTEMPTS`. + ### Invocation timeout `execution_timeout` and `invocation_timeout` are set on the runner diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py index 8d51e2f18..a243ec914 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py @@ -109,6 +109,20 @@ class DurableFunctionsTestError(Exception): """Base class for testing errors.""" +class InvalidInvocationOutputError(DurableFunctionsTestError): + """The handler's output cannot be parsed as an invocation output.""" + + def __init__(self, message: str, request_id: str | None = None) -> None: + """Initialize with the message and the invocation's request ID. + + Args: + message: Why the output cannot be parsed + request_id: Request ID of the invocation that returned the output + """ + self.request_id = request_id + super().__init__(message) + + # endregion Testing 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 44a01bc01..21d43da6e 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 @@ -54,6 +54,7 @@ ) from aws_durable_execution_sdk_python_testing.exceptions import ( IllegalStateException, + InvalidInvocationOutputError, InvalidParameterValueException, ResourceNotFoundException, ) @@ -126,6 +127,11 @@ class Executor(ExecutionObserver): # target and creating a durable child. Each takes milliseconds. QUICK_DISPATCH_WORKERS: int = 8 MAX_CONSECUTIVE_FAILED_ATTEMPTS: int = 5 + # The service fails an execution on an invalid invocation output at + # once, except PENDING with no pending operations. An SDK race that + # resolves on replay can produce that one, so it gets this many + # consecutive attempts. + MAX_CONSECUTIVE_INVALID_PENDING_ATTEMPTS: int = 3 RETRY_BACKOFF_SECONDS: int = 5 # GetDurableExecutionState page-count bounds, mirroring the service # (default 100, hard cap 1000 operations per page). @@ -1540,8 +1546,8 @@ def _finish_invocation( ) -> None: """Apply a completed handler invocation. - Records the attempt, validates the response (completing or - retrying), resets the retry counter on a clean run, and releases + Records the attempt, validates the response (completing, failing + or retrying), resets the retry counter on a clean run, and releases the gate to PRE_INVOKE (re-invoking if a trigger was deferred) or COMPLETED. """ @@ -1574,11 +1580,20 @@ def _finish_invocation( logger.warning( "[%s] Lambda output validation failure: %s", execution_arn, e ) - error_obj = ErrorObject.from_exception(e) + error_obj = self._invalid_output_error(str(e)) # Release the gate before retry scheduling so the retry's # invoke finds PRE_INVOKE. self._set_invocation_gate(execution_arn, InvocationState.PRE_INVOKE) - self._retry_invocation(execution, error_obj) + # Only PENDING with no pending operations is retried; every + # other invalid output fails the execution at once. + if response.status is InvocationStatus.PENDING: + self._retry_invocation( + execution, + error_obj, + max_attempts=self.MAX_CONSECUTIVE_INVALID_PENDING_ATTEMPTS, + ) + else: + self._fail_workflow(execution_arn, error_obj) return # A clean invocation resets the retry counter. @@ -1616,6 +1631,37 @@ def _fail_invocation_not_found( self._set_invocation_gate(execution_arn, InvocationState.PRE_INVOKE) self._fail_workflow(execution_arn, error) + @staticmethod + def _invalid_output_error(message: str) -> ErrorObject: + """The error the service fails an execution with for an invalid output.""" + return ErrorObject( + message=message, + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) + + def _fail_after_invalid_output( + self, + execution_arn: str, + error: ErrorObject, + invocation_start: datetime, + invocation_end: datetime, + request_id: str | None, + ) -> None: + """Fail the execution for an output the invoker could not parse.""" + execution = self._store.load(execution_arn) + execution.record_invocation_completion( + invocation_start, invocation_end, request_id or str(uuid.uuid4()) + ) + self._store.save(execution) + if execution.is_complete: + # A stop or timeout landed during the invocation and stands. + self._set_invocation_gate(execution_arn, InvocationState.COMPLETED) + return + self._set_invocation_gate(execution_arn, InvocationState.PRE_INVOKE) + self._fail_workflow(execution_arn, error) + def _retry_after_timeout( self, execution_arn: str, @@ -1721,6 +1767,32 @@ async def invoke() -> None: ) ) + except InvalidInvocationOutputError as err: + if self._closing: + return + # The handler ran, but the service cannot parse its output, + # so it fails the execution without invoking it again. + invocation_end = self._clock.now() + logger.warning( + "[%s] Lambda output validation failure: %s", execution_arn, err + ) + error_obj = self._invalid_output_error(str(err)) + request_id = err.request_id + await asyncio.wrap_future( + self._registry.submit( + execution_arn, + CallableTask( + lambda: self._fail_after_invalid_output( + execution_arn, + error_obj, + invocation_start, + invocation_end, + request_id, + ) + ), + ) + ) + except asyncio.TimeoutError: if self._closing: return @@ -1803,22 +1875,26 @@ def _fail_workflow(self, execution_arn: str, error: ErrorObject | None): self.fail_execution(execution_arn, error) - def _retry_invocation(self, execution: Execution, error: ErrorObject): + def _retry_invocation( + self, + execution: Execution, + error: ErrorObject, + max_attempts: int | None = None, + ) -> None: """Handle retry logic or fail execution if retries exhausted. - Budget: ``MAX_CONSECUTIVE_FAILED_ATTEMPTS`` attempts - per ``RETRY_BACKOFF_SECONDS`` backoff. Increments the counter - BEFORE the threshold check so an attempt-N failure that - reaches the ceiling fails the execution rather than scheduling - a pointless retry. + Budget: ``max_attempts`` attempts, by default + ``MAX_CONSECUTIVE_FAILED_ATTEMPTS``, per ``RETRY_BACKOFF_SECONDS`` + backoff. Increments the counter BEFORE the threshold check so an + attempt-N failure that reaches the ceiling fails the execution + rather than scheduling a pointless retry. """ + if max_attempts is None: + max_attempts = self.MAX_CONSECUTIVE_FAILED_ATTEMPTS execution.consecutive_failed_invocation_attempts += 1 self._store.save(execution) - if ( - execution.consecutive_failed_invocation_attempts - >= self.MAX_CONSECUTIVE_FAILED_ATTEMPTS - ): + if execution.consecutive_failed_invocation_attempts >= max_attempts: # Budget exhausted — fail the execution with the last # observed error. self._fail_workflow( diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py index 47e7ac5ed..4992212f4 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from dataclasses import dataclass +from dataclasses import dataclass, replace from threading import Lock from typing import TYPE_CHECKING, Any, Protocol from uuid import uuid4 @@ -15,12 +15,14 @@ DurableExecutionInvocationOutput, InitialExecutionState, ) +from aws_durable_execution_sdk_python.lambda_service import ErrorObject from aws_durable_execution_sdk_python_testing.checkpoint.processor import ( DEFAULT_MAX_INVOCATION_PAGE_BYTES, ) from aws_durable_execution_sdk_python_testing.exceptions import ( DurableFunctionsTestError, + InvalidInvocationOutputError, InvalidParameterValueException, ResourceNotFoundException, ) @@ -107,6 +109,45 @@ class InvokeResponse: invocation_output: DurableExecutionInvocationOutput request_id: str + @classmethod + def from_output(cls, output: Any, request_id: str) -> InvokeResponse: + """Create from the value a handler returned. + + Raises: + InvalidInvocationOutputError: If the value is not an object with + a known ``Status`` and, if present, an object ``Error``. + """ + try: + invocation_output = DurableExecutionInvocationOutput.from_dict(output) + # from_dict treats any falsey Error as absent, but only a missing + # or null Error is. An empty object is kept, so SUCCEEDED with it + # fails validation; a value that is not an object raises here. + error: Any = output.get("Error") + if error is not None and invocation_output.error is None: + invocation_output = replace( + invocation_output, error=ErrorObject.from_dict(error) + ) + except (AttributeError, TypeError, ValueError) as e: + msg: str = f"Invalid invocation output: {e}" + raise InvalidInvocationOutputError(msg, request_id) from e + return cls(invocation_output=invocation_output, request_id=request_id) + + @classmethod + def from_payload(cls, payload: bytes, request_id: str) -> InvokeResponse: + """Create from the payload a Lambda invocation returned. + + Raises: + InvalidInvocationOutputError: If the payload is not UTF-8 JSON or + not a valid invocation output. + """ + try: + # UnicodeDecodeError is a ValueError. + output: Any = json.loads(payload.decode("utf-8")) + except ValueError as e: + msg: str = f"Invalid invocation output: {e}" + raise InvalidInvocationOutputError(msg, request_id) from e + return cls.from_output(output, request_id) + # Values the in-process Lambda context reports when the caller gives none. DEFAULT_TEST_REGION = "us-west-2" @@ -261,10 +302,7 @@ def invoke( ) handler: Callable = self._resolve_handler(function_name) response_dict = handler(input_with_client, context) - output = DurableExecutionInvocationOutput.from_dict(response_dict) - return InvokeResponse( - invocation_output=output, request_id=context.aws_request_id - ) + return InvokeResponse.from_output(response_dict, context.aws_request_id) def update_endpoint(self, endpoint_url: str, region_name: str) -> None: """No-op for in-process invoker.""" @@ -459,6 +497,7 @@ def invoke( Raises: ResourceNotFoundException: If function does not exist InvalidParameterValueException: If parameters are invalid + InvalidInvocationOutputError: If the function's output cannot be parsed DurableFunctionsTestError: For other invocation failures """ @@ -496,22 +535,23 @@ def invoke( msg = f"Lambda invocation failed with status {status_code}: {error_payload}" raise DurableFunctionsTestError(msg) - # Parse response payload - response_payload = response["Payload"].read().decode("utf-8") - response_dict = json.loads(response_payload) - - # Extract request ID from response headers (x-amzn-RequestId or x-amzn-request-id) - headers = response.get("ResponseMetadata", {}).get("HTTPHeaders", {}) - request_id = ( - headers.get("x-amzn-RequestId") + # Read response payload; from_payload decodes it, below + response_payload: bytes = response["Payload"].read() + + # botocore puts the request ID in ResponseMetadata and keeps the + # response headers with lowercased names. + metadata: dict[str, Any] = response.get("ResponseMetadata", {}) + headers: dict[str, str] = { + name.lower(): value + for name, value in metadata.get("HTTPHeaders", {}).items() + } + request_id: str = ( + metadata.get("RequestId") + or headers.get("x-amzn-requestid") or headers.get("x-amzn-request-id") or f"local-{uuid4()}" ) - # Convert to DurableExecutionInvocationOutput - output = DurableExecutionInvocationOutput.from_dict(response_dict) - return InvokeResponse(invocation_output=output, request_id=request_id) - except client.exceptions.ResourceNotFoundException as e: msg = f"Function not found: {function_name}" raise ResourceNotFoundException(msg) from e @@ -565,3 +605,7 @@ def invoke( raise DurableFunctionsTestError(msg) from e msg = f"Unexpected error during Lambda invocation: {e}" raise DurableFunctionsTestError(msg) from e + + # Parsed outside the try: an output the service cannot parse fails + # the execution, so it must not be reported as a failed invocation. + return InvokeResponse.from_payload(response_payload, request_id) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py new file mode 100644 index 000000000..9aea9c713 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py @@ -0,0 +1,163 @@ +"""End-to-end handling of invalid invocation outputs through the test runner.""" + +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.execution import InvocationStatus +from aws_durable_execution_sdk_python.lambda_service import ErrorObject + +from aws_durable_execution_sdk_python_testing.execution import ExecutionStatus +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + DurableFunctionTestRunner, +) +from aws_durable_execution_sdk_python_testing.stores.base import ExecutionStore +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.sqlite import ( + SQLiteExecutionStore, +) + + +@pytest.mark.parametrize( + ("output", "message"), + [ + pytest.param( + {"Status": "FAILED", "Result": "result"}, + "Cannot provide a Result for FAILED status.", + id="failed-with-result", + ), + pytest.param( + {"Status": "SUCCEEDED", "Error": {"ErrorMessage": "boom"}}, + "Cannot provide an Error for SUCCEEDED status.", + id="succeeded-with-error", + ), + pytest.param( + {"Result": "result"}, + "Invalid invocation output: None is not a valid InvocationStatus", + id="missing-status", + ), + pytest.param( + {"Status": "DONE"}, + "Invalid invocation output: 'DONE' is not a valid InvocationStatus", + id="unknown-status", + ), + pytest.param( + {"Status": "RETRY"}, + "Unexpected invocation status: InvocationStatus.RETRY", + id="internal-status", + ), + pytest.param( + ["SUCCEEDED"], + "Invalid invocation output: 'list' object has no attribute 'get'", + id="not-an-object", + ), + ], +) +def test_invalid_output_fails_the_execution_after_one_invocation( + output: Any, message: str +) -> None: + """The service fails the execution on the first invalid output, with an + InvalidParameterValueException error, and does not invoke again.""" + invocations: list[Any] = [] + + def handler(event: Any, context: Any) -> Any: # noqa: ARG001 + invocations.append(event) + return output + + with DurableFunctionTestRunner(handler=handler, execution_timeout=10) as runner: + execution_arn = runner.run_async(input="input str") + result: DurableFunctionTestResult = runner.wait_for_result( + execution_arn, timeout=10 + ) + + assert len(invocations) == 1 + assert result.status is InvocationStatus.FAILED + assert result.execution_status is ExecutionStatus.FAILED + assert result.error == ErrorObject( + message=message, + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) + + +@pytest.mark.parametrize( + "store_factory", + [ + pytest.param(lambda _tmp: InMemoryExecutionStore(), id="memory"), + pytest.param( + lambda tmp: FileSystemExecutionStore.create(tmp / "fs-store"), + id="filesystem", + ), + pytest.param( + lambda tmp: SQLiteExecutionStore.create_and_initialize( + str(tmp / "exec.db") + ), + id="sqlite", + ), + ], +) +@pytest.mark.parametrize( + ("error", "message"), + [ + pytest.param( + {}, + "Cannot provide an Error for SUCCEEDED status.", + id="empty-object", + ), + pytest.param( + [], + "Invalid invocation output: 'list' object has no attribute 'get'", + id="empty-list", + ), + pytest.param( + 0, + "Invalid invocation output: 'int' object has no attribute 'get'", + id="zero", + ), + pytest.param( + False, + "Invalid invocation output: 'bool' object has no attribute 'get'", + id="false", + ), + ], +) +def test_falsey_error_fails_a_succeeded_output_on_each_store( + store_factory: Callable[[Path], ExecutionStore], + error: Any, + message: str, + tmp_path: Path, +) -> None: + """The SDK reads any falsey Error as no Error. An empty Error object is + still an Error, and the other falsey values are not objects, so each + fails the execution after one invocation, on every store.""" + invocations: list[Any] = [] + + def handler(event: Any, context: Any) -> Any: # noqa: ARG001 + invocations.append(event) + return {"Status": "SUCCEEDED", "Error": error} + + with DurableFunctionTestRunner( + handler=handler, store=store_factory(tmp_path), execution_timeout=10 + ) as runner: + execution_arn = runner.run_async(input="input str") + result: DurableFunctionTestResult = runner.wait_for_result( + execution_arn, timeout=10 + ) + + assert len(invocations) == 1 + assert result.status is InvocationStatus.FAILED + assert result.execution_status is ExecutionStatus.FAILED + assert result.error == ErrorObject( + message=message, + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py index 1f1ea5509..57a95366a 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py @@ -25,6 +25,22 @@ def test_durable_functions_test_error_base_exception() -> None: assert isinstance(error, Exception) +def test_invalid_invocation_output_error() -> None: + """Test InvalidInvocationOutputError for handler outputs that cannot be parsed.""" + error = exceptions.InvalidInvocationOutputError("Invalid invocation output: bad") + + assert str(error) == "Invalid invocation output: bad" + assert error.request_id is None + assert isinstance(error, exceptions.DurableFunctionsTestError) + + error = exceptions.InvalidInvocationOutputError( + "Invalid invocation output: bad", request_id="request-1" + ) + + assert str(error) == "Invalid invocation output: bad" + assert error.request_id == "request-1" + + def test_durable_functions_local_runner_error_base_exception() -> None: """Test DurableFunctionsLocalRunnerError base exception.""" error = exceptions.DurableFunctionsLocalRunnerError("Local runner error") diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py index c789dcd74..08edbac70 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py @@ -15,7 +15,10 @@ from unittest.mock import Mock import pytest -from aws_durable_execution_sdk_python.execution import InvocationStatus +from aws_durable_execution_sdk_python.execution import ( + DurableExecutionInvocationOutput, + InvocationStatus, +) from aws_durable_execution_sdk_python.lambda_service import ( ErrorObject, Operation as SvcOperation, @@ -31,8 +34,12 @@ from aws_durable_execution_sdk_python_testing.exceptions import ( InvalidParameterValueException, ) -from aws_durable_execution_sdk_python_testing.execution import Execution +from aws_durable_execution_sdk_python_testing.execution import ( + Execution, + ExecutionStatus, +) from aws_durable_execution_sdk_python_testing.executor import Executor, InvocationState +from aws_durable_execution_sdk_python_testing.invoker import InvokeResponse from aws_durable_execution_sdk_python_testing.model import StartDurableExecutionInput from aws_durable_execution_sdk_python_testing.stores.memory import ( InMemoryExecutionStore, @@ -583,6 +590,129 @@ def test_exhausted_retries_fail_the_execution(): assert final.close_status.value == "FAILED" +def _invoke_response( + status: InvocationStatus, result: str | None = None +) -> InvokeResponse: + return InvokeResponse( + invocation_output=DurableExecutionInvocationOutput( + status=status, result=result + ), + request_id="req", + ) + + +def test_pending_without_pending_operations_fails_after_three_attempts(): + """PENDING with nothing pending is the one invalid output the service + retries, because an SDK race that a replay resolves can produce it. + It allows three consecutive attempts, then fails the execution with + an InvalidParameterValueException error.""" + store = InMemoryExecutionStore() + scheduler = Mock() + invoker = Mock() + invoker.create_invocation_input.return_value = Mock() + invoker.invoke.return_value = _invoke_response(InvocationStatus.PENDING) + + execution = Execution.new(_make_start_input()) + execution.start() + store.save(execution) + arn = execution.durable_execution_arn + + executor = Executor(store, scheduler, invoker, Mock()) + + for attempt in (1, 2): + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + retrying = store.load(arn) + assert retrying.is_complete is False + assert retrying.consecutive_failed_invocation_attempts == attempt + assert scheduler.call_later.call_count == attempt + assert ( + scheduler.call_later.call_args.kwargs["delay"] + == Executor.RETRY_BACKOFF_SECONDS + ) + + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + + final = store.load(arn) + assert invoker.invoke.call_count == 3 + assert scheduler.call_later.call_count == 2 + assert final.close_status is ExecutionStatus.FAILED + assert final.result is not None + assert final.result.error == ErrorObject( + message="Cannot return PENDING status with no pending operations.", + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) + + +def test_pending_without_pending_operations_recovers_on_a_valid_output(): + """A retried PENDING followed by a valid output completes the + execution and resets the attempt counter.""" + store = InMemoryExecutionStore() + invoker = Mock() + invoker.create_invocation_input.return_value = Mock() + invoker.invoke.side_effect = [ + _invoke_response(InvocationStatus.PENDING), + _invoke_response(InvocationStatus.SUCCEEDED, result="ok"), + ] + + execution = Execution.new(_make_start_input()) + execution.start() + store.save(execution) + arn = execution.durable_execution_arn + + executor = Executor(store, Mock(), invoker, Mock()) + + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + assert store.load(arn).consecutive_failed_invocation_attempts == 1 + + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + final = store.load(arn) + assert final.close_status is ExecutionStatus.SUCCEEDED + assert final.result is not None + assert final.result.result == "ok" + assert final.consecutive_failed_invocation_attempts == 0 + + +@pytest.mark.parametrize( + "parse", + [ + pytest.param( + lambda request_id: InvokeResponse.from_output(["SUCCEEDED"], request_id), + id="from-output", + ), + pytest.param( + lambda request_id: InvokeResponse.from_payload(b"not json", request_id), + id="from-payload", + ), + pytest.param( + lambda request_id: InvokeResponse.from_payload(b"\xff", request_id), + id="from-payload-not-utf-8", + ), + ], +) +def test_unparsed_output_keeps_the_request_id(parse): + """An output the invoker cannot parse fails the execution, and the + recorded invocation keeps the request ID of that invocation.""" + store = InMemoryExecutionStore() + invoker = Mock() + invoker.create_invocation_input.return_value = Mock() + invoker.invoke.side_effect = lambda *_args, **_kwargs: parse("request-1") + + execution = Execution.new(_make_start_input()) + execution.start() + store.save(execution) + arn = execution.durable_execution_arn + + executor = Executor(store, Mock(), invoker, Mock()) + + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + + final = store.load(arn) + assert final.close_status is ExecutionStatus.FAILED + assert [c.request_id for c in final.invocation_completions] == ["request-1"] + + # endregion # region: earliest pending 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 bb6b71f4f..41dab117d 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 @@ -181,6 +181,16 @@ def mock_execution(): return execution +def _invalid_output_error(message: str) -> ErrorObject: + """The error the service fails an execution with for an invalid output.""" + return ErrorObject( + message=message, + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) + + def test_init(mock_store, mock_scheduler, mock_invoker, mock_checkpoint_processor): # Test that Executor can be constructed with dependencies # Dependency injection is implementation detail - test behavior instead @@ -504,10 +514,10 @@ def test_should_ignore_response_when_execution_already_complete( mock_invoker.invoke.assert_not_called() -def test_should_retry_when_response_has_no_status( +def test_should_fail_when_response_has_no_status( executor, mock_store, mock_scheduler, mock_invoker, start_input ): - """Test that invocation responses without status trigger retry logic.""" + """Test that invocation responses without status fail the execution without a retry.""" # Arrange mock_execution = Mock() mock_execution.durable_execution_arn = "test-arn" @@ -530,27 +540,30 @@ def test_should_retry_when_response_has_no_status( mock_execution_class.new.return_value = mock_execution mock_store.load.return_value = mock_execution - # Act - trigger invocation through public start_execution method - executor.start_execution(start_input) + with patch.object(executor, "fail_execution") as mock_fail: + # Act - trigger invocation through public start_execution method + executor.start_execution(start_input) - # Get the handler that was passed to the scheduler and execute it manually - assert mock_scheduler.call_later.call_count >= 1 - handler = mock_scheduler.call_later.call_args_list[-1][0][0] + # Get the handler that was passed to the scheduler and execute it manually + assert mock_scheduler.call_later.call_count >= 1 + handler = mock_scheduler.call_later.call_args_list[-1][0][0] - # Execute the handler to trigger the invocation logic - asyncio.run(handler()) + # Execute the handler to trigger the invocation logic + asyncio.run(handler()) - # Assert - verify retry was triggered due to validation error - assert mock_execution.consecutive_failed_invocation_attempts == 1 - mock_store.save.assert_called_with(mock_execution) - # Verify retry was scheduled (call_later should be called 3 times: timeout + initial + retry) - assert mock_scheduler.call_later.call_count == 3 + # Assert - verify the execution failed at once, as the service does + mock_fail.assert_called_once_with( + "test-arn", _invalid_output_error("Response status is required") + ) + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # Verify no retry was scheduled (call_later is called 2 times: timeout + initial) + assert mock_scheduler.call_later.call_count == 2 -def test_should_retry_when_failed_response_has_result( +def test_should_fail_when_failed_response_has_result( executor, mock_store, mock_scheduler, mock_invoker, start_input ): - """Test that failed responses with result trigger retry logic.""" + """Test that failed responses with result fail the execution without a retry.""" # Arrange mock_execution = Mock() mock_execution.durable_execution_arn = "test-arn" @@ -575,27 +588,31 @@ def test_should_retry_when_failed_response_has_result( mock_execution_class.new.return_value = mock_execution mock_store.load.return_value = mock_execution - # Act - trigger invocation through public start_execution method - executor.start_execution(start_input) + with patch.object(executor, "fail_execution") as mock_fail: + # Act - trigger invocation through public start_execution method + executor.start_execution(start_input) - # Get the handler that was passed to the scheduler and execute it manually - assert mock_scheduler.call_later.call_count >= 1 - handler = mock_scheduler.call_later.call_args_list[-1][0][0] + # Get the handler that was passed to the scheduler and execute it manually + assert mock_scheduler.call_later.call_count >= 1 + handler = mock_scheduler.call_later.call_args_list[-1][0][0] - # Execute the handler to trigger the invocation logic - asyncio.run(handler()) + # Execute the handler to trigger the invocation logic + asyncio.run(handler()) - # Assert - verify retry was triggered due to validation error - assert mock_execution.consecutive_failed_invocation_attempts == 1 - mock_store.save.assert_called_with(mock_execution) - # Verify retry was scheduled (call_later should be called 3 times: timeout + initial + retry) - assert mock_scheduler.call_later.call_count == 3 + # Assert - verify the execution failed at once, as the service does + mock_fail.assert_called_once_with( + "test-arn", + _invalid_output_error("Cannot provide a Result for FAILED status."), + ) + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # Verify no retry was scheduled (call_later is called 2 times: timeout + initial) + assert mock_scheduler.call_later.call_count == 2 -def test_should_retry_when_success_response_has_error( +def test_should_fail_when_success_response_has_error( executor, mock_store, mock_scheduler, mock_invoker, start_input ): - """Test that successful responses with error trigger retry logic.""" + """Test that successful responses with error fail the execution without a retry.""" # Arrange mock_execution = Mock() mock_execution.durable_execution_arn = "test-arn" @@ -621,21 +638,25 @@ def test_should_retry_when_success_response_has_error( mock_execution_class.new.return_value = mock_execution mock_store.load.return_value = mock_execution - # Act - trigger invocation through public start_execution method - executor.start_execution(start_input) + with patch.object(executor, "fail_execution") as mock_fail: + # Act - trigger invocation through public start_execution method + executor.start_execution(start_input) - # Get the handler that was passed to the scheduler and execute it manually - assert mock_scheduler.call_later.call_count >= 1 - handler = mock_scheduler.call_later.call_args_list[-1][0][0] + # Get the handler that was passed to the scheduler and execute it manually + assert mock_scheduler.call_later.call_count >= 1 + handler = mock_scheduler.call_later.call_args_list[-1][0][0] - # Execute the handler to trigger the invocation logic - asyncio.run(handler()) + # Execute the handler to trigger the invocation logic + asyncio.run(handler()) - # Assert - verify retry was triggered due to validation error - assert mock_execution.consecutive_failed_invocation_attempts == 1 - mock_store.save.assert_called_with(mock_execution) - # Verify retry was scheduled (call_later should be called 3 times: timeout + initial + retry) - assert mock_scheduler.call_later.call_count == 3 + # Assert - verify the execution failed at once, as the service does + mock_fail.assert_called_once_with( + "test-arn", + _invalid_output_error("Cannot provide an Error for SUCCEEDED status."), + ) + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # Verify no retry was scheduled (call_later is called 2 times: timeout + initial) + assert mock_scheduler.call_later.call_count == 2 def test_should_retry_when_pending_response_has_no_operations( @@ -648,6 +669,7 @@ def test_should_retry_when_pending_response_has_no_operations( mock_execution.is_complete = False mock_execution.start_input = start_input mock_execution.consecutive_failed_invocation_attempts = 0 + mock_execution.seq_counter = 0 mock_execution.has_pending_operations.return_value = False # No pending operations mock_execution.handler_seen_seq = 0 mock_execution.has_changes_after.return_value = False @@ -1236,7 +1258,12 @@ def test_should_retry_invocation_when_under_limit_through_public_api( mock_execution.durable_execution_arn = "test-arn" mock_execution.is_complete = False mock_execution.start_input = start_input - mock_execution.consecutive_failed_invocation_attempts = 3 # Under limit (5 is max) + # Under limit (3 is max for PENDING with no pending operations) + mock_execution.consecutive_failed_invocation_attempts = 1 + mock_execution.seq_counter = 0 + mock_execution.has_pending_operations.return_value = False + mock_execution.handler_seen_seq = 0 + mock_execution.has_changes_after.return_value = False # Configure invoker to fail initially with validation error, then succeed on retry mock_invocation_input = Mock() @@ -1244,8 +1271,7 @@ def test_should_retry_invocation_when_under_limit_through_public_api( # First invocation: invalid response triggers retry invalid_response = DurableExecutionInvocationOutput( - status=InvocationStatus.FAILED, - result="should not have result", # Invalid: failed response with result + status=InvocationStatus.PENDING, # Invalid: PENDING with nothing pending ) # Second invocation: valid success response success_response = DurableExecutionInvocationOutput( @@ -1289,8 +1315,8 @@ def test_should_retry_invocation_when_under_limit_through_public_api( # Assert - verify final outcome after retry sequence # Counter resets to 0 on a clean invocation. The first - # attempt failed validation (3 -> 4), the second attempt succeeded - # (4 -> 0). Previously this was 4 because the counter only ever + # attempt failed validation (1 -> 2), the second attempt succeeded + # (2 -> 0). Previously this was 2 because the counter only ever # grew; that was a bug. assert mock_execution.consecutive_failed_invocation_attempts == 0 assert retry_delay == Executor.RETRY_BACKOFF_SECONDS # Correct backoff delay used @@ -1522,10 +1548,10 @@ def test_on_failed(executor): mock_fail.assert_called_once_with("test-arn", error) -def test_should_retry_when_response_has_unexpected_status( +def test_should_fail_when_response_has_unexpected_status( executor, mock_store, mock_scheduler, mock_invoker, start_input ): - """Test that responses with unexpected status trigger retry logic.""" + """Test that responses with unexpected status fail the execution without a retry.""" # Arrange mock_execution = Mock() mock_execution.durable_execution_arn = "test-arn" @@ -1549,21 +1575,113 @@ def test_should_retry_when_response_has_unexpected_status( mock_execution_class.new.return_value = mock_execution mock_store.load.return_value = mock_execution - # Act - trigger invocation through public start_execution method - executor.start_execution(start_input) + with patch.object(executor, "fail_execution") as mock_fail: + # Act - trigger invocation through public start_execution method + executor.start_execution(start_input) - # Get the handler that was passed to the scheduler and execute it manually - assert mock_scheduler.call_later.call_count >= 1 - handler = mock_scheduler.call_later.call_args_list[-1][0][0] + # Get the handler that was passed to the scheduler and execute it manually + assert mock_scheduler.call_later.call_count >= 1 + handler = mock_scheduler.call_later.call_args_list[-1][0][0] - # Execute the handler to trigger the invocation logic - asyncio.run(handler()) + # Execute the handler to trigger the invocation logic + asyncio.run(handler()) - # Assert - verify retry was triggered due to validation error - assert mock_execution.consecutive_failed_invocation_attempts == 1 - mock_store.save.assert_called_with(mock_execution) - # Verify retry was scheduled (call_later should be called 3 times: timeout + initial + retry) - assert mock_scheduler.call_later.call_count == 3 + # Assert - verify the execution failed at once, as the service does + mock_fail.assert_called_once_with( + "test-arn", + _invalid_output_error("Unexpected invocation status: UNKNOWN_STATUS"), + ) + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # Verify no retry was scheduled (call_later is called 2 times: timeout + initial) + assert mock_scheduler.call_later.call_count == 2 + + +def test_invoke_handler_fails_execution_when_output_cannot_be_parsed( + executor, mock_store, mock_scheduler, mock_invoker, start_input +): + """An output the invoker cannot parse fails the execution at once, + without a retry, and the invocation is still recorded.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + mock_execution = Mock() + mock_execution.durable_execution_arn = "test-arn" + mock_execution.is_complete = False + mock_execution.start_input = start_input + mock_execution.consecutive_failed_invocation_attempts = 0 + + mock_invoker.create_invocation_input.return_value = Mock() + mock_invoker.invoke.side_effect = InvalidInvocationOutputError( + "Invalid invocation output: Expecting value: line 1 column 1 (char 0)" + ) + + with patch( + "aws_durable_execution_sdk_python_testing.executor.Execution" + ) as mock_execution_class: + mock_execution_class.new.return_value = mock_execution + mock_store.load.return_value = mock_execution + + with patch.object(executor, "fail_execution") as mock_fail: + executor.start_execution(start_input) + handler = mock_scheduler.call_later.call_args_list[-1][0][0] + asyncio.run(handler()) + + mock_fail.assert_called_once_with( + "test-arn", + _invalid_output_error( + "Invalid invocation output: Expecting value: line 1 column 1 (char 0)" + ), + ) + mock_execution.record_invocation_completion.assert_called_once() + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # timeout + initial invocation, no retry + assert mock_scheduler.call_later.call_count == 2 + + +def test_invoke_handler_unparsed_output_leaves_a_stop_during_invocation( + executor, mock_store, mock_scheduler, mock_invoker, start_input +): + """A stop that lands while the handler runs stands when its output + cannot be parsed: the execution is not failed over it.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + incomplete_execution = Mock(spec=Execution) + incomplete_execution.is_complete = False + incomplete_execution.start_input = start_input + incomplete_execution.consecutive_failed_invocation_attempts = 0 + incomplete_execution.durable_execution_arn = "test-arn" + incomplete_execution.seq_counter = 0 + + stopped_execution = Mock(spec=Execution) + stopped_execution.is_complete = True + + # Never let the side_effect iterator exhaust: see + # test_invoke_handler_execution_completed_during_invocation_async. + _loads = iter([incomplete_execution]) + mock_store.load.side_effect = lambda *a, **k: next(_loads, stopped_execution) + + mock_invoker.create_invocation_input.return_value = Mock() + mock_invoker.invoke.side_effect = InvalidInvocationOutputError( + "Invalid invocation output: 'str' object has no attribute 'get'" + ) + + with patch( + "aws_durable_execution_sdk_python_testing.executor.Execution" + ) as mock_execution_class: + mock_execution_class.new.return_value = incomplete_execution + + with patch.object(executor, "fail_execution") as mock_fail: + executor.start_execution(start_input) + handler = mock_scheduler.call_later.call_args_list[-1][0][0] + asyncio.run(handler()) + + mock_fail.assert_not_called() + stopped_execution.record_invocation_completion.assert_called_once() + # No retry is scheduled either: timeout + initial invocation only. + assert mock_scheduler.call_later.call_count == 2 def test_invoke_handler_execution_completed_during_invocation_async( diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py index 14e6f2344..0286c1c00 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py @@ -15,6 +15,7 @@ ) from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, ExecutionDetails, Operation, OperationStatus, @@ -179,6 +180,20 @@ def handler(event, context): # noqa: ARG001 assert seen[0][1].endswith(":function:child:prod") +def _response_metadata(request_id: str) -> dict: + """ResponseMetadata as boto3 returns it: the request ID, and the + response headers with lowercased names.""" + return { + "RequestId": request_id, + "HTTPStatusCode": 200, + "HTTPHeaders": { + "content-type": "application/json", + "x-amzn-requestid": request_id, + }, + "RetryAttempts": 0, + } + + def _stub_success_client() -> Mock: client = Mock() payload = Mock() @@ -271,6 +286,93 @@ def test_in_process_invoker_invoke(): assert isinstance(call_args[1], LambdaContext) +@pytest.mark.parametrize( + ("output", "message"), + [ + pytest.param( + {"Result": "result"}, + "Invalid invocation output: None is not a valid InvocationStatus", + id="missing-status", + ), + pytest.param( + {"Status": "DONE"}, + "Invalid invocation output: 'DONE' is not a valid InvocationStatus", + id="unknown-status", + ), + pytest.param( + {"Status": "FAILED", "Error": "boom"}, + "Invalid invocation output: 'str' object has no attribute 'get'", + id="error-not-an-object", + ), + pytest.param( + {"Status": "SUCCEEDED", "Error": []}, + "Invalid invocation output: 'list' object has no attribute 'get'", + id="error-empty-list", + ), + pytest.param( + {"Status": "SUCCEEDED", "Error": 0}, + "Invalid invocation output: 'int' object has no attribute 'get'", + id="error-zero", + ), + pytest.param( + {"Status": "SUCCEEDED", "Error": False}, + "Invalid invocation output: 'bool' object has no attribute 'get'", + id="error-false", + ), + pytest.param( + "SUCCEEDED", + "Invalid invocation output: 'str' object has no attribute 'get'", + id="not-an-object", + ), + pytest.param( + None, + "Invalid invocation output: 'NoneType' object has no attribute 'get'", + id="none", + ), + ], +) +def test_in_process_invoker_rejects_an_output_it_cannot_parse(output, message): + """An output the service cannot parse raises InvalidInvocationOutputError, + which the executor fails the execution on instead of retrying.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + invoker = InProcessInvoker(Mock(return_value=output), Mock()) + input_data = DurableExecutionInvocationInput( + durable_execution_arn="test-arn", + checkpoint_token="test-token", # noqa: S106 + initial_execution_state=InitialExecutionState(operations=[], next_marker=""), + ) + + with pytest.raises(InvalidInvocationOutputError) as exc_info: + invoker.invoke("test-function", input_data) + + assert str(exc_info.value) == message + assert exc_info.value.request_id == "test-invoke-12345" + + +@pytest.mark.parametrize( + ("error", "expected"), + [ + pytest.param( + {}, ErrorObject(None, None, None, None), id="empty-object-is-an-error" + ), + pytest.param(None, None, id="null-is-no-error"), + ], +) +def test_in_process_invoker_keeps_an_empty_error_object(error, expected): + """An empty Error object is still an Error, so the executor rejects it on + SUCCEEDED; only a missing or null Error means there is none.""" + invoker = InProcessInvoker( + Mock(return_value={"Status": "SUCCEEDED", "Error": error}), Mock() + ) + + response = invoker.invoke("test-function", _minimal_input()) + + assert response.invocation_output.error == expected + + def test_lambda_invoker_init(): """Test LambdaInvoker initialization.""" lambda_client = Mock() @@ -490,7 +592,7 @@ def new_client(*_a, **_k): client.invoke.return_value = { "StatusCode": 200, "Payload": payload, - "ResponseMetadata": {"HTTPHeaders": {"x-amzn-RequestId": "r"}}, + "ResponseMetadata": _response_metadata("r"), } return client @@ -612,7 +714,7 @@ def test_lambda_invoker_invoke_success(): lambda_client.invoke.return_value = { "StatusCode": 200, "Payload": mock_payload, - "ResponseMetadata": {"HTTPHeaders": {"x-amzn-RequestId": "test-request-id"}}, + "ResponseMetadata": _response_metadata("test-request-id"), } invoker = LambdaInvoker(lambda_client) @@ -790,9 +892,7 @@ def test_lambda_invoker_invoke_status_202(): lambda_client.invoke.return_value = { "StatusCode": 202, "Payload": mock_payload, - "ResponseMetadata": { - "HTTPHeaders": {"x-amzn-RequestId": "test-request-id-202"} - }, + "ResponseMetadata": _response_metadata("test-request-id-202"), } invoker = LambdaInvoker(lambda_client) @@ -839,6 +939,156 @@ def test_lambda_invoker_invoke_function_error(): invoker.invoke("test-function", input_data) +@pytest.mark.parametrize( + ("payload", "message"), + [ + pytest.param( + b"not json", + "Invalid invocation output: Expecting value: line 1 column 1 (char 0)", + id="not-json", + ), + pytest.param( + b'{"Status": "DONE"}', + "Invalid invocation output: 'DONE' is not a valid InvocationStatus", + id="unknown-status", + ), + pytest.param( + b'["SUCCEEDED"]', + "Invalid invocation output: 'list' object has no attribute 'get'", + id="not-an-object", + ), + pytest.param( + b"\xff", + "Invalid invocation output: 'utf-8' codec can't decode byte 0xff" + " in position 0: invalid start byte", + id="not-utf-8", + ), + ], +) +def test_lambda_invoker_rejects_an_output_it_cannot_parse(payload, message): + """A payload the service cannot parse raises InvalidInvocationOutputError, + not the DurableFunctionsTestError of a failed invocation.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + lambda_client, _ = _create_mock_lambda_client_with_exceptions() + mock_payload = Mock() + mock_payload.read.return_value = payload + lambda_client.invoke.return_value = { + "StatusCode": 200, + "Payload": mock_payload, + "ResponseMetadata": _response_metadata("test-request-id"), + } + + invoker = LambdaInvoker(lambda_client) + input_data = DurableExecutionInvocationInput( + durable_execution_arn="test-arn", + checkpoint_token="test-token", # noqa: S106 + initial_execution_state=InitialExecutionState(operations=[], next_marker=""), + ) + + with pytest.raises(InvalidInvocationOutputError) as exc_info: + invoker.invoke("test-function", input_data) + + assert str(exc_info.value) == message + assert exc_info.value.request_id == "test-request-id" + + +@pytest.mark.parametrize( + ("metadata", "request_id"), + [ + pytest.param( + { + "RequestId": "from-metadata", + "HTTPHeaders": {"x-amzn-requestid": "from-header"}, + }, + "from-metadata", + id="response-metadata", + ), + pytest.param( + {"HTTPHeaders": {"x-amzn-requestid": "from-header"}}, + "from-header", + id="lowercase-header", + ), + pytest.param( + {"HTTPHeaders": {"X-Amzn-RequestId": "from-header"}}, + "from-header", + id="mixed-case-header", + ), + pytest.param( + {"HTTPHeaders": {"x-amzn-request-id": "from-header"}}, + "from-header", + id="request-id-header", + ), + ], +) +def test_lambda_invoker_reads_the_request_id_from_the_response(metadata, request_id): + """The request ID comes from ResponseMetadata, then from the response + headers, whatever the case of their names.""" + client = _stub_success_client() + client.invoke.return_value["ResponseMetadata"] = metadata + + response = LambdaInvoker(client).invoke("test-function", _minimal_input()) + + assert response.request_id == request_id + + +def _botocore_answers_invoke(monkeypatch, body: bytes) -> LambdaInvoker: + """A LambdaInvoker on a real client whose Invoke returns ``body``. + + The response, with header ``x-amzn-RequestId: request-1``, is + returned at ``before-send``, so botocore parses it as it would one + from the endpoint. + """ + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "test") + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "test") + client = create_lambda_client("http://localhost:3001", "us-west-2") + + def answer(request, **_kwargs): + return AWSResponse( + request.url, + 200, + {"Content-Type": "application/json", "x-amzn-RequestId": "request-1"}, + _RawBody(body), + ) + + client.meta.events.register("before-send.lambda.Invoke", answer) + return LambdaInvoker(client) + + +def test_lambda_invoker_keeps_the_request_id_botocore_parses(monkeypatch): + """The response carries the request ID the endpoint sent, not a generated one.""" + invoker = _botocore_answers_invoke(monkeypatch, b'{"Status": "SUCCEEDED"}') + + response = invoker.invoke("test-function", _minimal_input()) + + assert response.request_id == "request-1" + + +@pytest.mark.parametrize( + "body", + [ + pytest.param(b"not json", id="not-json"), + pytest.param(b'{"Status": "DONE"}', id="unknown-status"), + pytest.param(b"\xff", id="not-utf-8"), + ], +) +def test_lambda_invoker_rejects_with_the_request_id_botocore_parses(monkeypatch, body): + """An output that cannot be parsed, undecodable bytes included, raises + InvalidInvocationOutputError with the request ID the endpoint sent.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + invoker = _botocore_answers_invoke(monkeypatch, body) + + with pytest.raises(InvalidInvocationOutputError) as exc_info: + invoker.invoke("test-function", _minimal_input()) + + assert exc_info.value.request_id == "request-1" + + def _create_mock_lambda_client_with_exceptions(): """Helper to create mock lambda client with all exception types.""" lambda_client = Mock()