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 da976f5b..a4f54b53 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 @@ -344,9 +344,21 @@ def get_navigable_operations(self) -> list[Operation]: return list(self.operations) def get_assertable_operations(self) -> list[Operation]: - """Get list of operations, but exclude the EXECUTION operations""" - # TODO: this excludes EXECUTION at start, but can there be an EXECUTION at the end if there was a checkpoint with large payload? - return self.operations[1:] + """Return a snapshot of operations, excluding EXECUTION operations. + + Filters by ``operation_type`` rather than assuming the EXECUTION + operation is always the first entry, so a second EXECUTION-type + entry (e.g. a future large-payload checkpoint recorded at the end + of the list) is excluded too. Snapshot taken under ``_state_lock``, + matching ``get_navigable_operations``, so a concurrent checkpoint + can't produce a torn read. + """ + with self._state_lock: + return [ + operation + for operation in self.operations + if operation.operation_type != OperationType.EXECUTION + ] def has_pending_operations(self, execution: Execution) -> bool: """True if execution has pending operations.""" 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 bbb2d0d1..7002d799 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 @@ -232,6 +232,80 @@ def test_get_assertable_operations(): assert result[0] == step_op +def test_get_assertable_operations_excludes_trailing_execution_operation(): + """A second EXECUTION-type entry at the end of the list (e.g. a future + large-payload checkpoint) must be excluded too, not just the one at + index 0.""" + 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_op = Operation( + operation_id="exec-op", + parent_id=None, + name="execution", + start_timestamp=datetime.now(timezone.utc), + operation_type=OperationType.EXECUTION, + status=OperationStatus.STARTED, + ) + step_op = Operation( + operation_id="step-op", + parent_id=None, + name="step", + start_timestamp=datetime.now(timezone.utc), + operation_type=OperationType.STEP, + status=OperationStatus.STARTED, + ) + trailing_execution_op = Operation( + operation_id="exec-op-end", + parent_id=None, + name="execution", + start_timestamp=datetime.now(timezone.utc), + operation_type=OperationType.EXECUTION, + status=OperationStatus.SUCCEEDED, + ) + operations = [execution_op, step_op, trailing_execution_op] + execution = Execution("test-arn", start_input, operations) + + result = execution.get_assertable_operations() + + assert result == [step_op] + + +def test_get_assertable_operations_returns_independent_snapshot(): + """The returned list must not alias `self.operations`, so later + mutation of the execution's operations doesn't retroactively change + a snapshot a caller already holds.""" + 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_op = Operation( + operation_id="exec-op", + parent_id=None, + name="execution", + start_timestamp=datetime.now(timezone.utc), + operation_type=OperationType.EXECUTION, + status=OperationStatus.STARTED, + ) + execution = Execution("test-arn", start_input, [execution_op]) + + result = execution.get_assertable_operations() + execution.operations.append(execution_op) + + assert result == [] + + def test_has_pending_operations_with_pending_step(): """Test has_pending_operations returns True for pending STEP operations.""" start_input = StartDurableExecutionInput(