Repository: getmoto/moto
Pull request #7563 (2024-04-04 14:12:30)

fix(cognito-idp): avoid KeyError on unexisting refresh token
If an unexisting refresh token was passed to the `initiate_auth` with the `REFRESH_TOKEN_AUTH` flow, firstly the operation fails because of `KeyError` and after a few retrials, Moto returns a 500 status code response.

The fix was to use the `.get()` method to get a dictionary member or `None` if such doesn't exist instead of direct dictionary access with `[]`.


Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index b9ffea9a2..0238a60f1 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -1039,7 +1039,7 @@
 
 ## cloudwatch
 <details>
-<summary>36% implemented</summary>
+<summary>39% implemented</summary>
 
 - [X] delete_alarms
 - [ ] delete_anomaly_detector
@@ -1047,7 +1047,7 @@
 - [ ] delete_insight_rules
 - [ ] delete_metric_stream
 - [ ] describe_alarm_history
-- [ ] describe_alarms
+- [X] describe_alarms
 - [ ] describe_alarms_for_metric
 - [ ] describe_anomaly_detectors
 - [ ] describe_insight_rules
diff --git a/docs/docs/services/cloudwatch.rst b/docs/docs/services/cloudwatch.rst
index 2578fc416..0606145d9 100644
--- a/docs/docs/services/cloudwatch.rst
+++ b/docs/docs/services/cloudwatch.rst
@@ -20,7 +20,7 @@ cloudwatch
 - [ ] delete_insight_rules
 - [ ] delete_metric_stream
 - [ ] describe_alarm_history
-- [ ] describe_alarms
+- [X] describe_alarms
 - [ ] describe_alarms_for_metric
 - [ ] describe_anomaly_detectors
 - [ ] describe_insight_rules
diff --git a/moto/applicationautoscaling/models.py b/moto/applicationautoscaling/models.py
index 41e7f2dda..91214821f 100644
--- a/moto/applicationautoscaling/models.py
+++ b/moto/applicationautoscaling/models.py
@@ -1,7 +1,7 @@
 import time
 from collections import OrderedDict
 from enum import Enum, unique
-from typing import Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
@@ -10,6 +10,9 @@ from moto.moto_api._internal import mock_random
 
 from .exceptions import AWSValidationException
 
+if TYPE_CHECKING:
+    from moto.cloudwatch.models import FakeAlarm
+
 
 @unique
 class ResourceTypeExceptionValueSet(Enum):
@@ -76,7 +79,6 @@ class ApplicationAutoscalingBackend(BaseBackend):
     def describe_scalable_targets(
         self, namespace: str, r_ids: Union[None, List[str]], dimension: Union[None, str]
     ) -> List["FakeScalableTarget"]:
-        """Describe scalable targets."""
         if r_ids is None:
             r_ids = []
         targets = self._flatten_scalable_targets(namespace)
@@ -105,7 +107,6 @@ class ApplicationAutoscalingBackend(BaseBackend):
         role_arn: str,
         suspended_state: str,
     ) -> "FakeScalableTarget":
-        """Registers or updates a scalable target."""
         _ = _target_params_are_valid(namespace, r_id, dimension)
         if namespace == ServiceNamespaceValueSet.ECS.value:
             _ = self._ecs_service_exists_for_target(r_id)
@@ -151,7 +152,6 @@ class ApplicationAutoscalingBackend(BaseBackend):
     def deregister_scalable_target(
         self, namespace: str, r_id: str, dimension: str
     ) -> None:
-        """Registers or updates a scalable target."""
         if self._scalable_target_exists(r_id, dimension):
             del self.targets[dimension][r_id]
         else:
@@ -165,7 +165,7 @@ class ApplicationAutoscalingBackend(BaseBackend):
         service_namespace: str,
         resource_id: str,
         scalable_dimension: str,
-        policy_body: str,
+        policy_body: Dict[str, Any],
         policy_type: Optional[None],
     ) -> "FakeApplicationAutoscalingPolicy":
         policy_key = FakeApplicationAutoscalingPolicy.formulate_key(
@@ -238,6 +238,8 @@ class ApplicationAutoscalingBackend(BaseBackend):
             service_namespace, resource_id, scalable_dimension, policy_name
         )
         if policy_key in self.policies:
+            policy = self.policies[policy_key]
+            policy.delete_alarms(self.account_id, self.region_name)
             del self.policies[policy_key]
         else:
             raise AWSValidationException(
@@ -438,7 +440,7 @@ class FakeApplicationAutoscalingPolicy(BaseModel):
         resource_id: str,
         scalable_dimension: str,
         policy_type: Optional[str],
-        policy_body: str,
+        policy_body: Dict[str, Any],
     ) -> None:
         self.step_scaling_policy_configuration = None
         self.target_tracking_scaling_policy_configuration = None
@@ -451,7 +453,7 @@ class FakeApplicationAutoscalingPolicy(BaseModel):
             self.target_tracking_scaling_policy_configuration = policy_body
         else:
             raise AWSValidationException(
-                f"Unknown policy type {policy_type} specified."
+                f"1 validation error detected: Value '{policy_type}' at 'policyType' failed to satisfy constraint: Member must satisfy enum value set: [PredictiveScaling, StepScaling, TargetTrackingScaling]"
             )
 
         self._policy_body = policy_body
@@ -463,6 +465,151 @@ class FakeApplicationAutoscalingPolicy(BaseModel):
         self._guid = mock_random.uuid4()
         self.policy_arn = f"arn:aws:autoscaling:{region_name}:{account_id}:scalingPolicy:{self._guid}:resource/{self.service_namespace}/{self.resource_id}:policyName/{self.policy_name}"
         self.creation_time = time.time()
