feat(testing-sdk): add dual-mode integration testing infrastructure

Add comprehensive integration testing infrastructure that supports
both local (in-memory) and cloud (AWS Lambda) test execution modes
with a unified test interface.

Author: vip-amzn

.github/workflows/deploy-examples.yml

@@ -71,6 +71,7 @@ jobs:
       run: hatch run examples:build
 
     - name: Deploy Lambda function - ${{ matrix.example.name }}
+      id: deploy
       env:
         AWS_ACCOUNT_ID: ${{ secrets.AWS_ACCOUNT_ID }}
         LAMBDA_ENDPOINT: ${{ secrets.LAMBDA_ENDPOINT }}
@@ -88,13 +89,36 @@ jobs:
         echo "Deploying ${{ matrix.example.name }} as $FUNCTION_NAME"
         hatch run examples:deploy "${{ matrix.example.name }}" --function-name "$FUNCTION_NAME"
         
-        # $LATEST is also a qualified version        
-        QUALIFIED_FUNCTION_NAME="$FUNCTION_NAME:\$LATEST"
+        # $LATEST is also a qualified version - use single quotes to preserve literal string
+        QUALIFIED_FUNCTION_NAME="${FUNCTION_NAME}:"'$LATEST'
         
         # Store both names for later steps
         echo "FUNCTION_NAME=$FUNCTION_NAME" >> $GITHUB_ENV
         echo "QUALIFIED_FUNCTION_NAME=$QUALIFIED_FUNCTION_NAME" >> $GITHUB_ENV
         echo "VERSION=$VERSION" >> $GITHUB_ENV
+        echo "DEPLOYED_FUNCTION_NAME=$FUNCTION_NAME" >> $GITHUB_OUTPUT
+        echo "QUALIFIED_FUNCTION_NAME=$QUALIFIED_FUNCTION_NAME" >> $GITHUB_OUTPUT
+
+    - name: Run Integration Tests - ${{ matrix.example.name }}
+      env:
+        AWS_REGION: ${{ env.AWS_REGION }}
+        LAMBDA_ENDPOINT: ${{ secrets.LAMBDA_ENDPOINT }}
+        EXAMPLE_NAME: ${{ matrix.example.name }}
+        QUALIFIED_FUNCTION_NAME: ${{ steps.deploy.outputs.QUALIFIED_FUNCTION_NAME }}
+        PYTEST_FUNCTION_NAME_MAP: ${{ format('{{"{0}":"{1}"}}', matrix.example.name, steps.deploy.outputs.QUALIFIED_FUNCTION_NAME) }}
+      run: |
+        echo "Running integration tests for ${EXAMPLE_NAME}"
+        echo "Function name: ${{ steps.deploy.outputs.DEPLOYED_FUNCTION_NAME }}"
+        echo "Qualified function name: ${QUALIFIED_FUNCTION_NAME}"
+        echo "Function name map: ${PYTEST_FUNCTION_NAME_MAP}"
+        echo "AWS Region: ${AWS_REGION}"
+        echo "Lambda Endpoint: ${LAMBDA_ENDPOINT}"
+        
+        # Convert example name to test name: "Hello World" -> "test_hello_world"
+        TEST_NAME="test_$(echo "${EXAMPLE_NAME}" | tr '[:upper:]' '[:lower:]' | tr ' ' '_')"
+        echo "Test name: ${TEST_NAME}"
+        
+        hatch run test:examples-integration
 
     - name: Invoke Lambda function - ${{ matrix.example.name }}
       env:

examples/cli.py

@@ -380,7 +380,7 @@ def deploy_function(example_name: str, function_name: str | None = None):
 
     # Update invoke permission for worker account using put_resource_policy
     function_arn = f"arn:aws:lambda:{config['region']}:{config['account_id']}:function:{function_name}"
-    
+
     policy_document = {
         "Version": "2012-10-17",
         "Statement": [
@@ -389,14 +389,13 @@ def deploy_function(example_name: str, function_name: str | None = None):
                 "Effect": "Allow",
                 "Principal": {"AWS": config["invoke_account_id"]},
                 "Action": "lambda:InvokeFunction",
-                "Resource": f"{function_arn}:*"
+                "Resource": f"{function_arn}:*",
             }
-        ]
+        ],
     }
 
     lambda_client.put_resource_policy(
-        ResourceArn=function_arn,
-        Policy=json.dumps(policy_document)
+        ResourceArn=function_arn, Policy=json.dumps(policy_document)
     )
 
     logger.info("Function deployed successfully! %s", function_name)

