Merged
Conversation
ghost
requested review from
ghost
changed the title
ghost
closed this
ghost
reopened this
ghost
changed the title
ghost
closed this
ghost
reopened this
ghost
commented
Nov 6, 2025
Comment on lines
+768
to
+804
| """Test map serializes items with item_serdes or fallback.""" | ||
| mock_serialize.return_value = '"serialized"' | ||
|
|
||
| parent_checkpoint = Mock() | ||
| parent_checkpoint.is_succeeded.return_value = False | ||
| parent_checkpoint.is_failed.return_value = False | ||
| parent_checkpoint.is_started.return_value = False | ||
| parent_checkpoint.is_existent.return_value = True | ||
| parent_checkpoint.is_replay_children.return_value = False | ||
|
|
||
| child_checkpoint = Mock() | ||
| child_checkpoint.is_succeeded.return_value = False | ||
| child_checkpoint.is_failed.return_value = False | ||
| child_checkpoint.is_started.return_value = False | ||
| child_checkpoint.is_existent.return_value = True | ||
| child_checkpoint.is_replay_children.return_value = False | ||
|
|
||
| def get_checkpoint(op_id): | ||
| return child_checkpoint if op_id.startswith("child-") else parent_checkpoint | ||
|
|
||
| mock_state = Mock() | ||
| mock_state.durable_execution_arn = "arn:test" | ||
| mock_state.get_checkpoint_result = Mock(side_effect=get_checkpoint) | ||
| mock_state.create_checkpoint = Mock() | ||
|
|
||
| context_map = {} | ||
|
|
||
| def create_id(self, i): | ||
| ctx_id = id(self) | ||
| if ctx_id not in context_map: | ||
| context_map[ctx_id] = [] | ||
| context_map[ctx_id].append(i) | ||
| return ( | ||
| "parent" | ||
| if len(context_map) == 1 and len(context_map[ctx_id]) == 1 | ||
| else f"child-{i}" | ||
| ) |
Author
There was a problem hiding this comment.
Setting up the state to allow us to properly reach children and execute them.
ghost
commented
Nov 6, 2025
Comment on lines
+815
to
+820
| assert mock_serialize.call_args_list[0][1]["serdes"] is expected | ||
| assert mock_serialize.call_args_list[0][1]["operation_id"] == "child-0" | ||
| assert mock_serialize.call_args_list[1][1]["serdes"] is expected | ||
| assert mock_serialize.call_args_list[1][1]["operation_id"] == "child-1" | ||
| assert mock_serialize.call_args_list[2][1]["serdes"] is batch_serdes | ||
| assert mock_serialize.call_args_list[2][1]["operation_id"] == "parent" |
Author
There was a problem hiding this comment.
Test that we are actually calling serialize.
Map and Parallel produce trees of operations. We expect to see N children first, and then 1 operation for the parent.
ghost
commented
Nov 6, 2025
Comment on lines
+833
to
+874
| """Test map deserializes items with item_serdes or fallback.""" | ||
| mock_deserialize.return_value = "deserialized" | ||
|
|
||
| parent_checkpoint = Mock() | ||
| parent_checkpoint.is_succeeded.return_value = False | ||
| parent_checkpoint.is_failed.return_value = False | ||
| parent_checkpoint.is_existent.return_value = False | ||
|
|
||
| child_checkpoint = Mock() | ||
| child_checkpoint.is_succeeded.return_value = True | ||
| child_checkpoint.is_failed.return_value = False | ||
| child_checkpoint.is_replay_children.return_value = False | ||
| child_checkpoint.result = '"cached"' | ||
|
|
||
| def get_checkpoint(op_id): | ||
| return child_checkpoint if op_id.startswith("child-") else parent_checkpoint | ||
|
|
||
| mock_state = Mock() | ||
| mock_state.durable_execution_arn = "arn:test" | ||
| mock_state.get_checkpoint_result = Mock(side_effect=get_checkpoint) | ||
| mock_state.create_checkpoint = Mock() | ||
|
|
||
| context_map = {} | ||
|
|
||
| def create_id(self, i): | ||
| ctx_id = id(self) | ||
| if ctx_id not in context_map: | ||
| context_map[ctx_id] = [] | ||
| context_map[ctx_id].append(i) | ||
| return ( | ||
| "parent" | ||
| if len(context_map) == 1 and len(context_map[ctx_id]) == 1 | ||
| else f"child-{i}" | ||
| ) | ||
|
|
||
| with patch.object(DurableContext, "_create_step_id_for_logical_step", create_id): | ||
| context = DurableContext(state=mock_state) | ||
| context.map( | ||
| ["a", "b"], | ||
| lambda ctx, item, idx, items: item, | ||
| config=MapConfig(serdes=batch_serdes, item_serdes=item_serdes), | ||
| ) |
ghost
commented
Nov 6, 2025
Comment on lines
+875
to
+880
|
|
||
| expected = item_serdes or batch_serdes | ||
| assert mock_deserialize.call_args_list[0][1]["serdes"] is expected | ||
| assert mock_deserialize.call_args_list[0][1]["operation_id"] == "child-0" | ||
| assert mock_deserialize.call_args_list[1][1]["serdes"] is expected | ||
| assert mock_deserialize.call_args_list[1][1]["operation_id"] == "child-1" |
Author
There was a problem hiding this comment.
Verify that we are calling deserialize on children.
ghost
commented
Nov 6, 2025
Comment on lines
+747
to
+792
| @patch("aws_durable_execution_sdk_python.operation.child.serialize") | ||
| def test_parallel_item_serialize(mock_serialize, item_serdes, batch_serdes): | ||
| """Test parallel serializes branches with item_serdes or fallback.""" | ||
| mock_serialize.return_value = '"serialized"' | ||
|
|
||
| parent_checkpoint = Mock() | ||
| parent_checkpoint.is_succeeded.return_value = False | ||
| parent_checkpoint.is_failed.return_value = False | ||
| parent_checkpoint.is_started.return_value = False | ||
| parent_checkpoint.is_existent.return_value = True | ||
| parent_checkpoint.is_replay_children.return_value = False | ||
|
|
||
| child_checkpoint = Mock() | ||
| child_checkpoint.is_succeeded.return_value = False | ||
| child_checkpoint.is_failed.return_value = False | ||
| child_checkpoint.is_started.return_value = False | ||
| child_checkpoint.is_existent.return_value = True | ||
| child_checkpoint.is_replay_children.return_value = False | ||
|
|
||
| def get_checkpoint(op_id): | ||
| return child_checkpoint if op_id.startswith("child-") else parent_checkpoint | ||
|
|
||
| mock_state = Mock() | ||
| mock_state.durable_execution_arn = "arn:test" | ||
| mock_state.get_checkpoint_result = Mock(side_effect=get_checkpoint) | ||
| mock_state.create_checkpoint = Mock() | ||
|
|
||
| context_map = {} | ||
|
|
||
| def create_id(self, i): | ||
| ctx_id = id(self) | ||
| if ctx_id not in context_map: | ||
| context_map[ctx_id] = [] | ||
| context_map[ctx_id].append(i) | ||
| return ( | ||
| "parent" | ||
| if len(context_map) == 1 and len(context_map[ctx_id]) == 1 | ||
| else f"child-{i}" | ||
| ) | ||
|
|
||
| with patch.object(DurableContext, "_create_step_id_for_logical_step", create_id): | ||
| context = DurableContext(state=mock_state) | ||
| context.parallel( | ||
| [lambda ctx: "a", lambda ctx: "b"], | ||
| config=ParallelConfig(serdes=batch_serdes, item_serdes=item_serdes), | ||
| ) |
ghost
commented
Nov 6, 2025
Comment on lines
+795
to
+800
| assert mock_serialize.call_args_list[0][1]["serdes"] is expected | ||
| assert mock_serialize.call_args_list[0][1]["operation_id"] == "child-0" | ||
| assert mock_serialize.call_args_list[1][1]["serdes"] is expected | ||
| assert mock_serialize.call_args_list[1][1]["operation_id"] == "child-1" | ||
| assert mock_serialize.call_args_list[2][1]["serdes"] is batch_serdes | ||
| assert mock_serialize.call_args_list[2][1]["operation_id"] == "parent" |
Author
There was a problem hiding this comment.
Verify that we are seeing children and parent in correct ordering here as we are executing recursively.
1. Pass item_serdes to executor factory methods. 2. Add tests to verify fallback and default behaviour.
yaythomas
approved these changes
Nov 6, 2025
This pull request was closed.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Issue #, if available:
Description of changes:
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.