+        self.alarms: List["FakeAlarm"] = []
+
+        self.account_id = account_id
+        self.region_name = region_name
+
+        self.create_alarms()
+
+    def create_alarms(self) -> None:
+        if self.policy_type == "TargetTrackingScaling":
+            if self.service_namespace == "dynamodb":
+                self.alarms.extend(self._generate_dynamodb_alarms())
+            if self.service_namespace == "ecs":
+                self.alarms.extend(self._generate_ecs_alarms())
+
+    def _generate_dynamodb_alarms(self) -> List["FakeAlarm"]:
+        from moto.cloudwatch.models import CloudWatchBackend, cloudwatch_backends
+
+        cloudwatch: CloudWatchBackend = cloudwatch_backends[self.account_id][
+            self.region_name
+        ]
+        alarms = []
+        table_name = self.resource_id.split("/")[-1]
+        alarm_action = f"{self.policy_arn}:createdBy/{mock_random.uuid4()}"
+        alarm1 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-table/{table_name}-AlarmHigh-{mock_random.uuid4()}",
+            namespace="AWS/DynamoDB",
+            metric_name="ConsumedReadCapacityUnits",
+            metric_data_queries=[],
+            comparison_operator="GreaterThanThreshold",
+            evaluation_periods=2,
+            period=60,
+            threshold=42.0,
+            statistic="Sum",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[{"name": "TableName", "value": table_name}],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm1)
+        alarm2 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-table/{table_name}-AlarmLow-{mock_random.uuid4()}",
+            namespace="AWS/DynamoDB",
+            metric_name="ConsumedReadCapacityUnits",
+            metric_data_queries=[],
+            comparison_operator="LessThanThreshold",
+            evaluation_periods=15,
+            period=60,
+            threshold=30.0,
+            statistic="Sum",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[{"name": "TableName", "value": table_name}],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm2)
+        alarm3 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-table/{table_name}-ProvisionedCapacityHigh-{mock_random.uuid4()}",
+            namespace="AWS/DynamoDB",
+            metric_name="ProvisionedReadCapacityUnits",
+            metric_data_queries=[],
+            comparison_operator="GreaterThanThreshold",
+            evaluation_periods=2,
+            period=300,
+            threshold=1.0,
+            statistic="Average",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[{"name": "TableName", "value": table_name}],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm3)
+        alarm4 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-table/{table_name}-ProvisionedCapacityLow-{mock_random.uuid4()}",
+            namespace="AWS/DynamoDB",
+            metric_name="ProvisionedReadCapacityUnits",
+            metric_data_queries=[],
+            comparison_operator="LessThanThreshold",
+            evaluation_periods=3,
+            period=300,
+            threshold=1.0,
+            statistic="Average",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[{"name": "TableName", "value": table_name}],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm4)
+        return alarms
+
+    def _generate_ecs_alarms(self) -> List["FakeAlarm"]:
+        from moto.cloudwatch.models import CloudWatchBackend, cloudwatch_backends
+
+        cloudwatch: CloudWatchBackend = cloudwatch_backends[self.account_id][
+            self.region_name
+        ]
+        alarms: List["FakeAlarm"] = []
+        alarm_action = f"{self.policy_arn}:createdBy/{mock_random.uuid4()}"
+        config = self.target_tracking_scaling_policy_configuration or {}
+        metric_spec = config.get("PredefinedMetricSpecification", {})
+        if "Memory" in metric_spec.get("PredefinedMetricType", ""):
+            metric_name = "MemoryUtilization"
+        else:
+            metric_name = "CPUUtilization"
+        _, cluster_name, service_name = self.resource_id.split("/")
+        alarm1 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-{self.resource_id}-AlarmHigh-{mock_random.uuid4()}",
+            namespace="AWS/ECS",
+            metric_name=metric_name,
+            metric_data_queries=[],
+            comparison_operator="GreaterThanThreshold",
+            evaluation_periods=3,
+            period=60,
+            threshold=6,
+            unit="Percent",
+            statistic="Average",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[
+                {"name": "ClusterName", "value": cluster_name},
+                {"name": "ServiceName", "value": service_name},
+            ],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm1)
+        alarm2 = cloudwatch.put_metric_alarm(
+            name=f"TargetTracking-{self.resource_id}-AlarmLow-{mock_random.uuid4()}",
+            namespace="AWS/ECS",
+            metric_name=metric_name,
+            metric_data_queries=[],
+            comparison_operator="LessThanThreshold",
+            evaluation_periods=15,
+            period=60,
+            threshold=6,
+            unit="Percent",
+            statistic="Average",
+            description=f"DO NOT EDIT OR DELETE. For TargetTrackingScaling policy {alarm_action}",
+            dimensions=[
+                {"name": "ClusterName", "value": cluster_name},
+                {"name": "ServiceName", "value": service_name},
+            ],
+            alarm_actions=[alarm_action],
+        )
+        alarms.append(alarm2)
+        return alarms
+
+    def delete_alarms(self, account_id: str, region_name: str) -> None:
+        from moto.cloudwatch.models import CloudWatchBackend, cloudwatch_backends
+
+        cloudwatch: CloudWatchBackend = cloudwatch_backends[account_id][region_name]
+        cloudwatch.delete_alarms([a.name for a in self.alarms])
 
     @staticmethod
     def formulate_key(
diff --git a/moto/applicationautoscaling/responses.py b/moto/applicationautoscaling/responses.py
index fdf2358d7..f1ccca509 100644
--- a/moto/applicationautoscaling/responses.py
+++ b/moto/applicationautoscaling/responses.py
@@ -1,5 +1,5 @@
 import json
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from moto.core.responses import BaseResponse
 
@@ -79,7 +79,9 @@ class ApplicationAutoScalingResponse(BaseResponse):
                 self._get_param("TargetTrackingScalingPolicyConfiguration"),
             ),
         )
-        return json.dumps({"PolicyARN": policy.policy_arn, "Alarms": []})  # ToDo
+        return json.dumps(
+            {"PolicyARN": policy.policy_arn, "Alarms": _build_alarms(policy)}
+        )
 
     def describe_scaling_policies(self) -> str:
         (
@@ -205,6 +207,10 @@ def _build_target(t: FakeScalableTarget) -> Dict[str, Any]:
     }
 
 
+def _build_alarms(policy: FakeApplicationAutoscalingPolicy) -> List[Dict[str, str]]:
+    return [{"AlarmARN": a.alarm_arn, "AlarmName": a.name} for a in policy.alarms]
+
+
 def _build_policy(p: FakeApplicationAutoscalingPolicy) -> Dict[str, Any]:
     response = {
         "PolicyARN": p.policy_arn,
@@ -214,6 +220,7 @@ def _build_policy(p: FakeApplicationAutoscalingPolicy) -> Dict[str, Any]:
         "ScalableDimension": p.scalable_dimension,
         "PolicyType": p.policy_type,
         "CreationTime": p.creation_time,
+        "Alarms": _build_alarms(p),
     }
     if p.policy_type == "StepScaling":
         response["StepScalingPolicyConfiguration"] = p.step_scaling_policy_configuration
diff --git a/moto/cloudwatch/models.py b/moto/cloudwatch/models.py
index 60a5dd03f..b0f50f9d1 100644
--- a/moto/cloudwatch/models.py
+++ b/moto/cloudwatch/models.py
@@ -119,32 +119,32 @@ class FakeAlarm(BaseModel):
         name: str,
         namespace: str,
         metric_name: str,