examples/examples-catalog.json

@@ -110,6 +110,50 @@
         "ExecutionTimeout": 300
       },
       "path": "./src/map_operations.py"
+    },
+    {
+      "name": "Block Example",
+      "description": "Nested child contexts demonstrating block operations",
+      "handler": "block_example.handler",
+      "integration": true,
+      "durableConfig": {
+        "RetentionPeriodInDays": 7,
+        "ExecutionTimeout": 300
+      },
+      "path": "./src/block_example.py"
+    },
+    {
+      "name": "Logger Example",
+      "description": "Demonstrating logger usage and enrichment in DurableContext",
+      "handler": "logger_example.handler",
+      "integration": true,
+      "durableConfig": {
+        "RetentionPeriodInDays": 7,
+        "ExecutionTimeout": 300
+      },
+      "path": "./src/logger_example.py"
+    },
+    {
+      "name": "Steps with Retry",
+      "description": "Multiple steps with retry logic in a polling pattern",
+      "handler": "steps_with_retry.handler",
+      "integration": true,
+      "durableConfig": {
+        "RetentionPeriodInDays": 7,
+        "ExecutionTimeout": 300
+      },
+      "path": "./src/steps_with_retry.py"
+    },
+    {
+      "name": "Wait for Condition",
+      "description": "Polling pattern that waits for a condition to be met",
+      "handler": "wait_for_condition.handler",
+      "integration": true,
+      "durableConfig": {
+        "RetentionPeriodInDays": 7,
+        "ExecutionTimeout": 300
+      },
+      "path": "./src/wait_for_condition.py"
     }
   ]
 }

examples/src/block_example.py

@@ -0,0 +1,46 @@
+"""Example demonstrating nested child contexts (blocks)."""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    durable_with_child_context,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+
+
+@durable_with_child_context
+def nested_block(ctx: DurableContext) -> str:
+    """Nested block with its own child context."""
+    # Wait in the nested block
+    ctx.wait(seconds=1)
+    return "nested block result"
+
+
+@durable_with_child_context
+def parent_block(ctx: DurableContext) -> dict[str, str]:
+    """Parent block with nested operations."""
+    # Nested step
+    nested_result: str = ctx.step(
+        lambda _: "nested step result",
+        name="nested_step",
+    )
+
+    # Nested block with its own child context
+    nested_block_result: str = ctx.run_in_child_context(nested_block())
+
+    return {
+        "nestedStep": nested_result,
+        "nestedBlock": nested_block_result,
+    }
+
+
+@durable_execution
+def handler(_event: Any, context: DurableContext) -> dict[str, str]:
+    """Handler demonstrating nested child contexts."""
+    # Run parent block which contains nested operations
+    result: dict[str, str] = context.run_in_child_context(
+        parent_block(), name="parent_block"
+    )
+
+    return result

examples/src/logger_example.py

@@ -0,0 +1,50 @@
+"""Example demonstrating logger usage in DurableContext."""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    durable_with_child_context,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+
+
+@durable_with_child_context
+def child_workflow(ctx: DurableContext) -> str:
+    """Child workflow with its own logging context."""
+    # Child context logger has step_id populated with child context ID
+    ctx.logger.info("Running in child context")
+
+    # Step in child context has nested step ID
+    child_result: str = ctx.step(
+        lambda _: "child-processed",
+        name="child_step",
+    )
+
+    ctx.logger.info("Child workflow completed", extra={"result": child_result})
+
+    return child_result
+
+
+@durable_execution
+def handler(event: Any, context: DurableContext) -> str:
+    """Handler demonstrating logger usage."""
+    # Top-level context logger: no step_id field
+    context.logger.info("Starting workflow", extra={"eventId": event.get("id")})
+
+    # Logger in steps - gets enriched with step ID and attempt number
+    result1: str = context.step(
+        lambda _: "processed",
+        name="process_data",
+    )
+
+    context.logger.info("Step 1 completed", extra={"result": result1})
+
+    # Child contexts inherit the parent's logger and have their own step ID
+    result2: str = context.run_in_child_context(child_workflow(), name="child_workflow")
+
+    context.logger.info(
+        "Workflow completed", extra={"result1": result1, "result2": result2}
+    )
+
+    return f"{result1}-{result2}"

examples/src/map_operations.py

@@ -1,3 +1,5 @@
+"""Example demonstrating map-like operations for processing collections durably."""
+
 from typing import Any
 
 from aws_durable_execution_sdk_python.context import DurableContext
