Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -502,12 +502,18 @@ def handle(self, parsed_route: Route, request: HTTPRequest) -> HTTPResponse:
if include_execution_data_str
else False
)
reverse_order_str: str | None = self._parse_query_param(
request, "ReverseOrder"
)
reverse_order: bool = (
reverse_order_str == "true" if reverse_order_str else False
)

history_response: GetDurableExecutionHistoryResponse = (
self.executor.get_execution_history(
execution_arn,
include_execution_data=include_execution_data,
reverse_order=False,
reverse_order=reverse_order,
marker=marker,
max_items=int(max_items) if max_items else None,
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1377,6 +1377,53 @@ def test_get_durable_execution_history_handler_with_include_execution_data_false
)


@pytest.mark.parametrize(
("reverse_order_param", "expected_reverse_order"),
[
(None, False),
("false", False),
("true", True),
],
)
def test_get_durable_execution_history_handler_with_reverse_order(
reverse_order_param: str | None, expected_reverse_order: bool
):
"""Test GetDurableExecutionHistoryHandler with ReverseOrder."""

executor = Mock()
handler = GetDurableExecutionHistoryHandler(executor)
executor.get_execution_history.return_value = GetDurableExecutionHistoryResponse(
events=[], next_marker=None
)

router = Router()
typed_route = router.find_route(
"/2025-12-01/durable-executions/test-arn/history", "GET"
)
query_params: dict[str, list[str]] = {}
if reverse_order_param is not None:
query_params["ReverseOrder"] = [reverse_order_param]
request = HTTPRequest(
method="GET",
path=typed_route,
headers={},
query_params=query_params,
body={},
)

response = handler.handle(typed_route, request)

assert response.status_code == 200
assert response.body == {"Events": []}
executor.get_execution_history.assert_called_once_with(
"test-arn",
include_execution_data=False,
reverse_order=expected_reverse_order,
marker=None,
max_items=None,
)


def test_list_durable_executions_handler_success():
"""Test ListDurableExecutionsHandler with successful execution listing."""
executor = Mock()
Expand Down
Loading