fix: make retry examples deterministic using seeded random

Author: vip-amzn

examples/src/step_with_retry.py

@@ -1,4 +1,4 @@
-from random import random
+import random
 from typing import Any
 
 from aws_durable_execution_sdk_python.config import StepConfig
@@ -18,15 +18,20 @@
 def unreliable_operation(
     _step_context: StepContext,
 ) -> str:
+    # Use seeded random for deterministic behavior
+    # With seed 42, this will fail twice then succeed
     failure_threshold = 0.5
-    if random() > failure_threshold:  # noqa: S311
+    if random.random() > failure_threshold:  # noqa: S311
         msg = "Random error occurred"
         raise RuntimeError(msg)
     return "Operation succeeded"
 
 
 @durable_execution
 def handler(_event: Any, context: DurableContext) -> str:
+    # Seed random for deterministic behavior
+    random.seed(42)
+
     retry_config = RetryStrategyConfig(
         max_attempts=3,
         retryable_error_types=[RuntimeError],

examples/src/steps_with_retry.py

@@ -1,26 +1,27 @@
 """Example demonstrating multiple steps with retry logic."""
 
-from random import 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.context import DurableContext, StepContext
 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."""
+def simulated_get_item(_step_context: StepContext, name: str) -> dict[str, Any] | None:
+    """Simulate getting an item with deterministic seeded random behavior."""
+    # Use seeded random for deterministic behavior
     # Fail 50% of the time
-    if random() < 0.5:  # noqa: S311
+    if random.random() < 0.5:  # noqa: S311
         msg = "Random failure"
         raise RuntimeError(msg)
 
     # Simulate finding item after some attempts
-    if random() > 0.3:  # noqa: S311
+    if random.random() > 0.3:  # noqa: S311
         return {"id": name, "data": "item data"}
 
     return None
@@ -29,6 +30,9 @@ def simulated_get_item(name: str) -> dict[str, Any] | None:
 @durable_execution
 def handler(event: Any, context: DurableContext) -> dict[str, Any]:
     """Handler demonstrating polling with retry logic."""
+    # Seed random for deterministic behavior
+    random.seed(42)
+
     name = event.get("name", "test-item")
 
     # Retry configuration for steps
@@ -49,7 +53,7 @@ def handler(event: Any, context: DurableContext) -> dict[str, Any]:
 
             # Try to get the item with retry
             get_response = context.step(
-                lambda _, n=name: simulated_get_item(n),
+                lambda _, n=name: simulated_get_item(_, n),
                 name=f"get_item_poll_{poll_count}",
                 config=step_config,
             )