Skip to content
Open
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 @@ -3,6 +3,7 @@
from __future__ import annotations

import json
import re
from typing import TYPE_CHECKING, ClassVar

from aws_durable_execution_sdk_python.lambda_service import (
Expand Down Expand Up @@ -43,6 +44,7 @@
)
from aws_durable_execution_sdk_python_testing.exceptions import (
InvalidParameterValueException,
ValidationException,
)


Expand All @@ -52,6 +54,9 @@
from aws_durable_execution_sdk_python_testing.execution import Execution

MAX_ERROR_PAYLOAD_SIZE_BYTES = 1048576
SUBTYPE_PATTERN = "[a-zA-Z0-9-_]+"
_SUBTYPE_RE = re.compile(SUBTYPE_PATTERN)
_MAX_VALIDATION_ERRORS = 10


class CheckpointValidator:
Expand All @@ -74,12 +79,51 @@ def validate_input(updates: list[OperationUpdate], execution: Execution) -> None
if not updates:
return

# Reject service-model violations before any stateful validation or writes.
CheckpointValidator._validate_subtypes(updates)
CheckpointValidator._validate_conflicting_execution_update(updates)
CheckpointValidator._validate_parent_id_and_duplicate_id(updates, execution)

for update in updates:
CheckpointValidator._validate_operation_update(update, execution)

@staticmethod
def _validate_subtypes(updates: list[OperationUpdate]) -> None:
"""Validate SubType constraints before checkpoint state is mutated.

The service lists up to ten errors in update order. The empty string
violates both the minimum length and the required pattern.
"""
errors: list[str] = []
for index, update in enumerate(updates, start=1):
sub_type = update.sub_type
if sub_type is None:
continue
prefix = (
f"Value '{sub_type}' at 'updates.{index}.member.subType' "
"failed to satisfy constraint:"
)
if len(sub_type) < 1:
errors.append(
f"{prefix} Member must have length greater than or equal to 1"
)
if len(sub_type) > 32:
errors.append(
f"{prefix} Member must have length less than or equal to 32"
)
if _SUBTYPE_RE.fullmatch(sub_type) is None:
errors.append(
f"{prefix} Member must satisfy regular expression pattern: {SUBTYPE_PATTERN}"
)

if errors:
reported = errors[:_MAX_VALIDATION_ERRORS]
count = len(reported)
noun = "error" if count == 1 else "errors"
raise ValidationException(
f"{count} validation {noun} detected: {'; '.join(reported)}"
)

@staticmethod
def _validate_conflicting_execution_update(updates: list[OperationUpdate]) -> None:
"""Validate that there are no conflicting execution updates."""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

import datetime

from aws_durable_execution_sdk_python.exceptions import (
CheckpointError,
DurableApiErrorCategory,
)
from aws_durable_execution_sdk_python.lambda_service import (
CheckpointOutput,
DurableServiceClient,
Expand All @@ -10,6 +14,7 @@
)

