fix(testing): reject blank execution ARNs

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py

@@ -196,6 +196,22 @@ def timeout_handler():
             execution_arn=execution.durable_execution_arn
         )
 
+    @staticmethod
+    def _validate_execution_arn(execution_arn: str) -> None:
+        """Reject a blank or non-string execution ARN before it reaches the registry or store.
+
+        This runner's execution ARNs are opaque local identifiers, not
+        real AWS ARNs (see ``Execution.new``), so this only rules out
+        empty/malformed input rather than checking AWS ARN shape.
+
+        Raises:
+            InvalidParameterValueException: If the ARN is not a
+                non-empty string.
+        """
+        if not isinstance(execution_arn, str) or not execution_arn.strip():
+            msg: str = "Invalid Durable Execution ARN"
+            raise InvalidParameterValueException(msg)
+
     def get_execution(self, execution_arn: str) -> Execution:
         """Get execution by ARN.
 
@@ -206,8 +222,10 @@ def get_execution(self, execution_arn: str) -> Execution:
             Execution: The execution object
 
         Raises:
+            InvalidParameterValueException: If the ARN is blank.
             ResourceNotFoundException: If execution does not exist
         """
+        self._validate_execution_arn(execution_arn)
         try:
             return self._store.load(execution_arn)
         except KeyError as e:
@@ -376,8 +394,10 @@ def stop_execution(
             StopDurableExecutionResponse: Response containing end timestamp
 
         Raises:
+            InvalidParameterValueException: If the ARN is blank.
             ResourceNotFoundException: If execution does not exist
         """
+        self._validate_execution_arn(execution_arn)
         return self._registry.submit(
             execution_arn,
             CallableTask(lambda: self._apply_stop(execution_arn, error)),
@@ -415,7 +435,12 @@ def get_execution_state(
         marker: str | None = None,
         max_items: int | None = None,
     ) -> GetDurableExecutionStateResponse:
-        """Return a page of operations, serialized on the execution's worker."""
+        """Return a page of operations, serialized on the execution's worker.
+
+        Raises:
+            InvalidParameterValueException: If the ARN is blank.
+        """
+        self._validate_execution_arn(execution_arn)
         return self._registry.submit(
             execution_arn,
             CallableTask(
@@ -795,7 +820,11 @@ def checkpoint_execution(
 
         Routes through the per-execution worker so checkpoints for one
         execution never overlap.
+
+        Raises:
+            InvalidParameterValueException: If the ARN is blank.
         """
+        self._validate_execution_arn(execution_arn)
         return self._registry.submit(
             execution_arn,
             CallableTask(

packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py

@@ -1846,6 +1846,44 @@ def test_get_execution_not_found(executor, mock_store):
         executor.get_execution("test-arn")
 
 
+@pytest.mark.parametrize("blank_arn", ["", "   ", None])
+def test_get_execution_rejects_blank_arn(executor, mock_store, blank_arn):
+    with pytest.raises(InvalidParameterValueException):
+        executor.get_execution(blank_arn)
+
+    mock_store.load.assert_not_called()
+
+
+@pytest.mark.parametrize("blank_arn", ["", "   ", None])
+def test_stop_execution_rejects_blank_arn(executor, mock_store, blank_arn):
+    """A blank ARN must be rejected before it reaches the registry, so it
+    never creates a permanent phantom worker for an execution that was
+    never going to exist."""
+    with pytest.raises(InvalidParameterValueException):
+        executor.stop_execution(blank_arn)
+
+    mock_store.load.assert_not_called()
+    assert executor._registry.active_count() == 0  # noqa: SLF001
+
+
+@pytest.mark.parametrize("blank_arn", ["", "   ", None])
+def test_get_execution_state_rejects_blank_arn(executor, mock_store, blank_arn):
+    with pytest.raises(InvalidParameterValueException):
+        executor.get_execution_state(blank_arn, checkpoint_token="token")
+
+    mock_store.load.assert_not_called()
+    assert executor._registry.active_count() == 0  # noqa: SLF001
+
+
+@pytest.mark.parametrize("blank_arn", ["", "   ", None])
+def test_checkpoint_execution_rejects_blank_arn(executor, mock_store, blank_arn):
+    with pytest.raises(InvalidParameterValueException):
+        executor.checkpoint_execution(blank_arn, checkpoint_token="token")
+
+    mock_store.load.assert_not_called()
+    assert executor._registry.active_count() == 0  # noqa: SLF001
+
+
 def test_get_execution_state(mock_scheduler, mock_invoker, mock_checkpoint_processor):
     """GetDurableExecutionState is a pure read from the pinned
     snapshot, bounded by the configured byte cap. Uses a real store