@@ -9,8 +11,8 @@ def square(x: int) -> int:
 
 
 @durable_execution
-def handler(_event: Any, context: DurableContext) -> str:
-    # Process a list of items using map-like operations
+def handler(_event: Any, context: DurableContext) -> list[int]:
+    """Process a list of items using map-like operations."""
     items = [1, 2, 3, 4, 5]
 
     # Process each item as a separate durable step
@@ -19,4 +21,4 @@ def handler(_event: Any, context: DurableContext) -> str:
         result = context.step(lambda _, x=item: square(x), name=f"square_{i}")
         results.append(result)
 
-    return f"Squared results: {results}"
+    return results

examples/src/parallel.py

@@ -1,15 +1,17 @@
+"""Example demonstrating parallel-like operations for concurrent execution."""
+
 from typing import Any
 
 from aws_durable_execution_sdk_python.context import DurableContext
 from aws_durable_execution_sdk_python.execution import durable_execution
 
 
 @durable_execution
-def handler(_event: Any, context: DurableContext) -> str:
-    # Execute multiple operations in parallel
+def handler(_event: Any, context: DurableContext) -> list[str]:
+    # Execute multiple operations
     task1 = context.step(lambda _: "Task 1 complete", name="task1")
     task2 = context.step(lambda _: "Task 2 complete", name="task2")
     task3 = context.step(lambda _: "Task 3 complete", name="task3")
 
     # All tasks execute concurrently and results are collected
-    return f"Results: {task1}, {task2}, {task3}"
+    return [task1, task2, task3]

examples/src/steps_with_retry.py

@@ -0,0 +1,73 @@
+"""Example demonstrating multiple steps with retry logic."""
+
+from random import random
+from typing import Any
+
+from aws_durable_execution_sdk_python.config import StepConfig
+from aws_durable_execution_sdk_python.context import DurableContext
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.retries import (
+    RetryStrategyConfig,
+    create_retry_strategy,
+)
+
+
+def simulated_get_item(name: str) -> dict[str, Any] | None:
+    """Simulate getting an item that may fail randomly."""
+    # Fail 50% of the time
+    if random() < 0.5:  # noqa: S311
+        msg = "Random failure"
+        raise RuntimeError(msg)
+
+    # Simulate finding item after some attempts
+    if random() > 0.3:  # noqa: S311
+        return {"id": name, "data": "item data"}
+
+    return None
+
+
+@durable_execution
+def handler(event: Any, context: DurableContext) -> dict[str, Any]:
+    """Handler demonstrating polling with retry logic."""
+    name = event.get("name", "test-item")
+
+    # Retry configuration for steps
+    retry_config = RetryStrategyConfig(
+        max_attempts=5,
+        retryable_error_types=[RuntimeError],
+    )
+
+    step_config = StepConfig(create_retry_strategy(retry_config))
+
+    item = None
+    poll_count = 0
+    max_polls = 5
+
+    try:
+        while poll_count < max_polls:
+            poll_count += 1
+
+            # Try to get the item with retry
+            get_response = context.step(
+                lambda _, n=name: simulated_get_item(n),
+                name=f"get_item_poll_{poll_count}",
+                config=step_config,
+            )
+
+            # Did we find the item?
+            if get_response:
+                item = get_response
+                break
+
+            # Wait 1 second until next poll
+            context.wait(seconds=1)
+
+    except RuntimeError as e:
+        # Retries exhausted
+        return {"error": "DDB Retries Exhausted", "message": str(e)}
+
+    if not item:
+        return {"error": "Item Not Found"}
+
+    # We found the item!
+    return {"success": True, "item": item, "pollsRequired": poll_count}

examples/src/wait_for_condition.py

@@ -0,0 +1,33 @@
+"""Example demonstrating wait-for-condition pattern."""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.context import DurableContext
+from aws_durable_execution_sdk_python.execution import durable_execution
+
+
+@durable_execution
+def handler(_event: Any, context: DurableContext) -> int:
+    """Handler demonstrating wait-for-condition pattern."""
+    state = 0
+    attempt = 0
+    max_attempts = 5
+
+    while attempt < max_attempts:
+        attempt += 1
+
+        # Execute step to update state
+        state = context.step(
+            lambda _, s=state: s + 1,
+            name=f"increment_state_{attempt}",
+        )
+
+        # Check condition
+        if state >= 3:
+            # Condition met, stop
+            break
+
+        # Wait before next attempt
+        context.wait(seconds=1)
+
+    return state