from aws_durable_execution_sdk_python_testing.clock import real_now
from aws_durable_execution_sdk_python_testing.exceptions import ValidationException
from aws_durable_execution_sdk_python_testing.checkpoint.processor import (
CheckpointProcessor,
)
Expand Down Expand Up @@ -41,7 +46,13 @@ def checkpoint(
task = CheckpointTask(
self._checkpoint_processor, checkpoint_token, updates, client_token
)
return self._registry.submit(durable_execution_arn, task).result()
try:
return self._registry.submit(durable_execution_arn, task).result()
except ValidationException as exc:
# Match the boto client: service-model 400s terminate the execution.
raise CheckpointError(
str(exc), error_category=DurableApiErrorCategory.EXECUTION
) from exc

def get_execution_state(
self,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,20 @@ def to_dict(self) -> dict[str, Any]:
return {"Type": "InvalidParameterValueException", "message": self.message}


class ValidationException(AwsApiException):
"""Service-model constraint error returned by CheckpointDurableExecution."""

http_status_code = 400

def __init__(self, message: str) -> None:
self.message = message
super().__init__(message)

def to_dict(self) -> dict[str, Any]:
# Botocore recognizes the unmodeled ValidationException via __type.
return {"__type": "ValidationException", "message": self.message}


class ResourceNotFoundException(AwsApiException):
"""Exception for resource not found errors."""

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
"""Regression coverage for CheckpointDurableExecution SubType validation (#836)."""

from datetime import UTC, datetime
from unittest.mock import Mock

import pytest
from aws_durable_execution_sdk_python.lambda_service import (
OperationAction,
OperationType,
OperationUpdate,
)

from aws_durable_execution_sdk_python_testing.checkpoint.core import CheckpointCore
from aws_durable_execution_sdk_python_testing.checkpoint.validators.checkpoint import (
CheckpointValidator,
)
from aws_durable_execution_sdk_python_testing.exceptions import AwsApiException
from aws_durable_execution_sdk_python_testing.execution import Execution
from aws_durable_execution_sdk_python_testing.model import StartDurableExecutionInput
from aws_durable_execution_sdk_python_testing.web.models import HTTPResponse


PATTERN = "[a-zA-Z0-9-_]+"


def _execution() -> Execution:
execution = Execution.new(
StartDurableExecutionInput(
account_id="123456789012",
function_name="test-function",
function_qualifier="$LATEST",
execution_name="subtype-validation-test",
execution_timeout_seconds=900,
execution_retention_period_days=7,
input="{}",
invocation_id="test-invocation-id",
)
)
execution.start()
return execution


def _update(sub_type: str | None, operation_id: str = "step-1") -> OperationUpdate:
return OperationUpdate(
operation_id=operation_id,
operation_type=OperationType.STEP,
action=OperationAction.START,
sub_type=sub_type,
)


@pytest.mark.parametrize("sub_type", ["a", "x" * 32, "Order_Charge-21"])
def test_accepted_subtypes_at_boundaries(sub_type: str) -> None:
CheckpointValidator.validate_input([_update(sub_type)], _execution())


def test_missing_optional_subtype_is_allowed() -> None:
CheckpointValidator.validate_input([_update(None)], _execution())


@pytest.mark.parametrize(
("sub_type", "constraints"),
[
pytest.param(
"",
(
"Member must have length greater than or equal to 1",
f"Member must satisfy regular expression pattern: {PATTERN}",
),
id="empty-has-two-errors",
),
pytest.param(
"x" * 33,
("Member must have length less than or equal to 32",),
id="over-max-length",
),
pytest.param(
"order charge",
(f"Member must satisfy regular expression pattern: {PATTERN}",),
id="spaces",
),
pytest.param(
"ücret",
(f"Member must satisfy regular expression pattern: {PATTERN}",),
id="non-ascii",
),
],
)
def test_invalid_subtype_is_rejected_with_service_errors(
sub_type: str, constraints: tuple[str, ...]
) -> None:
with pytest.raises(AwsApiException) as exc_info:
CheckpointValidator.validate_input([_update(sub_type)], _execution())
message = str(exc_info.value)
assert message.startswith(f"{len(constraints)} validation error")
for constraint in constraints:
assert (
f"Value '{sub_type}' at 'updates.1.member.subType' "
f"failed to satisfy constraint: {constraint}"
) in message
response = HTTPResponse.create_error_from_exception(exc_info.value)
assert response.status_code == 400
assert response.body == {"__type": "ValidationException", "message": message}


def test_multiple_invalid_updates_report_in_input_order() -> None:
updates = [
_update("order charge", operation_id="step-1"),
_update("x" * 33, operation_id="step-2"),
]
with pytest.raises(AwsApiException) as exc_info:
CheckpointValidator.validate_input(updates, _execution())
message = str(exc_info.value)
assert message.startswith("2 validation errors detected:")
assert message.index("updates.1.member.subType") < message.index(
"updates.2.member.subType"
)


def test_rejects_whole_checkpoint_without_mutating_execution() -> None:
execution = _execution()
dispatcher = Mock()
operations_before = list(execution.operations)
checkpoint_before = execution.last_checkpoint
token_before = execution.token_sequence
updates = [_update("valid"), _update("invalid subtype", "step-2")]
with pytest.raises(AwsApiException):
CheckpointCore.apply(
execution=execution,
checkpoint_token="original-token",
updates=updates,
client_token="client-1",
dispatcher=dispatcher,
now=datetime(2026, 1, 1, tzinfo=UTC),
)
dispatcher.apply_updates.assert_not_called()
assert list(execution.operations) == operations_before
assert execution.last_checkpoint == checkpoint_before
assert execution.token_sequence == token_before


def test_model_validation_caps_reported_errors_at_ten() -> None:
updates = [_update("x" * 33, operation_id=f"step-{index}") for index in range(12)]
with pytest.raises(AwsApiException) as exc_info:
CheckpointValidator.validate_input(updates, _execution())
message = str(exc_info.value)
assert message.startswith("10 validation errors detected:")
assert "updates.10.member.subType" in message
assert "updates.11.member.subType" not in message
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
import datetime
from unittest.mock import Mock

import pytest
from aws_durable_execution_sdk_python.exceptions import CheckpointError
from aws_durable_execution_sdk_python.lambda_service import (
CheckpointOutput,
OperationAction,
Expand All @@ -12,6 +14,7 @@
)

from aws_durable_execution_sdk_python_testing.client import InMemoryServiceClient
from aws_durable_execution_sdk_python_testing.exceptions import ValidationException
from aws_durable_execution_sdk_python_testing.worker.registry import ExecutionRegistry


Expand Down Expand Up @@ -52,6 +55,20 @@ def test_checkpoint():
)


def test_checkpoint_validation_error_is_non_retryable():
"""Local model validation has the same terminal behavior as HTTP 400."""
processor = Mock()
processor.process_checkpoint.side_effect = ValidationException(
"Value 'bad subtype' at 'updates.1.member.subType' failed validation"
)
client = _client(processor)

with pytest.raises(CheckpointError, match="updates.1.member.subType") as exc_info:
client.checkpoint("execution-arn", "token", [], None)

assert not exc_info.value.is_retryable()


def test_get_execution_state():
"""Test get_execution_state method delegates to processor."""
processor = Mock()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
"""End-to-end parity: invalid SubType fails the durable execution (#836)."""

from typing import Any

import pytest
from aws_durable_execution_sdk_python.config import StepConfig
from aws_durable_execution_sdk_python.context import DurableContext, durable_step
from aws_durable_execution_sdk_python.execution import (
InvocationStatus,
durable_execution,
)
from aws_durable_execution_sdk_python.types import StepContext

from aws_durable_execution_sdk_python_testing.execution import ExecutionStatus
from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner


@pytest.mark.parametrize(
("sub_type", "constraint"),
[
("order charge", "Member must satisfy regular expression pattern"),
("x" * 33, "Member must have length less than or equal to 32"),
],
)
def test_invalid_step_subtype_fails_execution(sub_type: str, constraint: str) -> None:
reached_after_step: list[bool] = []

@durable_step
def charge(step_context: StepContext) -> str: # noqa: ARG001
return "ok"

# The SDK validates StepConfig at construction. Test-only mutation bypasses
# that client guard to exercise the independent local service constraint.
config = StepConfig(sub_type="ValidSubtype")
object.__setattr__(config, "sub_type", sub_type)

@durable_execution
def handler(event: Any, context: DurableContext) -> str: # noqa: ARG001
context.step(charge(), config=config)
reached_after_step.append(True)
return "done"

with DurableFunctionTestRunner(handler=handler, execution_timeout=10) as runner:
result = runner.run(input="{}")

assert result.status is InvocationStatus.FAILED
assert result.execution_status is ExecutionStatus.FAILED
assert result.error is not None
assert result.error.message is not None
assert ".member.subType" in result.error.message
assert constraint in result.error.message
assert not reached_after_step