Fix SFN.client.start_execution() idempotency

moto/stepfunctions/models.py

@@ -123,15 +123,21 @@ def start_execution(
         execution_name: str,
         execution_input: str,
     ) -> "Execution":
-        self._ensure_execution_name_doesnt_exist(execution_name)
         self._validate_execution_input(execution_input)
+        existing_execution = self._handle_name_input_idempotency(
+            execution_name, execution_input
+        )
+        if existing_execution is not None:
+            # If we found a match for the name and input, return the existing execution.
+            return existing_execution
+
         execution = Execution(
             region_name=region_name,
             account_id=account_id,
             state_machine_name=self.name,
             execution_name=execution_name,
             state_machine_arn=self.arn,
-            execution_input=json.loads(execution_input),
+            execution_input=execution_input,
         )
         self.executions.append(execution)
         return execution
@@ -147,12 +153,20 @@ def stop_execution(self, execution_arn: str) -> "Execution":
         execution.stop(stop_date=datetime.now(), error="", cause="")
         return execution
 
-    def _ensure_execution_name_doesnt_exist(self, name: str) -> None:
+    def _handle_name_input_idempotency(
+        self, name: str, execution_input: str
+    ) -> Optional["Execution"]:
         for execution in self.executions:
             if execution.name == name:
+                # Executions with the same name and input are considered idempotent
+                if execution_input == execution.execution_input:
+                    return execution
+
+                # If the inputs are different, raise
                 raise ExecutionAlreadyExists(
                     "Execution Already Exists: '" + execution.execution_arn + "'"
                 )
+        return None
 
     def _validate_execution_input(self, execution_input: str) -> None:
         try:

moto/stepfunctions/parser/backend/execution.py

@@ -133,7 +133,7 @@ def __init__(
         start_date: Timestamp,
         cloud_watch_logging_session: Optional[CloudWatchLoggingSession],
         activity_store: dict[Arn, Activity],
-        input_data: Optional[json] = None,
+        input_data: str,
         trace_header: Optional[TraceHeader] = None,
     ):
         self.name = name
@@ -145,7 +145,7 @@ def __init__(
         self.region_name = region_name
         self.state_machine = state_machine
         self._cloud_watch_logging_session = cloud_watch_logging_session
-        self.input_data = input_data
+        self.input_data = json.loads(input_data)
         self.input_details = CloudWatchEventsExecutionDataDetails(included=True)
         self.trace_header = trace_header
         self.exec_status = None

moto/stepfunctions/parser/models.py

@@ -169,10 +169,13 @@ def start_execution(
         state_machine_clone = copy.deepcopy(state_machine)
 
         if execution_input is None:
-            input_data = {}
+            input_data = "{}"
         else:
+            input_data = execution_input
             try:
-                input_data = json.loads(execution_input)
+                # Make sure input is valid json
+                json.loads(execution_input)
+
             except Exception as ex:
                 raise InvalidExecutionInput(
                     str(ex)

moto/stepfunctions/responses.py

@@ -207,7 +207,7 @@ def describe_execution(self) -> TYPE_RESPONSE:
         execution = self.stepfunction_backend.describe_execution(arn)
         response = {
             "executionArn": arn,
-            "input": json.dumps(execution.execution_input),
+            "input": execution.execution_input,
             "name": execution.name,
             "startDate": iso_8601_datetime_with_milliseconds(execution.start_date),
             "stateMachineArn": execution.state_machine_arn,

tests/test_stepfunctions/test_stepfunctions.py

@@ -540,25 +540,56 @@ def test_state_machine_start_execution_with_custom_name():
 
 
 @mock_aws
-def test_state_machine_start_execution_fails_on_duplicate_execution_name():
+def test_state_machine_start_execution_fails_on_duplicate_execution_name_with_different_input():
     client = boto3.client("stepfunctions", region_name=region)
     #
     sm = client.create_state_machine(
         name="name", definition=str(simple_definition), roleArn=_get_default_role()
     )
     execution_one = client.start_execution(
-        stateMachineArn=sm["stateMachineArn"], name="execution_name"
+        stateMachineArn=sm["stateMachineArn"],
+        name="execution_name",
+        input='{"a": "b", "c": "d"}',
     )
     #
     with pytest.raises(ClientError) as ex:
         _ = client.start_execution(
-            stateMachineArn=sm["stateMachineArn"], name="execution_name"
+            stateMachineArn=sm["stateMachineArn"],
+            name="execution_name",
+            # Input is different (even though the decoded json is equivalent)
+            input='{"c": "d", "a": "b"}',
         )
     assert ex.value.response["Error"]["Message"] == (
         "Execution Already Exists: '" + execution_one["executionArn"] + "'"
     )
 
 
+@mock_aws
+def test_state_machine_start_execution_is_idempotent_by_name_and_input():
+    client = boto3.client("stepfunctions", region_name=region)
+    #
+    sm = client.create_state_machine(
+        name="name", definition=str(simple_definition), roleArn=_get_default_role()
+    )
+    execution_input = '{"a": "b", "c": "d"}'
+    execution_one = client.start_execution(
+        stateMachineArn=sm["stateMachineArn"],
+        name="execution_name",
+        input=execution_input,
+    )
+    #
+    execution_two = client.start_execution(
+        stateMachineArn=sm["stateMachineArn"],
+        name="execution_name",
+        input=execution_input,
+    )
+    assert execution_one["executionArn"] == execution_two["executionArn"]
+
+    # Check idempotency
+    list_execs = client.list_executions(stateMachineArn=sm["stateMachineArn"])
+    assert len(list_execs["executions"]) == 1
+
+
 @mock_aws
 def test_state_machine_start_execution_with_custom_input():
     client = boto3.client("stepfunctions", region_name=region)