-        metric_data_queries: List[MetricDataQuery],
+        metric_data_queries: Optional[List[MetricDataQuery]],
         comparison_operator: str,
         evaluation_periods: int,
-        datapoints_to_alarm: int,
+        datapoints_to_alarm: Optional[int],
         period: int,
         threshold: float,
         statistic: str,
-        extended_statistic: str,
+        extended_statistic: Optional[str],
         description: str,
         dimensions: List[Dict[str, str]],
         alarm_actions: List[str],
-        ok_actions: List[str],
-        insufficient_data_actions: List[str],
-        unit: str,
+        ok_actions: Optional[List[str]],
+        insufficient_data_actions: Optional[List[str]],
+        unit: Optional[str],
         actions_enabled: bool,
-        treat_missing_data: str,
-        evaluate_low_sample_count_percentile: str,
-        threshold_metric_id: str,
-        rule: str,
+        treat_missing_data: Optional[str],
+        evaluate_low_sample_count_percentile: Optional[str],
+        threshold_metric_id: Optional[str],
+        rule: Optional[str],
     ):
         self.region_name = region_name
         self.name = name
         self.alarm_arn = make_arn_for_alarm(region_name, account_id, name)
         self.namespace = namespace
         self.metric_name = metric_name
-        self.metric_data_queries = metric_data_queries
+        self.metric_data_queries = metric_data_queries or []
         self.comparison_operator = comparison_operator
         self.evaluation_periods = evaluation_periods
         self.datapoints_to_alarm = datapoints_to_alarm
@@ -158,8 +158,8 @@ class FakeAlarm(BaseModel):
         ]
         self.actions_enabled = True if actions_enabled is None else actions_enabled
         self.alarm_actions = alarm_actions
-        self.ok_actions = ok_actions
-        self.insufficient_data_actions = insufficient_data_actions
+        self.ok_actions = ok_actions or []
+        self.insufficient_data_actions = insufficient_data_actions or []
         self.unit = unit
         self.configuration_updated_timestamp = iso_8601_datetime_with_nanoseconds()
         self.treat_missing_data = treat_missing_data
@@ -457,26 +457,26 @@ class CloudWatchBackend(BaseBackend):
         name: str,
         namespace: str,
         metric_name: str,
-        metric_data_queries: List[MetricDataQuery],
         comparison_operator: str,
         evaluation_periods: int,
-        datapoints_to_alarm: int,
         period: int,
         threshold: float,
         statistic: str,
-        extended_statistic: str,
         description: str,
         dimensions: List[Dict[str, str]],
         alarm_actions: List[str],
-        ok_actions: List[str],
-        insufficient_data_actions: List[str],
-        unit: str,
-        actions_enabled: bool,
-        treat_missing_data: str,
-        evaluate_low_sample_count_percentile: str,
-        threshold_metric_id: str,
-        rule: str,
-        tags: List[Dict[str, str]],
+        metric_data_queries: Optional[List[MetricDataQuery]] = None,
+        datapoints_to_alarm: Optional[int] = None,
+        extended_statistic: Optional[str] = None,
+        ok_actions: Optional[List[str]] = None,
+        insufficient_data_actions: Optional[List[str]] = None,
+        unit: Optional[str] = None,
+        actions_enabled: bool = True,
+        treat_missing_data: Optional[str] = None,
+        evaluate_low_sample_count_percentile: Optional[str] = None,
+        threshold_metric_id: Optional[str] = None,
+        rule: Optional[str] = None,
+        tags: Optional[List[Dict[str, str]]] = None,
     ) -> FakeAlarm:
         if extended_statistic and not extended_statistic.startswith("p"):
             raise InvalidParameterValue(
@@ -519,11 +519,12 @@ class CloudWatchBackend(BaseBackend):
         )
 
         self.alarms[name] = alarm
-        self.tagger.tag_resource(alarm.alarm_arn, tags)
+        if tags:
+            self.tagger.tag_resource(alarm.alarm_arn, tags)
 
         return alarm
 
-    def get_all_alarms(self) -> Iterable[FakeAlarm]:
+    def describe_alarms(self) -> Iterable[FakeAlarm]:
         return self.alarms.values()
 
     @staticmethod
@@ -889,7 +890,7 @@ class CloudWatchBackend(BaseBackend):
     def tag_resource(self, arn: str, tags: List[Dict[str, str]]) -> None:
         # From boto3:
         # Currently, the only CloudWatch resources that can be tagged are alarms and Contributor Insights rules.
-        all_arns = [alarm.alarm_arn for alarm in self.get_all_alarms()]
+        all_arns = [alarm.alarm_arn for alarm in self.describe_alarms()]
         if arn not in all_arns:
             raise ResourceNotFoundException
 
diff --git a/moto/cloudwatch/responses.py b/moto/cloudwatch/responses.py
index e9bbb3f65..6675d0998 100644
--- a/moto/cloudwatch/responses.py
+++ b/moto/cloudwatch/responses.py
@@ -147,7 +147,7 @@ class CloudWatchResponse(BaseResponse):
         elif state_value:
             alarms = self.cloudwatch_backend.get_alarms_by_state_value(state_value)
         else:
-            alarms = self.cloudwatch_backend.get_all_alarms()
+            alarms = self.cloudwatch_backend.describe_alarms()
 
         metric_alarms = [a for a in alarms if a.rule is None]
         composite_alarms = [a for a in alarms if a.rule is not None]
@@ -244,9 +244,6 @@ class CloudWatchResponse(BaseResponse):
         template = self.response_template(DELETE_DASHBOARD_TEMPLATE)
         return template.render()
 
-    def describe_alarm_history(self) -> None:
-        raise NotImplementedError()
-
     @staticmethod
     def filter_alarms(
         alarms: Iterable[FakeAlarm], metric_name: str, namespace: str
@@ -259,7 +256,7 @@ class CloudWatchResponse(BaseResponse):
         return metric_filtered_alarms
 
     def describe_alarms_for_metric(self) -> str:
-        alarms = self.cloudwatch_backend.get_all_alarms()
+        alarms = self.cloudwatch_backend.describe_alarms()
         namespace = self._get_param("Namespace")
         metric_name = self._get_param("MetricName")
         filtered_alarms = self.filter_alarms(alarms, metric_name, namespace)
diff --git a/moto/cognitoidp/models.py b/moto/cognitoidp/models.py
index e1dd68cc9..2b8917197 100644
--- a/moto/cognitoidp/models.py
+++ b/moto/cognitoidp/models.py
@@ -1964,7 +1964,7 @@ class CognitoIdpBackend(BaseBackend):
             if not refresh_token:
                 raise ResourceNotFoundError(refresh_token)
 
-            res = user_pool.refresh_tokens[refresh_token]
+            res = user_pool.refresh_tokens.get(refresh_token)
             if res is None:
                 raise NotAuthorizedError("Refresh Token has been revoked")
 
diff --git a/moto/iot/models.py b/moto/iot/models.py
index 7b0e2a48c..2193a960c 100644
--- a/moto/iot/models.py
+++ b/moto/iot/models.py
@@ -1796,6 +1796,9 @@ class IoTBackend(BaseBackend):
         return self.rules[rule_name].to_get_dict()
 
     def create_topic_rule(self, rule_name: str, sql: str, **kwargs: Any) -> None:
+        if not re.match("^[a-zA-Z0-9_]+$", rule_name):
+            msg = f"1 validation error detected: Value '{rule_name}' at 'ruleName' failed to satisfy constraint: Member must satisfy regular expression pattern: ^[a-zA-Z0-9_]+$"
+            raise InvalidRequestException(msg)
         if rule_name in self.rules:
             raise ResourceAlreadyExistsException(
                 "Rule with given name already exists", "", self.rules[rule_name].arn
diff --git a/moto/iotdata/models.py b/moto/iotdata/models.py
index cb4b7cce1..8f25f13f0 100644
--- a/moto/iotdata/models.py
+++ b/moto/iotdata/models.py
@@ -153,7 +153,7 @@ class FakeShadow(BaseModel):
 class IoTDataPlaneBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
-        self.published_payloads: List[Tuple[str, str]] = list()
+        self.published_payloads: List[Tuple[str, bytes]] = list()
 
     @property
     def iot_backend(self) -> IoTBackend:
@@ -211,7 +211,7 @@ class IoTDataPlaneBackend(BaseBackend):
         thing.thing_shadows[shadow_name] = new_shadow
         return new_shadow
 
-    def publish(self, topic: str, payload: str) -> None:
+    def publish(self, topic: str, payload: bytes) -> None:
         self.published_payloads.append((topic, payload))
 
     def list_named_shadows_for_thing(self, thing_name: str) -> List[FakeShadow]:
diff --git a/moto/stepfunctions/models.py b/moto/stepfunctions/models.py
index 0a5e392f1..ab20597db 100644
--- a/moto/stepfunctions/models.py
+++ b/moto/stepfunctions/models.py
@@ -60,7 +60,7 @@ class StateMachine(CloudFormationModel):
             state_machine_name=self.name,
             execution_name=execution_name,
             state_machine_arn=self.arn,
-            execution_input=execution_input,
+            execution_input=json.loads(execution_input),
         )
         self.executions.append(execution)
         return execution
diff --git a/setup.cfg b/setup.cfg
index a7d287bc5..1594df76a 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -54,7 +54,7 @@ all =
     openapi-spec-validator>=0.5.0
     pyparsing>=3.0.7
     jsondiff>=1.1.2
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
     aws-xray-sdk!=0.96,>=0.93
     setuptools
     multipart
@@ -69,7 +69,7 @@ proxy =
     openapi-spec-validator>=0.5.0
     pyparsing>=3.0.7
     jsondiff>=1.1.2
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
     aws-xray-sdk!=0.96,>=0.93
     setuptools
     multipart
@@ -84,7 +84,7 @@ server =
     openapi-spec-validator>=0.5.0
     pyparsing>=3.0.7
     jsondiff>=1.1.2
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
     aws-xray-sdk!=0.96,>=0.93
     setuptools
     flask!=2.2.0,!=2.2.1
@@ -119,7 +119,7 @@ cloudformation =
     openapi-spec-validator>=0.5.0
     pyparsing>=3.0.7
     jsondiff>=1.1.2
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
     aws-xray-sdk!=0.96,>=0.93
     setuptools
 cloudfront =
@@ -141,10 +141,10 @@ dms =
 ds =
 dynamodb =
     docker>=3.0.0
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
 dynamodbstreams =
     docker>=3.0.0
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
 ebs =
 ec2 =
 ec2instanceconnect =
@@ -208,15 +208,15 @@ resourcegroupstaggingapi =
     openapi-spec-validator>=0.5.0
     pyparsing>=3.0.7
     jsondiff>=1.1.2
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
 route53 =
 route53resolver =
 s3 =
     PyYAML>=5.1
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
 s3crc32c =
     PyYAML>=5.1
-    py-partiql-parser==0.5.2
+    py-partiql-parser==0.5.4
     crc32c
 s3control =
 sagemaker =
@@ -273,7 +273,7 @@ disable = W,C,R,E
 enable = arguments-renamed, deprecated-module, function-redefined, redefined-outer-name, signature-differs
 
 [mypy]
-files= moto, tests/test_core, tests/test_batch_simple
+files= moto, tests/test_core, tests/test_batch_simple, tests/test_iotdata/
 exclude = moto/stepfunctions/parser
 show_column_numbers=True
 show_error_codes = True


Test patch:
diff --git a/tests/test_applicationautoscaling/__init__.py b/tests/test_applicationautoscaling/__init__.py
index e69de29bb..d197ac727 100644
--- a/tests/test_applicationautoscaling/__init__.py
+++ b/tests/test_applicationautoscaling/__init__.py
@@ -0,0 +1,71 @@
+import os
+from functools import wraps
+
+import boto3
+
+from moto import mock_aws
+from tests.test_dynamodb import dynamodb_aws_verified
+
+
+def application_autoscaling_aws_verified():
+    """
+    Function that is verified to work against AWS.
+    Can be run against AWS at any time by setting:
+      MOTO_TEST_ALLOW_AWS_REQUEST=true
+
+    If this environment variable is not set, the function runs in a `mock_aws` context.
+
+    This decorator will:
+      - Register a Scalable Target
+      - Run the test
+      - Deregister the Scalable Target
+    """
+
+    def inner(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            allow_aws_request = (
+                os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
+            )
+
+            @dynamodb_aws_verified()
+            def scalable_target_with_dynamodb_table(table_name):
+                namespace = "dynamodb"
+                resource_id = f"table/{table_name}"
+                scalable_dimension = "dynamodb:table:ReadCapacityUnits"
+
+                client = boto3.client(
+                    "application-autoscaling", region_name="us-east-1"
+                )
+                kwargs["client"] = client
+                kwargs["namespace"] = namespace
+                kwargs["resource_id"] = resource_id
+                kwargs["scalable_dimension"] = scalable_dimension
+
+                client.register_scalable_target(
+                    ServiceNamespace=namespace,
+                    ResourceId=resource_id,
+                    ScalableDimension=scalable_dimension,
+                    MinCapacity=1,
+                    MaxCapacity=4,
+                )
+                try:
+                    resp = func(*args, **kwargs)
+                finally:
+                    client.deregister_scalable_target(
+                        ServiceNamespace=namespace,
+                        ResourceId=resource_id,
+                        ScalableDimension=scalable_dimension,
+                    )
+
+                return resp
+
+            if allow_aws_request:
+                return scalable_target_with_dynamodb_table()
+            else:
+                with mock_aws():
+                    return scalable_target_with_dynamodb_table()
+
+        return wrapper
+
+    return inner
diff --git a/tests/test_applicationautoscaling/test_applicationautoscaling.py b/tests/test_applicationautoscaling/test_applicationautoscaling.py
index 94d6239aa..a068dfa8d 100644
--- a/tests/test_applicationautoscaling/test_applicationautoscaling.py
+++ b/tests/test_applicationautoscaling/test_applicationautoscaling.py
@@ -306,80 +306,6 @@ def test_register_scalable_target_updates_existing_target():
     )
 
 
-@pytest.mark.parametrize(
-    ["policy_type", "policy_body_kwargs"],
-    [
-        [
-            "TargetTrackingScaling",
-            {
-                "TargetTrackingScalingPolicyConfiguration": {
-                    "TargetValue": 70.0,
-                    "PredefinedMetricSpecification": {
-                        "PredefinedMetricType": "SageMakerVariantInvocationsPerInstance"
-                    },
-                }
-            },
-        ],
-        [
-            "TargetTrackingScaling",
-            {
-                "StepScalingPolicyConfiguration": {
-                    "AdjustmentType": "ChangeCapacity",
-                    "StepAdjustments": [{"ScalingAdjustment": 10}],
-                    "MinAdjustmentMagnitude": 2,
-                },
-            },
-        ],
-    ],
-)
-@mock_aws
-def test_put_scaling_policy(policy_type, policy_body_kwargs):
-    client = boto3.client("application-autoscaling", region_name=DEFAULT_REGION)
-    namespace = "sagemaker"
-    resource_id = "endpoint/MyEndPoint/variant/MyVariant"
-    scalable_dimension = "sagemaker:variant:DesiredInstanceCount"
-
-    client.register_scalable_target(
-        ServiceNamespace=namespace,
-        ResourceId=resource_id,
-        ScalableDimension=scalable_dimension,
-        MinCapacity=1,
-        MaxCapacity=8,
-    )
-
-    policy_name = "MyPolicy"
-
-    with pytest.raises(client.exceptions.ValidationException) as e:
-        client.put_scaling_policy(
-            PolicyName=policy_name,
-            ServiceNamespace=namespace,
-            ResourceId=resource_id,
-            ScalableDimension=scalable_dimension,
-            PolicyType="ABCDEFG",
-            **policy_body_kwargs,
-        )
-    assert (
-        e.value.response["Error"]["Message"] == "Unknown policy type ABCDEFG specified."
-    )
-
-    response = client.put_scaling_policy(
-        PolicyName=policy_name,
-        ServiceNamespace=namespace,
-        ResourceId=resource_id,
-        ScalableDimension=scalable_dimension,
-        PolicyType=policy_type,
-        **policy_body_kwargs,
-    )
-    assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
-    assert (
-        re.match(
-            pattern=rf"arn:aws:autoscaling:{DEFAULT_REGION}:{ACCOUNT_ID}:scalingPolicy:.*:resource/{namespace}/{resource_id}:policyName/{policy_name}",
-            string=response["PolicyARN"],
-        )
-        is not None
-    )
-
-
 @mock_aws
 def test_describe_scaling_policies():
     client = boto3.client("application-autoscaling", region_name=DEFAULT_REGION)
diff --git a/tests/test_applicationautoscaling/test_applicationautoscaling_policies.py b/tests/test_applicationautoscaling/test_applicationautoscaling_policies.py
new file mode 100644
index 000000000..51772d879
--- /dev/null
+++ b/tests/test_applicationautoscaling/test_applicationautoscaling_policies.py
@@ -0,0 +1,190 @@
+import boto3
+import pytest
+
+from moto import mock_aws
+
+from . import application_autoscaling_aws_verified
+
+target_tracking_config = {
+    "TargetTrackingScalingPolicyConfiguration": {
+        "TargetValue": 70.0,
+        "PredefinedMetricSpecification": {
+            "PredefinedMetricType": "DynamoDBReadCapacityUtilization"
+        },
+    }
+}
+step_scaling_config = {
+    "StepScalingPolicyConfiguration": {
+        "AdjustmentType": "ChangeInCapacity",
+        "StepAdjustments": [{"ScalingAdjustment": 10}],
+        "MinAdjustmentMagnitude": 2,
+    }
+}
+
+
+@pytest.mark.parametrize(
+    ["policy_type", "policy_body_kwargs"],
+    [
+        ["TargetTrackingScaling", target_tracking_config],
+        ["TargetTrackingScaling", step_scaling_config],
+    ],
+)
+@application_autoscaling_aws_verified()
+def test_put_scaling_policy_with_unknown_policytype(
+    policy_type,
+    policy_body_kwargs,
+    client=None,
+    namespace=None,
+    resource_id=None,
+    scalable_dimension=None,
+):
+    with pytest.raises(client.exceptions.ValidationException) as exc:
+        client.put_scaling_policy(
+            PolicyName="MyPolicy",
+            ServiceNamespace=namespace,
+            ResourceId=resource_id,
+            ScalableDimension=scalable_dimension,
+            PolicyType="ABCDEFG",
+            **policy_body_kwargs,
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ValidationException"
+    assert (
+        err["Message"]
+        == "1 validation error detected: Value 'ABCDEFG' at 'policyType' failed to satisfy constraint: Member must satisfy enum value set: [PredictiveScaling, StepScaling, TargetTrackingScaling]"
+    )
+
+
+@pytest.mark.parametrize(
+    ["policy_type", "policy_body_kwargs"],
+    [["TargetTrackingScaling", target_tracking_config]],
+)
+@application_autoscaling_aws_verified()
+def test_describe_scaling_policy_alarms(
+    policy_type,
+    policy_body_kwargs,
+    client=None,
+    namespace=None,
+    resource_id=None,
+    scalable_dimension=None,
+):
+    policy_name = "MyPolicy"
+
+    response = client.put_scaling_policy(
+        PolicyName=policy_name,
+        ServiceNamespace=namespace,
+        ResourceId=resource_id,
+        ScalableDimension=scalable_dimension,
+        PolicyType=policy_type,
+        **policy_body_kwargs,
+    )
+    assert len(response["Alarms"]) == 4
+
+    alarms = client.describe_scaling_policies(
+        PolicyNames=[policy_name],
+        ServiceNamespace=namespace,
+        ResourceId=resource_id,
+        ScalableDimension=scalable_dimension,
+    )["ScalingPolicies"][0]["Alarms"]
+    assert len(alarms) == 4
+
+    cloudwatch = boto3.client("cloudwatch", "us-east-1")
+    for alarm in alarms:
+        cw_alarm = cloudwatch.describe_alarms(AlarmNames=[alarm["AlarmName"]])[
+            "MetricAlarms"
+        ][0]
+        assert cw_alarm["AlarmName"] == alarm["AlarmName"]
+        assert cw_alarm["AlarmArn"] == alarm["AlarmARN"]
+
+    client.delete_scaling_policy(
+        PolicyName=policy_name,
+        ServiceNamespace=namespace,
+        ResourceId=resource_id,
+        ScalableDimension=scalable_dimension,
+    )
+
+    for alarm in response["Alarms"]:
+        assert (
+            cloudwatch.describe_alarms(AlarmNames=[alarm["AlarmName"]])["MetricAlarms"]
+            == []
+        )
+
+
+@mock_aws
+def test_describe_scaling_policy_with_ecs_alarms():
+    policy_name = "MyPolicy"
+
+    create_ecs_service("test", "servicename")
+
+    namespace = "ecs"
+    resource_id = "service/test/servicename"
+    scalable_dimension = "ecs:service:DesiredCount"
+
+    client = boto3.client("application-autoscaling", region_name="us-east-1")
+
+    client.register_scalable_target(
+        ServiceNamespace=namespace,
+        ResourceId=resource_id,
+        ScalableDimension=scalable_dimension,
+        MinCapacity=1,
+        MaxCapacity=4,
+    )
+
+    client.put_scaling_policy(
+        PolicyName=policy_name,
+        PolicyType="TargetTrackingScaling",
+        ResourceId=resource_id,
+        ScalableDimension=scalable_dimension,
+        ServiceNamespace=namespace,
+        TargetTrackingScalingPolicyConfiguration={
+            "PredefinedMetricSpecification": {
+                "PredefinedMetricType": "ECSServiceAverageMemoryUtilization",
+            },
+            "ScaleInCooldown": 2,
+            "ScaleOutCooldown": 12,
+            "TargetValue": 6,
+        },
+    )
+
+    alarms = client.describe_scaling_policies(
+        ServiceNamespace="ecs", ResourceId=resource_id
+    )["ScalingPolicies"][0]["Alarms"]
+
+    assert len(alarms) == 2
+
+    cloudwatch = boto3.client("cloudwatch", "us-east-1")
+    for alarm in alarms:
+        cw_alarm = cloudwatch.describe_alarms(AlarmNames=[alarm["AlarmName"]])[
+            "MetricAlarms"
+        ][0]
+        assert f"TargetTracking-{resource_id}" in cw_alarm["AlarmName"]
+        assert {"Name": "ClusterName", "Value": "test"} in cw_alarm["Dimensions"]
+
+
+def create_ecs_service(cluster_name, service_name):
+    client = boto3.client("ecs", region_name="us-east-1")
+    client.create_cluster(clusterName=cluster_name)
+
+    client.register_task_definition(
+        family="test_ecs_task",
+        containerDefinitions=[
+            {
+                "name": "hello_world",
+                "image": "docker/hello-world:latest",
+                "cpu": 1024,
+                "memory": 400,
+                "essential": True,
+                "environment": [
+                    {"name": "AWS_ACCESS_KEY_ID", "value": "SOME_ACCESS_KEY"}
+                ],
+                "logConfiguration": {"logDriver": "json-file"},
+            }
+        ],
+    )
+    client.create_service(
+        cluster=cluster_name,
+        serviceName=service_name,
+        taskDefinition="test_ecs_task",
+        desiredCount=2,
+        platformVersion="2",
+    )
diff --git a/tests/test_cloudwatch/test_cloudwatch_alarms.py b/tests/test_cloudwatch/test_cloudwatch_alarms.py
index 172b6360a..f8fd9d016 100644
--- a/tests/test_cloudwatch/test_cloudwatch_alarms.py
+++ b/tests/test_cloudwatch/test_cloudwatch_alarms.py
@@ -181,7 +181,7 @@ def test_describe_alarms():
     ][0]
 
     assert single_metric_alarm["MetricName"] == "cpu"
-    assert "Metrics" not in single_metric_alarm
+    assert single_metric_alarm["Metrics"] == []
     assert single_metric_alarm["Namespace"] == "blah"
     assert single_metric_alarm["Period"] == 10
     assert single_metric_alarm["EvaluationPeriods"] == 5
diff --git a/tests/test_iot/test_iot_topic_rules.py b/tests/test_iot/test_iot_topic_rules.py
index 89a425e22..f088f9fd4 100644
--- a/tests/test_iot/test_iot_topic_rules.py
+++ b/tests/test_iot/test_iot_topic_rules.py
@@ -4,7 +4,7 @@ from botocore.exceptions import ClientError
 
 from moto import mock_aws
 
-name = "my-rule"
+name = "myrule"
 payload = {
     "sql": "SELECT * FROM 'topic/*' WHERE something > 0",
     "actions": [
@@ -19,6 +19,17 @@ payload = {
 }
 
 
+@mock_aws
+@pytest.mark.parametrize("invalid_name", ["t-r", "t.r", "t r"])
+def test_topic_rule_create_with_invalid_name(invalid_name):
+    client = boto3.client("iot", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_topic_rule(ruleName=invalid_name, topicRulePayload=payload)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidRequestException"
+    assert "Member must satisfy regular expression pattern" in err["Message"]
+
+
 @mock_aws
 def test_topic_rule_create():
     client = boto3.client("iot", region_name="ap-northeast-1")
@@ -41,11 +52,11 @@ def test_topic_rule_list():
     assert len(res["rules"]) == 0
 
     client.create_topic_rule(ruleName=name, topicRulePayload=payload)
-    client.create_topic_rule(ruleName="my-rule-2", topicRulePayload=payload)
+    client.create_topic_rule(ruleName=f"{name}2", topicRulePayload=payload)
 
     res = client.list_topic_rules()
     assert len(res["rules"]) == 2
-    for rule, rule_name in zip(res["rules"], [name, "my-rule-2"]):
+    for rule, rule_name in zip(res["rules"], [name, f"{name}2"]):
         assert rule["ruleName"] == rule_name
         assert rule["createdAt"] is not None
         assert rule["ruleArn"] is not None
diff --git a/tests/test_iotdata/test_iotdata.py b/tests/test_iotdata/test_iotdata.py
index 60e58c081..8f369d4d0 100644
--- a/tests/test_iotdata/test_iotdata.py
+++ b/tests/test_iotdata/test_iotdata.py
@@ -15,7 +15,7 @@ boto3_version = sys.modules["botocore"].__version__
 
 
 @mock_aws
-def test_basic():
+def test_basic() -> None:
     iot_client = boto3.client("iot", region_name="ap-northeast-1")
     client = boto3.client("iot-data", region_name="ap-northeast-1")
     name = "my-thing"
@@ -36,7 +36,7 @@ def test_basic():
 
     res = client.get_thing_shadow(thingName=name)
     payload = json.loads(res["payload"].read())
-    expected_state = b'{"desired": {"led": "on"}, "delta": {"led": "on"}}'
+    expected_state = '{"desired": {"led": "on"}, "delta": {"led": "on"}}'
     assert payload["state"] == json.loads(expected_state)
     assert "led" in payload["metadata"]["desired"]
     assert payload["version"] == 1
@@ -48,7 +48,7 @@ def test_basic():
 
 
 @mock_aws
-def test_update():
+def test_update() -> None:
     iot_client = boto3.client("iot", region_name="ap-northeast-1")
     client = boto3.client("iot-data", region_name="ap-northeast-1")
     name = "my-thing"
@@ -66,7 +66,7 @@ def test_update():
 
     res = client.get_thing_shadow(thingName=name)
     payload = json.loads(res["payload"].read())
-    expected_state = b'{"desired": {"led": "on"}, "delta": {"led": "on"}}'
+    expected_state = '{"desired": {"led": "on"}, "delta": {"led": "on"}}'
     assert payload["state"] == json.loads(expected_state)
     assert "led" in payload["metadata"]["desired"]
     assert payload["version"] == 1
@@ -84,7 +84,7 @@ def test_update():
 
     res = client.get_thing_shadow(thingName=name)
     payload = json.loads(res["payload"].read())
-    expected_state = b'{"desired": {"led": "on"}, "reported": {"led": "on"}}'
+    expected_state = '{"desired": {"led": "on"}, "reported": {"led": "on"}}'
     assert payload["state"] == json.loads(expected_state)
     assert "led" in payload["metadata"]["desired"]
     assert payload["version"] == 2
@@ -98,7 +98,7 @@ def test_update():
 
 
 @mock_aws
-def test_create_named_shadows():
+def test_create_named_shadows() -> None:
     if LooseVersion(boto3_version) < LooseVersion("1.29.0"):
         raise SkipTest("Parameter only available in newer versions")
     iot_client = boto3.client("iot", region_name="ap-northeast-1")
@@ -154,23 +154,25 @@ def test_create_named_shadows():
 
 
 @mock_aws
-def test_publish():
+def test_publish() -> None:
     region_name = "ap-northeast-1"
     client = boto3.client("iot-data", region_name=region_name)
     client.publish(topic="test/topic1", qos=1, payload=b"pl1")
     client.publish(topic="test/topic2", qos=1, payload=b"pl2")
     client.publish(topic="test/topic3", qos=1, payload=b"\xbf")
+    client.publish(topic="test/topic4", qos=1, payload="string")
 
     if not settings.TEST_SERVER_MODE:
         mock_backend = moto.iotdata.models.iotdata_backends[ACCOUNT_ID][region_name]
-        assert len(mock_backend.published_payloads) == 3
+        assert len(mock_backend.published_payloads) == 4
         assert ("test/topic1", b"pl1") in mock_backend.published_payloads
         assert ("test/topic2", b"pl2") in mock_backend.published_payloads
         assert ("test/topic3", b"\xbf") in mock_backend.published_payloads
+        assert ("test/topic4", b"string") in mock_backend.published_payloads
 
 
 @mock_aws
-def test_delete_field_from_device_shadow():
+def test_delete_field_from_device_shadow() -> None:
     test_thing_name = "TestThing"
 
     iot_raw_client = boto3.client("iot", region_name="eu-central-1")
diff --git a/tests/test_iotdata/test_server.py b/tests/test_iotdata/test_server.py
index a6a9d595c..c0085d902 100644
--- a/tests/test_iotdata/test_server.py
+++ b/tests/test_iotdata/test_server.py
@@ -11,7 +11,7 @@ Test the different server responses
 
 
 @mock_aws
-def test_iotdata_list():
+def test_iotdata_list() -> None:
     backend = server.create_backend_app("iot-data")
     test_client = backend.test_client()
 
@@ -29,7 +29,7 @@ def test_iotdata_list():
     ],
 )
 @mock_aws
-def test_publish(url_encode_topic):
+def test_publish(url_encode_topic: bool) -> None:
     backend = server.create_backend_app("iot-data")
     test_client = backend.test_client()
 
diff --git a/tests/test_stepfunctions/parser/__init__.py b/tests/test_stepfunctions/parser/__init__.py
index eb8c59407..34a513126 100644
--- a/tests/test_stepfunctions/parser/__init__.py
+++ b/tests/test_stepfunctions/parser/__init__.py
@@ -94,15 +94,20 @@ def verify_execution_result(
 
 
 def _start_execution(client, definition, exec_input, sfn_role):
-    name = "sfn_name"
+    name = f"sfn_name_{str(uuid4())[0:6]}"
     #
     response = client.create_state_machine(
         name=name, definition=json.dumps(definition), roleArn=sfn_role
     )
     state_machine_arn = response["stateMachineArn"]
-    execution = client.start_execution(
-        name="exec1", stateMachineArn=state_machine_arn, input=exec_input or "{}"
-    )
+    if exec_input:
+        execution = client.start_execution(
+            name="exec1", stateMachineArn=state_machine_arn, input=exec_input or "{}"
+        )
+    else:
+        execution = client.start_execution(
+            name="exec1", stateMachineArn=state_machine_arn
+        )
     execution_arn = execution["executionArn"]
     return execution_arn, state_machine_arn
 
diff --git a/tests/test_stepfunctions/parser/templates/failure.json b/tests/test_stepfunctions/parser/templates/failure.json
new file mode 100644
index 000000000..45a41d23f
--- /dev/null
+++ b/tests/test_stepfunctions/parser/templates/failure.json
@@ -0,0 +1,11 @@
+{
+    "Comment": "An example of the Amazon States Language using a choice state.",
+    "StartAt": "DefaultState",
+    "States": {
+        "DefaultState": {
+            "Type": "Fail",
+            "Error": "DefaultStateError",
+            "Cause": "No Matches!"
+        }
+    }
+}
\ No newline at end of file
diff --git a/tests/test_stepfunctions/parser/test_stepfunctions.py b/tests/test_stepfunctions/parser/test_stepfunctions.py
index 2a09a4785..c0eec65ea 100644
--- a/tests/test_stepfunctions/parser/test_stepfunctions.py
+++ b/tests/test_stepfunctions/parser/test_stepfunctions.py
@@ -1,72 +1,59 @@
 import json
-from time import sleep
 from unittest import SkipTest
 
-import boto3
+import pytest
 
-from moto import mock_aws, settings
-from tests import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+from moto import settings
 
-failure_definition = {
-    "Comment": "An example of the Amazon States Language using a choice state.",
-    "StartAt": "DefaultState",
-    "States": {
-        "DefaultState": {
-            "Type": "Fail",
-            "Error": "DefaultStateError",
-            "Cause": "No Matches!",
-        }
-    },
-}
+from . import aws_verified, verify_execution_result
 
 
-@mock_aws(config={"stepfunctions": {"execute_state_machine": True}})
+@aws_verified
+@pytest.mark.aws_verified
 def test_state_machine_with_simple_failure_state():
     if settings.TEST_SERVER_MODE:
         raise SkipTest("Don't need to test this in ServerMode")
-    client = boto3.client("stepfunctions", region_name="us-east-1")
-    execution_arn, state_machine_arn = _start_execution(client, failure_definition)
 
-    for _ in range(5):
-        execution = client.describe_execution(executionArn=execution_arn)
-        if execution["status"] == "FAILED":
-            assert "stopDate" in execution
-            assert execution["error"] == "DefaultStateError"
-            assert execution["cause"] == "No Matches!"
-
-            history = client.get_execution_history(executionArn=execution_arn)
-            assert len(history["events"]) == 3
-            assert [e["type"] for e in history["events"]] == [
-                "ExecutionStarted",
-                "FailStateEntered",
-                "ExecutionFailed",
-            ]
-
-            executions = client.list_executions(stateMachineArn=state_machine_arn)[
-                "executions"
-            ]
-            assert len(executions) == 1
-            assert executions[0]["executionArn"] == execution_arn
-            assert executions[0]["stateMachineArn"] == state_machine_arn
-            assert executions[0]["status"] == "FAILED"
+    def _verify_result(client, execution, execution_arn):
+        assert "stopDate" in execution
+        assert execution["error"] == "DefaultStateError"
+        assert execution["cause"] == "No Matches!"
+        assert execution["input"] == "{}"
+
+        history = client.get_execution_history(executionArn=execution_arn)
+        assert len(history["events"]) == 3
+        assert [e["type"] for e in history["events"]] == [
+            "ExecutionStarted",
+            "FailStateEntered",
+            "ExecutionFailed",
+        ]
+
+        state_machine_arn = execution["stateMachineArn"]
+        executions = client.list_executions(stateMachineArn=state_machine_arn)[
+            "executions"
+        ]
+        assert len(executions) == 1
+        assert executions[0]["executionArn"] == execution_arn
+        assert executions[0]["stateMachineArn"] == state_machine_arn
+        assert executions[0]["status"] == "FAILED"
+
+    verify_execution_result(_verify_result, "FAILED", "failure")
+
+
+@aws_verified
+@pytest.mark.aws_verified
+def test_state_machine_with_input():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest("Don't need to test this in ServerMode")
 
-            break
-        sleep(0.2)
-    else:
-        assert False, "Should have failed already"
+    _input = {"my": "data"}
 
+    def _verify_result(client, execution, execution_arn):
+        assert execution["input"] == json.dumps(_input)
 
-def _start_execution(client, definition):
-    name = "sfn_name"
-    #
-    response = client.create_state_machine(
-        name=name, definition=json.dumps(definition), roleArn=_get_default_role()
+    verify_execution_result(
+        _verify_result,
+        "FAILED",
+        "failure",
+        exec_input=json.dumps(_input),
     )
-    state_machine_arn = response["stateMachineArn"]
-    execution = client.start_execution(name="exec1", stateMachineArn=state_machine_arn)
-    execution_arn = execution["executionArn"]
-    return execution_arn, state_machine_arn
-
-
-def _get_default_role():
-    return "arn:aws:iam::" + ACCOUNT_ID + ":role/unknown_sf_role"
diff --git a/tests/test_stepfunctions/test_stepfunctions.py b/tests/test_stepfunctions/test_stepfunctions.py
index afeb80cf4..f7e758331 100644
--- a/tests/test_stepfunctions/test_stepfunctions.py
+++ b/tests/test_stepfunctions/test_stepfunctions.py
@@ -544,9 +544,9 @@ def test_state_machine_start_execution_with_invalid_input():
         name="name", definition=str(simple_definition), roleArn=_get_default_role()
     )
     with pytest.raises(ClientError):
-        _ = client.start_execution(stateMachineArn=sm["stateMachineArn"], input="")
+        client.start_execution(stateMachineArn=sm["stateMachineArn"], input="")
     with pytest.raises(ClientError):
-        _ = client.start_execution(stateMachineArn=sm["stateMachineArn"], input="{")
+        client.start_execution(stateMachineArn=sm["stateMachineArn"], input="{")
 
 
 @mock_aws
@@ -657,7 +657,7 @@ def test_state_machine_describe_execution_with_no_input():
     #
     assert description["ResponseMetadata"]["HTTPStatusCode"] == 200
     assert description["executionArn"] == execution["executionArn"]
-    assert description["input"] == '"{}"'
+    assert description["input"] == "{}"
     assert re.match("[-0-9a-z]+", description["name"])
     assert description["startDate"] == execution["startDate"]
     assert description["stateMachineArn"] == sm["stateMachineArn"]
@@ -680,7 +680,7 @@ def test_state_machine_describe_execution_with_custom_input():
     #
     assert description["ResponseMetadata"]["HTTPStatusCode"] == 200
     assert description["executionArn"] == execution["executionArn"]
-    assert json.loads(description["input"]) == execution_input
+    assert description["input"] == execution_input
     assert re.match("[-a-z0-9]+", description["name"])
     assert description["startDate"] == execution["startDate"]
     assert description["stateMachineArn"] == sm["stateMachineArn"]