Repository: getmoto/moto
Pull request #7380 (2024-02-21 12:59:02)

DynamoDB: fix: support top level lists for Dynamo Item to_json
The tests in this example (prior to the fix) reproduce an issue where the exception `TypeError: Object of type DynamoType is not JSON serializable` is raised when `Dynamo.Item.to_json` is passed a top level list.

I have chosen to test this via a condition expression with `ReturnValuesOnConditionCheckFailure="ALL_OLD"` as this is a very close reproduction to the issue I faced.
Some additional tests for top level string sets and lists in dictionaries were added to ensure different paths through the new and existing code are handled correctly.

Also fixes issue with `UPDATED_NEW` handling for lists:
- Before this change: when `update` is called to change one value in a list, `_build_updated_new_attributes` attempts to return the list with the value at the position of the changed element as an empty dictionary (which leads to a deserialization error)
- With this change: when `update` is called to change one value in a list, `_build_updated_new_attributes` returns the whole changed list if any element in the original list differs from the changed list - otherwise the whole original list is returned.

verysamuel: @bblommers Having a difficult time understanding why [the test called `test_dynamodb/test_dynamodb.py::test_update_return_updated_new_attributes_when_same` failed](https://github.com/getmoto/moto/actions/runs/8001300362/job/21855180785#step:9:2065) when it seems unrelated to this PR.

That being said, it seems to point to an existing issue in how lists are handled with `UPDATED_NEW`. As such, I've included a fix for it in this PR. I've updated the description to reflect that.

Patch:
diff --git a/docs/docs/services/cf.rst b/docs/docs/services/cf.rst
index b3e227189..86d7797bc 100644
--- a/docs/docs/services/cf.rst
+++ b/docs/docs/services/cf.rst
@@ -121,6 +121,8 @@ Please let us know if you'd like support for a resource not yet listed here.
   +---------------------------------------+--------+--------+--------+ - [ ] Id                               |
   |                                       |        |        |        |                                        |
   +---------------------------------------+--------+--------+--------+----------------------------------------+
+  |AWS::EMR::Cluster                      |    x   |        |        | - [ ] MasterPublicDNS                  |
+  +---------------------------------------+--------+--------+--------+----------------------------------------+
   |AWS::Events::Archive                   |    x   |    x   |        | - [x] Arn                              |
   +---------------------------------------+--------+--------+--------+----------------------------------------+
   |AWS::Events::EventBus                  |    x   |    x   |    x   | - [x] Arn                              |
diff --git a/moto/autoscaling/models.py b/moto/autoscaling/models.py
index e835e71fa..2f27e3610 100644
--- a/moto/autoscaling/models.py
+++ b/moto/autoscaling/models.py
@@ -1190,13 +1190,40 @@ class AutoScalingBackend(BaseBackend):
         return group
 
     def describe_auto_scaling_groups(
-        self, names: List[str]
+        self, names: List[str], filters: Optional[List[Dict[str, str]]] = None
     ) -> List[FakeAutoScalingGroup]:
-        groups = self.autoscaling_groups.values()
+
+        groups = list(self.autoscaling_groups.values())
+
+        if filters:
+            for f in filters:
+                if f["Name"] == "tag-key":
+                    groups = [
+                        group
+                        for group in groups
+                        if any(tag["Key"] in f["Values"] for tag in group.tags)
+                    ]
+                elif f["Name"] == "tag-value":
+                    groups = [
+                        group
+                        for group in groups
+                        if any(tag["Value"] in f["Values"] for tag in group.tags)
+                    ]
+                elif f["Name"].startswith("tag:"):
+                    tag_key = f["Name"][4:]
+                    groups = [
+                        group
+                        for group in groups
+                        if any(
+                            tag["Key"] == tag_key and tag["Value"] in f["Values"]
+                            for tag in group.tags
+                        )
+                    ]
+
         if names:
-            return [group for group in groups if group.name in names]
-        else:
-            return list(groups)
+            groups = [group for group in groups if group.name in names]
+
+        return groups
 
     def delete_auto_scaling_group(self, group_name: str) -> None:
         self.set_desired_capacity(group_name, 0)
diff --git a/moto/autoscaling/responses.py b/moto/autoscaling/responses.py
index d83ca039e..e9aa149ca 100644
--- a/moto/autoscaling/responses.py
+++ b/moto/autoscaling/responses.py
@@ -228,7 +228,10 @@ class AutoScalingResponse(BaseResponse):
     def describe_auto_scaling_groups(self) -> str:
         names = self._get_multi_param("AutoScalingGroupNames.member")
         token = self._get_param("NextToken")
-        all_groups = self.autoscaling_backend.describe_auto_scaling_groups(names)
+        filters = self._get_params().get("Filters", [])
+        all_groups = self.autoscaling_backend.describe_auto_scaling_groups(
+            names, filters=filters
+        )
         all_names = [group.name for group in all_groups]
         if token:
             start = all_names.index(token) + 1
diff --git a/moto/cloudformation/parsing.py b/moto/cloudformation/parsing.py
index 8b0eaa608..239cc886d 100644
--- a/moto/cloudformation/parsing.py
+++ b/moto/cloudformation/parsing.py
@@ -33,8 +33,6 @@ from moto.awslambda import models as lambda_models  # noqa  # pylint: disable=al
 from moto.batch import models as batch_models  # noqa  # pylint: disable=all
 from moto.cloudformation.custom_model import CustomModel
 from moto.cloudwatch import models as cw_models  # noqa  # pylint: disable=all
-
-# End ugly list of imports
 from moto.core.common_models import CloudFormationModel
 from moto.datapipeline import models as data_models  # noqa  # pylint: disable=all
 from moto.dynamodb import models as ddb_models  # noqa  # pylint: disable=all
@@ -45,6 +43,7 @@ from moto.ecs import models as ecs_models  # noqa  # pylint: disable=all
 from moto.efs import models as efs_models  # noqa  # pylint: disable=all
 from moto.elb import models as elb_models  # noqa  # pylint: disable=all
 from moto.elbv2 import models as elbv2_models  # noqa  # pylint: disable=all
+from moto.emr import models as emr_models  # noqa  # pylint: disable=all
 from moto.events import models as events_models  # noqa  # pylint: disable=all
 from moto.iam import models as iam_models  # noqa  # pylint: disable=all
 from moto.kinesis import models as kinesis_models  # noqa  # pylint: disable=all
@@ -62,6 +61,7 @@ from moto.ssm import models as ssm_models  # noqa  # pylint: disable=all
 from moto.ssm import ssm_backends
 from moto.stepfunctions import models as sfn_models  # noqa  # pylint: disable=all
 
+# End ugly list of imports
 from .exceptions import (
     ExportNotFound,
     MissingParameterError,
diff --git a/moto/cognitoidp/models.py b/moto/cognitoidp/models.py
index 15416d4e7..0e960764e 100644
--- a/moto/cognitoidp/models.py
+++ b/moto/cognitoidp/models.py
@@ -1466,6 +1466,16 @@ class CognitoIdpBackend(BaseBackend):
                     "Session": session,
                 }
 
+            if user.sms_mfa_enabled and user.preferred_mfa_setting == "SMS_MFA":
+                session = str(random.uuid4())
+                self.sessions[session] = user_pool
+
+                return {
+                    "ChallengeName": "SMS_MFA",
+                    "ChallengeParameters": {},
+                    "Session": session,
+                }
+
             return self._log_user_in(user_pool, client, username)
         elif auth_flow in (AuthFlow.REFRESH_TOKEN, AuthFlow.REFRESH_TOKEN_AUTH):
             refresh_token: str = auth_parameters.get("REFRESH_TOKEN")  # type: ignore[assignment]
@@ -1578,13 +1588,13 @@ class CognitoIdpBackend(BaseBackend):
 
             del self.sessions[session]
             return self._log_user_in(user_pool, client, username)
-        elif challenge_name == "SOFTWARE_TOKEN_MFA":
+        elif challenge_name == "SOFTWARE_TOKEN_MFA" or challenge_name == "SMS_MFA":
             username: str = challenge_responses.get("USERNAME")  # type: ignore[no-redef]
             self.admin_get_user(user_pool.id, username)
 
-            software_token_mfa_code = challenge_responses.get("SOFTWARE_TOKEN_MFA_CODE")
-            if not software_token_mfa_code:
-                raise ResourceNotFoundError(software_token_mfa_code)
+            mfa_code = challenge_responses.get(f"{challenge_name}_CODE")
+            if not mfa_code:
+                raise ResourceNotFoundError(mfa_code)
 
             if client.generate_secret:
                 secret_hash = challenge_responses.get("SECRET_HASH")
diff --git a/moto/dynamodb/models/__init__.py b/moto/dynamodb/models/__init__.py
index c3be0ff5d..0627a238e 100644
--- a/moto/dynamodb/models/__init__.py
+++ b/moto/dynamodb/models/__init__.py
@@ -579,7 +579,11 @@ class DynamoDBBackend(BaseBackend):
                     item = item["Put"]
                     attrs = item["Item"]
                     table_name = item["TableName"]
-                    check_unicity(table_name, item)
+                    table = self.get_table(table_name)
+                    key = {table.hash_key_attr: attrs[table.hash_key_attr]}
+                    if table.range_key_attr is not None:
+                        key[table.range_key_attr] = attrs[table.range_key_attr]
+                    check_unicity(table_name, key)
                     condition_expression = item.get("ConditionExpression", None)
                     expression_attribute_names = item.get(
                         "ExpressionAttributeNames", None
diff --git a/moto/dynamodb/models/dynamo_type.py b/moto/dynamodb/models/dynamo_type.py
index 28490526d..0adbe95f5 100644
--- a/moto/dynamodb/models/dynamo_type.py
+++ b/moto/dynamodb/models/dynamo_type.py
@@ -316,14 +316,20 @@ class Item(BaseModel):
         return sum(bytesize(key) + value.size() for key, value in self.attrs.items())
 
     def to_json(self) -> Dict[str, Any]:
-        attributes = {}
+        attributes: Dict[str, Any] = {}
         for attribute_key, attribute in self.attrs.items():
             if isinstance(attribute.value, dict):
-                attr_value = {
+                attr_dict_value = {
                     key: value.to_regular_json()
                     for key, value in attribute.value.items()
                 }
-                attributes[attribute_key] = {attribute.type: attr_value}
+                attributes[attribute_key] = {attribute.type: attr_dict_value}
+            elif isinstance(attribute.value, list):
+                attr_list_value = [
+                    value.to_regular_json() if isinstance(value, DynamoType) else value
+                    for value in attribute.value
+                ]
+                attributes[attribute_key] = {attribute.type: attr_list_value}
             else:
                 attributes[attribute_key] = {attribute.type: attribute.value}
 
diff --git a/moto/dynamodb/responses.py b/moto/dynamodb/responses.py
index 03aeea2f3..9f03efa2c 100644
--- a/moto/dynamodb/responses.py
+++ b/moto/dynamodb/responses.py
@@ -964,12 +964,11 @@ class DynamoHandler(BaseResponse):
                 if len(changed) != len(original):
                     return changed
                 else:
-                    return [
-                        self._build_updated_new_attributes(
-                            original[index], changed[index]
-                        )
+                    any_element_has_changed = any(
+                        changed[index] != original[index]
                         for index in range(len(changed))
-                    ]
+                    )
+                    return changed if any_element_has_changed else original
             else:
                 return changed
 
diff --git a/moto/emr/models.py b/moto/emr/models.py
index fa0d7100e..58f65613c 100644
--- a/moto/emr/models.py
+++ b/moto/emr/models.py
@@ -4,7 +4,7 @@ from typing import Any, Dict, List, Optional, Tuple
 from dateutil.parser import parse as dtparse
 
 from moto.core.base_backend import BackendDict, BaseBackend
-from moto.core.common_models import BaseModel
+from moto.core.common_models import BaseModel, CloudFormationModel
 from moto.emr.exceptions import (
     InvalidRequestException,
     ResourceNotFoundException,
@@ -151,7 +151,7 @@ class FakeStep(BaseModel):
         self.start_datetime = datetime.now(timezone.utc)
 
 
-class FakeCluster(BaseModel):
+class FakeCluster(CloudFormationModel):
     def __init__(
         self,
         emr_backend: "ElasticMapReduceBackend",
@@ -384,6 +384,71 @@ class FakeCluster(BaseModel):
     def set_visibility(self, visibility: str) -> None:
         self.visible_to_all_users = visibility
 
+    @property
+    def physical_resource_id(self) -> str:
+        return self.id
+
+    @staticmethod
+    def cloudformation_type() -> str:
+        return "AWS::EMR::Cluster"
+
+    @classmethod
+    def has_cfn_attr(cls, attr: str) -> bool:
+        return attr in ["Id"]
+
+    def get_cfn_attribute(self, attribute_name: str) -> str:
+        from moto.cloudformation.exceptions import UnformattedGetAttTemplateException
+
+        if attribute_name == "Id":
+            return self.id
+        raise UnformattedGetAttTemplateException()
+
+    @classmethod
+    def create_from_cloudformation_json(  # type: ignore[misc]
+        cls,
+        resource_name: str,
+        cloudformation_json: Any,
+        account_id: str,
+        region_name: str,
+        **kwargs: Any,
+    ) -> "FakeCluster":
+
+        properties = cloudformation_json["Properties"]
+
+        instance_attrs = properties.get("Instances", {})
+        instance_attrs["ec2_subnet_id"] = instance_attrs.get("Ec2SubnetId")
+        instance_attrs["emr_managed_master_security_group"] = instance_attrs.get(
+            "EmrManagedMasterSecurityGroup"
+        )
+        instance_attrs["emr_managed_slave_security_group"] = instance_attrs.get(
+            "EmrManagedSlaveSecurityGroup"
+        )
+        instance_attrs["service_access_security_group"] = instance_attrs.get(
+            "ServiceAccessSecurityGroup"
+        )
+        instance_attrs["additional_master_security_groups"] = instance_attrs.get(
+            "AdditionalMasterSecurityGroups", []
+        )
+        instance_attrs["additional_slave_security_groups"] = instance_attrs.get(
+            "AdditionalSlaveSecurityGroups", []
+        )
+
+        emr_backend: ElasticMapReduceBackend = emr_backends[account_id][region_name]
+        cluster = emr_backend.run_job_flow(
+            name=properties["Name"],
+            log_uri=properties.get("LogUri"),
+            job_flow_role=properties["JobFlowRole"],
+            service_role=properties["ServiceRole"],
+            steps=[],
+            instance_attrs=instance_attrs,
+            kerberos_attributes=properties.get("KerberosAttributes", {}),
+            release_label=properties.get("ReleaseLabel"),
+            custom_ami_id=properties.get("CustomAmiId"),
+        )
+        tags = {item["Key"]: item["Value"] for item in properties.get("Tags", [])}
+        cluster.add_tags(tags)
+        return cluster
+
 
 class FakeSecurityConfiguration(BaseModel):
     def __init__(self, name: str, security_configuration: str):
diff --git a/moto/kms/models.py b/moto/kms/models.py
index f567a9c58..d12d9834a 100644
--- a/moto/kms/models.py
+++ b/moto/kms/models.py
@@ -233,11 +233,11 @@ class Key(CloudFormationModel):
             policy=properties["KeyPolicy"],
             key_usage="ENCRYPT_DECRYPT",
             key_spec="SYMMETRIC_DEFAULT",
-            description=properties["Description"],
+            description=properties.get("Description"),
             tags=properties.get("Tags", []),
         )
-        key.key_rotation_status = properties["EnableKeyRotation"]
-        key.enabled = properties["Enabled"]
+        key.key_rotation_status = properties.get("EnableKeyRotation", False)
+        key.enabled = properties.get("Enabled", True)
 
         return key
 


Test patch:
diff --git a/tests/test_autoscaling/test_autoscaling_group_filters.py b/tests/test_autoscaling/test_autoscaling_group_filters.py
new file mode 100644
index 000000000..914c23238
--- /dev/null
+++ b/tests/test_autoscaling/test_autoscaling_group_filters.py
@@ -0,0 +1,116 @@
+import boto3
+
+from moto import mock_aws
+from tests import EXAMPLE_AMI_ID
+
+from .utils import setup_networking
+
+
+@mock_aws
+def test_describe_autoscaling_groups_filter_by_tag_key():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_auto_scaling_groups(
+        Filters=[{"Name": "tag-key", "Values": ["test_key1"]}]
+    )
+    group = response["AutoScalingGroups"][0]
+    tags = group["Tags"]
+
+    assert len(response["AutoScalingGroups"]) == 1
+    assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
+    assert {
+        "Key": "test_key1",
+        "PropagateAtLaunch": False,
+        "ResourceId": "test_asg1",
+        "ResourceType": "auto-scaling-group",
+        "Value": "test_value1",
+    } in tags
+
+
+@mock_aws
+def test_describe_autoscaling_groups_filter_by_tag_value():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_auto_scaling_groups(
+        Filters=[{"Name": "tag-value", "Values": ["test_value1"]}]
+    )
+    group = response["AutoScalingGroups"][0]
+    tags = group["Tags"]
+
+    assert len(response["AutoScalingGroups"]) == 1
+    assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
+    assert {
+        "Key": "test_key1",
+        "PropagateAtLaunch": False,
+        "ResourceId": "test_asg1",
+        "ResourceType": "auto-scaling-group",
+        "Value": "test_value1",
+    } in tags
+
+
+@mock_aws
+def test_describe_autoscaling_groups_filter_by_tag_key_value():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_auto_scaling_groups(
+        Filters=[{"Name": "tag:test_key1", "Values": ["test_value1"]}]
+    )
+    group = response["AutoScalingGroups"][0]
+    tags = group["Tags"]
+
+    assert len(response["AutoScalingGroups"]) == 1
+    assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
+    assert {
+        "Key": "test_key1",
+        "PropagateAtLaunch": False,
+        "ResourceId": "test_asg1",
+        "ResourceType": "auto-scaling-group",
+        "Value": "test_value1",
+    } in tags
+
+
+@mock_aws
+def test_describe_autoscaling_groups_no_filter():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_auto_scaling_groups()
+
+    assert len(response["AutoScalingGroups"]) == 2
+    assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
+    assert response["AutoScalingGroups"][1]["AutoScalingGroupName"] == "test_asg2"
+
+
+def create_asgs(client, subnet):
+    _ = client.create_launch_configuration(
+        LaunchConfigurationName="test_launch_configuration",
+        ImageId=EXAMPLE_AMI_ID,
+        InstanceType="t2.medium",
+    )
+    client.create_auto_scaling_group(
+        AutoScalingGroupName="test_asg1",
+        LaunchConfigurationName="test_launch_configuration",
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        Tags=[
+            {"Key": "test_key1", "Value": "test_value1"},
+            {"Key": "test_key2", "Value": "test_value2"},
+        ],
+        VPCZoneIdentifier=subnet,
+    )
+    client.create_auto_scaling_group(
+        AutoScalingGroupName="test_asg2",
+        LaunchConfigurationName="test_launch_configuration",
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        VPCZoneIdentifier=subnet,
+    )
diff --git a/tests/test_cognitoidp/test_cognitoidp.py b/tests/test_cognitoidp/test_cognitoidp.py
index 8867282e4..1ff151bfb 100644
--- a/tests/test_cognitoidp/test_cognitoidp.py
+++ b/tests/test_cognitoidp/test_cognitoidp.py
@@ -4454,6 +4454,56 @@ def test_admin_initiate_auth_when_token_totp_enabled():
     assert result["AuthenticationResult"]["TokenType"] == "Bearer"
 
 
+@mock_aws
+def test_admin_initiate_auth_when_sms_mfa_enabled():
+    conn = boto3.client("cognito-idp", "us-west-2")
+
+    result = authentication_flow(conn, "ADMIN_NO_SRP_AUTH")
+    user_pool_id = result["user_pool_id"]
+    username = result["username"]
+    client_id = result["client_id"]
+    password = result["password"]
+
+    # Set MFA SMS methods
+    conn.admin_set_user_mfa_preference(
+        Username=username,
+        UserPoolId=user_pool_id,
+        SMSMfaSettings={"Enabled": True, "PreferredMfa": True},
+    )
+    result = conn.admin_get_user(UserPoolId=user_pool_id, Username=username)
+    assert len(result["UserMFASettingList"]) == 1
+    assert result["PreferredMfaSetting"] == "SMS_MFA"
+
+    result = conn.admin_initiate_auth(
+        UserPoolId=user_pool_id,
+        ClientId=client_id,
+        AuthFlow="ADMIN_NO_SRP_AUTH",
+        AuthParameters={
+            "USERNAME": username,
+            "PASSWORD": password,
+        },
+    )
+
+    assert result["ChallengeName"] == "SMS_MFA"
+    assert result["Session"] != ""
+
+    result = conn.admin_respond_to_auth_challenge(
+        UserPoolId=user_pool_id,
+        ClientId=client_id,
+        ChallengeName="SMS_MFA",
+        Session=result["Session"],
+        ChallengeResponses={
+            "SMS_MFA_CODE": "123456",
+            "USERNAME": username,
+        },
+    )
+
+    assert result["AuthenticationResult"]["IdToken"] != ""
+    assert result["AuthenticationResult"]["AccessToken"] != ""
+    assert result["AuthenticationResult"]["RefreshToken"] != ""
+    assert result["AuthenticationResult"]["TokenType"] == "Bearer"
+
+
 @mock_aws
 def test_admin_setting_mfa_when_token_not_verified():
     conn = boto3.client("cognito-idp", "us-west-2")
diff --git a/tests/test_dynamodb/exceptions/test_dynamodb_transactions.py b/tests/test_dynamodb/exceptions/test_dynamodb_transactions.py
index 61f2c59dd..212739388 100644
--- a/tests/test_dynamodb/exceptions/test_dynamodb_transactions.py
+++ b/tests/test_dynamodb/exceptions/test_dynamodb_transactions.py
@@ -46,6 +46,55 @@ def test_multiple_transactions_on_same_item():
     )
 
 
+@mock_aws
+def test_transact_write_items__put_and_delete_on_same_item():
+    schema = {
+        "KeySchema": [
+            {"AttributeName": "pk", "KeyType": "HASH"},
+            {"AttributeName": "sk", "KeyType": "RANGE"},
+        ],
+        "AttributeDefinitions": [
+            {"AttributeName": "pk", "AttributeType": "S"},
+            {"AttributeName": "sk", "AttributeType": "S"},
+        ],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **schema
+    )
+
+    with pytest.raises(ClientError) as exc:
+        dynamodb.transact_write_items(
+            TransactItems=[
+                {
+                    "Put": {
+                        "TableName": "test-table",
+                        "Item": {
+                            "pk": {"S": "test-pk"},
+                            "sk": {"S": "test-sk"},
+                            "field": {"S": "test-field"},
+                        },
+                    }
+                },
+                {
+                    "Delete": {
+                        "TableName": "test-table",
+                        "Key": {
+                            "pk": {"S": "test-pk"},
+                            "sk": {"S": "test-sk"},
+                        },
+                    }
+                },
+            ]
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ValidationException"
+    assert (
+        err["Message"]
+        == "Transaction request cannot include multiple operations on one item"
+    )
+
+
 @mock_aws
 def test_transact_write_items__too_many_transactions():
     schema = {
diff --git a/tests/test_dynamodb/test_dynamodb.py b/tests/test_dynamodb/test_dynamodb.py
index 26d5e0545..808d48d2f 100644
--- a/tests/test_dynamodb/test_dynamodb.py
+++ b/tests/test_dynamodb/test_dynamodb.py
@@ -3856,24 +3856,27 @@ def test_transact_write_items_conditioncheck_passes():
     dynamodb.create_table(
         TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
     )
-    # Insert an item without email address
-    dynamodb.put_item(TableName="test-table", Item={"id": {"S": "foo"}})
-    # Put an email address, after verifying it doesn't exist yet
+    # Insert an item with email address
+    dynamodb.put_item(
+        TableName="test-table",
+        Item={"id": {"S": "foo"}, "email_address": {"S": "foo@moto.com"}},
+    )
+    # Put a new item, after verifying the exising item also has email address
     dynamodb.transact_write_items(
         TransactItems=[
             {
                 "ConditionCheck": {
                     "Key": {"id": {"S": "foo"}},
                     "TableName": "test-table",
-                    "ConditionExpression": "attribute_not_exists(#e)",
+                    "ConditionExpression": "attribute_exists(#e)",
                     "ExpressionAttributeNames": {"#e": "email_address"},
                 }
             },
             {
                 "Put": {
                     "Item": {
-                        "id": {"S": "foo"},
-                        "email_address": {"S": "test@moto.com"},
+                        "id": {"S": "bar"},
+                        "email_address": {"S": "bar@moto.com"},
                     },
                     "TableName": "test-table",
                 }
@@ -3882,8 +3885,9 @@ def test_transact_write_items_conditioncheck_passes():
     )
     # Assert all are present
     items = dynamodb.scan(TableName="test-table")["Items"]
-    assert len(items) == 1
-    assert items[0] == {"email_address": {"S": "test@moto.com"}, "id": {"S": "foo"}}
+    assert len(items) == 2
+    assert items[0] == {"email_address": {"S": "foo@moto.com"}, "id": {"S": "foo"}}
+    assert items[1] == {"email_address": {"S": "bar@moto.com"}, "id": {"S": "bar"}}
 
 
 @mock_aws
@@ -3896,12 +3900,12 @@ def test_transact_write_items_conditioncheck_fails():
     dynamodb.create_table(
         TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
     )
-    # Insert an item with email address
+    # Insert an item without email address
     dynamodb.put_item(
         TableName="test-table",
-        Item={"id": {"S": "foo"}, "email_address": {"S": "test@moto.com"}},
+        Item={"id": {"S": "foo"}},
     )
-    # Try to put an email address, but verify whether it exists
+    # Try putting a new item, after verifying the exising item also has email address
     # ConditionCheck should fail
     with pytest.raises(ClientError) as ex:
         dynamodb.transact_write_items(
@@ -3910,15 +3914,15 @@ def test_transact_write_items_conditioncheck_fails():
                     "ConditionCheck": {
                         "Key": {"id": {"S": "foo"}},
                         "TableName": "test-table",
-                        "ConditionExpression": "attribute_not_exists(#e)",
+                        "ConditionExpression": "attribute_exists(#e)",
                         "ExpressionAttributeNames": {"#e": "email_address"},
                     }
                 },
                 {
                     "Put": {
                         "Item": {
-                            "id": {"S": "foo"},
-                            "email_address": {"S": "update@moto.com"},
+                            "id": {"S": "bar"},
+                            "email_address": {"S": "bar@moto.com"},
                         },
                         "TableName": "test-table",
                     }
@@ -3929,10 +3933,10 @@ def test_transact_write_items_conditioncheck_fails():
     assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
     assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
 
-    # Assert the original email address is still present
+    # Assert the original item is still present
     items = dynamodb.scan(TableName="test-table")["Items"]
     assert len(items) == 1
-    assert items[0] == {"email_address": {"S": "test@moto.com"}, "id": {"S": "foo"}}
+    assert items[0] == {"id": {"S": "foo"}}
 
 
 @mock_aws
diff --git a/tests/test_dynamodb/test_dynamodb_condition_expressions.py b/tests/test_dynamodb/test_dynamodb_condition_expressions.py
index 9f74f2d61..d512f1411 100644
--- a/tests/test_dynamodb/test_dynamodb_condition_expressions.py
+++ b/tests/test_dynamodb/test_dynamodb_condition_expressions.py
@@ -3,6 +3,7 @@ from decimal import Decimal
 
 import boto3
 import pytest
+from botocore.exceptions import ClientError
 
 from moto import mock_aws
 
@@ -421,3 +422,153 @@ def test_condition_expression_with_reserved_keyword_as_attr_name():
 
     item = table.get_item(Key={"id": "key-0"})["Item"]
     assert item == {"id": "key-0", "first": {}}
+
+
+@mock_aws
+def test_condition_check_failure_exception_is_raised_when_values_are_returned_for_an_item_with_a_top_level_list():
+    # This explicitly tests for a failure in handling JSONification of DynamoType
+    # when lists are at the top level of an item.
+    # This exception should not be raised:
+    # TypeError: Object of type DynamoType is not JSON serializable
+
+    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb_client.create_table(
+        TableName="example_table",
+        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
+        AttributeDefinitions=[
+            {"AttributeName": "id", "AttributeType": "S"},
+        ],
+        BillingMode="PAY_PER_REQUEST",
+    )
+    record = {
+        "id": {"S": "example_id"},
+        "some_list": {"L": [{"M": {"hello": {"S": "h"}}}]},
+    }
+    dynamodb_client.put_item(
+        TableName="example_table",
+        Item=record,
+    )
+
+    with pytest.raises(ClientError) as error:
+        dynamodb_client.update_item(
+            TableName="example_table",
+            Key={"id": {"S": "example_id"}},
+            UpdateExpression="set some_list=list_append(some_list, :w)",
+            ExpressionAttributeValues={
+                ":w": {"L": [{"M": {"world": {"S": "w"}}}]},
+                ":id": {"S": "incorrect id"},
+            },
+            ConditionExpression="id = :id",
+            ReturnValuesOnConditionCheckFailure="ALL_OLD",
+        )
+
+    assert error.type.__name__ == "ConditionalCheckFailedException"
+    assert error.value.response["Error"] == {
+        "Message": "The conditional request failed",
+        "Code": "ConditionalCheckFailedException",
+    }
+    assert error.value.response["Item"] == {
+        "id": {"S": "example_id"},
+        "some_list": {"L": [{"M": {"hello": {"S": "h"}}}]},
+    }
+
+
+@mock_aws
+def test_condition_check_failure_exception_is_raised_when_values_are_returned_for_an_item_with_a_top_level_string_set():
+    # This explicitly tests for a failure in handling JSONification of DynamoType
+    # when string sets are at the top level of an item.
+    # These exception should not be raised:
+    # TypeError: Object of type DynamoType is not JSON serializable
+    # AttributeError: 'str' object has no attribute 'to_regular_json'
+
+    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb_client.create_table(
+        TableName="example_table",
+        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
+        AttributeDefinitions=[
+            {"AttributeName": "id", "AttributeType": "S"},
+        ],
+        BillingMode="PAY_PER_REQUEST",
+    )
+    record = {
+        "id": {"S": "example_id"},
+        "some_list": {"SS": ["hello"]},
+    }
+    dynamodb_client.put_item(
+        TableName="example_table",
+        Item=record,
+    )
+
+    with pytest.raises(ClientError) as error:
+        dynamodb_client.update_item(
+            TableName="example_table",
+            Key={"id": {"S": "example_id"}},
+            UpdateExpression="set some_list=list_append(some_list, :w)",
+            ExpressionAttributeValues={
+                ":w": {"SS": ["world"]},
+                ":id": {"S": "incorrect id"},
+            },
+            ConditionExpression="id = :id",
+            ReturnValuesOnConditionCheckFailure="ALL_OLD",
+        )
+
+    assert error.type.__name__ == "ConditionalCheckFailedException"
+    assert error.value.response["Error"] == {
+        "Message": "The conditional request failed",
+        "Code": "ConditionalCheckFailedException",
+    }
+    assert error.value.response["Item"] == {
+        "id": {"S": "example_id"},
+        "some_list": {"SS": ["hello"]},
+    }
+
+
+@mock_aws
+def test_condition_check_failure_exception_is_raised_when_values_are_returned_for_an_item_with_a_list_in_a_map():
+    # This explicitly tests for a failure in handling JSONification of DynamoType
+    # when lists are inside a map
+
+    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb_client.create_table(
+        TableName="example_table",
+        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
+        AttributeDefinitions=[
+            {"AttributeName": "id", "AttributeType": "S"},
+        ],
+        BillingMode="PAY_PER_REQUEST",
+    )
+    record = {
+        "id": {"S": "example_id"},
+        "some_list_in_a_map": {
+            "M": {"some_list": {"L": [{"M": {"hello": {"S": "h"}}}]}}
+        },
+    }
+    dynamodb_client.put_item(
+        TableName="example_table",
+        Item=record,
+    )
+
+    with pytest.raises(ClientError) as error:
+        dynamodb_client.update_item(
+            TableName="example_table",
+            Key={"id": {"S": "example_id"}},
+            UpdateExpression="set some_list_in_a_map.some_list=list_append(some_list_in_a_map.some_list, :w)",
+            ExpressionAttributeValues={
+                ":w": {"L": [{"M": {"world": {"S": "w"}}}]},
+                ":id": {"S": "incorrect id"},
+            },
+            ConditionExpression="id = :id",
+            ReturnValuesOnConditionCheckFailure="ALL_OLD",
+        )
+
+    assert error.type.__name__ == "ConditionalCheckFailedException"
+    assert error.value.response["Error"] == {
+        "Message": "The conditional request failed",
+        "Code": "ConditionalCheckFailedException",
+    }
+    assert error.value.response["Item"] == {
+        "id": {"S": "example_id"},
+        "some_list_in_a_map": {
+            "M": {"some_list": {"L": [{"M": {"hello": {"S": "h"}}}]}}
+        },
+    }
diff --git a/tests/test_emr/test_emr_cloudformation.py b/tests/test_emr/test_emr_cloudformation.py
new file mode 100644
index 000000000..3bebfd22a
--- /dev/null
+++ b/tests/test_emr/test_emr_cloudformation.py
@@ -0,0 +1,620 @@
+import json
+
+import boto3
+
+from moto import mock_aws
+from tests import EXAMPLE_AMI_ID
+
+template = """{
+  "AWSTemplateFormatVersion": "2010-09-09",
+  "Description": "The AWS CloudFormation template for this Serverless application",
+  "Resources": {
+    "Cluster1": {
+        "Type" : "AWS::EMR::Cluster",
+        "Properties" : {
+          "Instances" : {
+              "CoreInstanceGroup": {
+                  "InstanceCount": 3,
+                  "InstanceType": "m3g",
+              }
+          },
+          "JobFlowRole" : "EMR_EC2_DefaultRole",
+          "Name" : "my cluster",
+          "ServiceRole" : "EMR_DefaultRole",
+        }
+    },
+  },
+  "Outputs": {
+    "ClusterId": {
+        "Description": "Cluster info",
+        "Value": {
+            "Fn::GetAtt": ["Cluster1", "Id"]
+        },
+    }
+  }
+}"""
+
+
+@mock_aws
+def test_create_simple_cluster__using_cloudformation():
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(StackName="teststack", TemplateBody=template)
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"][0]
+    cluster_id = res["PhysicalResourceId"]
+    assert res["LogicalResourceId"] == "Cluster1"
+    assert res["ResourceType"] == "AWS::EMR::Cluster"
+    assert cluster_id.startswith("j-")
+
+    # Verify outputs
+    stack = cf.describe_stacks(StackName="teststack")["Stacks"][0]
+    assert {"OutputKey": "ClusterId", "OutputValue": cluster_id} in stack["Outputs"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Name"] == "my cluster"
+    assert cl["Ec2InstanceAttributes"]["IamInstanceProfile"] == "EMR_EC2_DefaultRole"
+    assert cl["ServiceRole"] == "EMR_DefaultRole"
+    assert cl["Tags"] == []
+
+
+template_with_tags = {
+    "Resources": {
+        "Cluster1": {
+            "Type": "AWS::EMR::Cluster",
+            "Properties": {
+                "Instances": {
+                    "CoreInstanceGroup": {
+                        "InstanceCount": 3,
+                        "InstanceType": "m3g",
+                    }
+                },
+                "JobFlowRole": "EMR_EC2_DefaultRole",
+                "Name": "my cluster",
+                "ServiceRole": "EMR_DefaultRole",
+                "Tags": [
+                    {"Key": "k1", "Value": "v1"},
+                    {"Key": "k2", "Value": "v2"},
+                ],
+            },
+        },
+    },
+}
+
+
+@mock_aws
+def test_create_simple_cluster_with_tags():
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(StackName="teststack", TemplateBody=json.dumps(template_with_tags))
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"][0]
+    cluster_id = res["PhysicalResourceId"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Tags"] == [{"Key": "k1", "Value": "v1"}, {"Key": "k2", "Value": "v2"}]
+
+
+template_with_custom_ami = {
+    "AWSTemplateFormatVersion": "2010-09-09",
+    "Parameters": {
+        "SubnetId": {"Type": "String"},
+        "InstanceType": {"Type": "String"},
+        "TerminationProtected": {"Type": "String", "Default": "false"},
+        "ElasticMapReducePrincipal": {"Type": "String"},
+        "Ec2Principal": {"Type": "String"},
+    },
+    "Resources": {
+        "cluster": {
+            "Type": "AWS::EMR::Cluster",
+            "Properties": {
+                "CustomAmiId": EXAMPLE_AMI_ID,
+                "Instances": {
+                    "MasterInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnMaster",
+                    },
+                    "CoreInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnCore",
+                    },
+                    "TaskInstanceGroups": [
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-1",
+                        },
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-2",
+                        },
+                    ],
+                    "TerminationProtected": {"Ref": "TerminationProtected"},
+                    "Ec2SubnetId": {"Ref": "SubnetId"},
+                },
+                "Name": "CFNtest",
+                "JobFlowRole": {"Ref": "emrEc2InstanceProfile"},
+                "ServiceRole": {"Ref": "emrRole"},
+                "ReleaseLabel": "release_2024",
+                "VisibleToAllUsers": True,
+                "Tags": [{"Key": "key1", "Value": "value1"}],
+            },
+        },
+        "emrRole": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {
+                                "Service": {"Ref": "ElasticMapReducePrincipal"}
+                            },
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
+                ],
+            },
+        },
+        "emrEc2Role": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": {"Ref": "Ec2Principal"}},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role"
+                ],
+            },
+        },
+        "emrEc2InstanceProfile": {
+            "Type": "AWS::IAM::InstanceProfile",
+            "Properties": {"Path": "/", "Roles": [{"Ref": "emrEc2Role"}]},
+        },
+    },
+}
+
+
+@mock_aws
+def test_create_cluster_with_custom_ami():
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
+    subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
+
+    params = [
+        {"ParameterKey": "InstanceType", "ParameterValue": "i4s"},
+        {"ParameterKey": "SubnetId", "ParameterValue": subnet.id},
+        {
+            "ParameterKey": "ElasticMapReducePrincipal",
+            "ParameterValue": "emr_principal",
+        },
+        {"ParameterKey": "Ec2Principal", "ParameterValue": "ec2_principal"},
+    ]
+
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(
+        StackName="teststack",
+        TemplateBody=json.dumps(template_with_custom_ami),
+        Parameters=params,
+    )
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"]
+    cluster = [r for r in res if r["ResourceType"] == "AWS::EMR::Cluster"][0]
+    cluster_id = cluster["PhysicalResourceId"]
+
+    instance_profile = [
+        r for r in res if r["ResourceType"] == "AWS::IAM::InstanceProfile"
+    ][0]
+    instance_profile_id = instance_profile["PhysicalResourceId"]
+
+    role = [r for r in res if r["ResourceType"] == "AWS::IAM::Role"][0]
+    role_id = role["PhysicalResourceId"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Name"] == "CFNtest"
+    assert cl["Ec2InstanceAttributes"]["Ec2SubnetId"] == subnet.id
+    assert cl["Ec2InstanceAttributes"]["IamInstanceProfile"] == instance_profile_id
+    assert cl["ServiceRole"] == role_id
+    assert cl["ReleaseLabel"] == "release_2024"
+    assert cl["CustomAmiId"] == EXAMPLE_AMI_ID
+
+
+template_with_root_volume = {
+    "AWSTemplateFormatVersion": "2010-09-09",
+    "Parameters": {
+        "InstanceType": {"Type": "String"},
+        "ReleaseLabel": {"Type": "String"},
+        "SubnetId": {"Type": "String"},
+        "TerminationProtected": {"Type": "String", "Default": "false"},
+        "EbsRootVolumeSize": {"Type": "String"},
+    },
+    "Resources": {
+        "cluster": {
+            "Type": "AWS::EMR::Cluster",
+            "Properties": {
+                "EbsRootVolumeSize": {"Ref": "EbsRootVolumeSize"},
+                "Instances": {
+                    "MasterInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnMaster",
+                    },
+                    "CoreInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnCore",
+                    },
+                    "TaskInstanceGroups": [
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-1",
+                        },
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-2",
+                        },
+                    ],
+                    "TerminationProtected": {"Ref": "TerminationProtected"},
+                    "Ec2SubnetId": {"Ref": "SubnetId"},
+                },
+                "Name": "CFNtest",
+                "JobFlowRole": {"Ref": "emrEc2InstanceProfile"},
+                "ServiceRole": {"Ref": "emrRole"},
+                "ReleaseLabel": {"Ref": "ReleaseLabel"},
+                "VisibleToAllUsers": True,
+                "Tags": [{"Key": "key1", "Value": "value1"}],
+            },
+        },
+        "emrRole": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": "elasticmapreduce.amazonaws.com"},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
+                ],
+            },
+        },
+        "emrEc2Role": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": "ec2.amazonaws.com"},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role"
+                ],
+            },
+        },
+        "emrEc2InstanceProfile": {
+            "Type": "AWS::IAM::InstanceProfile",
+            "Properties": {"Path": "/", "Roles": [{"Ref": "emrEc2Role"}]},
+        },
+    },
+}
+
+
+@mock_aws
+def test_create_cluster_with_root_volume():
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
+    subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
+
+    params = [
+        {"ParameterKey": "InstanceType", "ParameterValue": "i4s"},
+        {"ParameterKey": "ReleaseLabel", "ParameterValue": "latest"},
+        {"ParameterKey": "EbsRootVolumeSize", "ParameterValue": "15"},
+        {"ParameterKey": "SubnetId", "ParameterValue": subnet.id},
+        {
+            "ParameterKey": "ElasticMapReducePrincipal",
+            "ParameterValue": "emr_principal",
+        },
+        {"ParameterKey": "Ec2Principal", "ParameterValue": "ec2_principal"},
+    ]
+
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(
+        StackName="teststack",
+        TemplateBody=json.dumps(template_with_root_volume),
+        Parameters=params,
+    )
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"]
+    cluster = [r for r in res if r["ResourceType"] == "AWS::EMR::Cluster"][0]
+    cluster_id = cluster["PhysicalResourceId"]
+
+    instance_profile = [
+        r for r in res if r["ResourceType"] == "AWS::IAM::InstanceProfile"
+    ][0]
+    instance_profile_id = instance_profile["PhysicalResourceId"]
+
+    role = [r for r in res if r["ResourceType"] == "AWS::IAM::Role"][0]
+    role_id = role["PhysicalResourceId"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Name"] == "CFNtest"
+    assert cl["Ec2InstanceAttributes"]["Ec2SubnetId"] == subnet.id
+    assert cl["Ec2InstanceAttributes"]["IamInstanceProfile"] == instance_profile_id
+    assert cl["ServiceRole"] == role_id
+
+
+template_with_kerberos_attrs = {
+    "AWSTemplateFormatVersion": "2010-09-09",
+    "Parameters": {
+        "CrossRealmTrustPrincipalPassword": {"Type": "String"},
+        "KdcAdminPassword": {"Type": "String"},
+        "InstanceType": {"Type": "String"},
+        "ReleaseLabel": {"Type": "String"},
+        "SubnetId": {"Type": "String"},
+    },
+    "Resources": {
+        "cluster": {
+            "Type": "AWS::EMR::Cluster",
+            "Properties": {
+                "Instances": {
+                    "MasterInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnMaster",
+                    },
+                    "CoreInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnCore",
+                    },
+                    "TaskInstanceGroups": [
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-1",
+                        },
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-2",
+                        },
+                    ],
+                    "Ec2SubnetId": {"Ref": "SubnetId"},
+                },
+                "Name": "CFNtest",
+                "JobFlowRole": {"Ref": "emrEc2InstanceProfile"},
+                "KerberosAttributes": {
+                    "CrossRealmTrustPrincipalPassword": {
+                        "Ref": "CrossRealmTrustPrincipalPassword"
+                    },
+                    "KdcAdminPassword": {"Ref": "KdcAdminPassword"},
+                    "Realm": "EC2.INTERNAL",
+                },
+                "ServiceRole": {"Ref": "emrRole"},
+                "ReleaseLabel": {"Ref": "ReleaseLabel"},
+                "SecurityConfiguration": {"Ref": "securityConfiguration"},
+                "VisibleToAllUsers": True,
+                "Tags": [{"Key": "key1", "Value": "value1"}],
+            },
+        },
+        "key": {
+            "Type": "AWS::KMS::Key",
+            "Properties": {
+                "KeyPolicy": {
+                    "Version": "2012-10-17",
+                    "Id": "key-default-1",
+                    "Statement": [
+                        {
+                            "Sid": "Enable IAM User Permissions",
+                            "Effect": "Allow",
+                            "Principal": {"AWS": {"Fn::GetAtt": ["emrEc2Role", "Arn"]}},
+                            "Action": "kms:*",
+                            "Resource": "*",
+                        },
+                        {
+                            "Sid": "Enable IAM User Permissions",
+                            "Effect": "Allow",
+                            "Principal": {
+                                "AWS": {
+                                    "Fn::Join": [
+                                        "",
+                                        [
+                                            "arn:aws:iam::",
+                                            {"Ref": "AWS::AccountId"},
+                                            ":root",
+                                        ],
+                                    ]
+                                }
+                            },
+                            "Action": "kms:*",
+                            "Resource": "*",
+                        },
+                    ],
+                }
+            },
+        },
+        "securityConfiguration": {
+            "Type": "AWS::EMR::SecurityConfiguration",
+            "Properties": {
+                "SecurityConfiguration": {
+                    "AuthenticationConfiguration": {
+                        "KerberosConfiguration": {
+                            "Provider": "ClusterDedicatedKdc",
+                            "ClusterDedicatedKdcConfiguration": {
+                                "TicketLifetimeInHours": 24,
+                                "CrossRealmTrustConfiguration": {
+                                    "Realm": "AD.DOMAIN.COM",
+                                    "Domain": "ad.domain.com",
+                                    "AdminServer": "ad.domain.com",
+                                    "KdcServer": "ad.domain.com",
+                                },
+                            },
+                        }
+                    }
+                }
+            },
+        },
+        "emrRole": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": "elasticmapreduce.amazonaws.com"},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
+                ],
+            },
+        },
+        "emrEc2Role": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": "ec2.amazonaws.com"},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role"
+                ],
+            },
+        },
+        "emrEc2InstanceProfile": {
+            "Type": "AWS::IAM::InstanceProfile",
+            "Properties": {"Path": "/", "Roles": [{"Ref": "emrEc2Role"}]},
+        },
+    },
+    "Outputs": {"keyArn": {"Value": {"Fn::GetAtt": ["key", "Arn"]}}},
+}
+
+
+@mock_aws
+def test_create_cluster_with_kerberos_attrs():
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
+    subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
+
+    params = [
+        {"ParameterKey": "CrossRealmTrustPrincipalPassword", "ParameterValue": "p2ss"},
+        {"ParameterKey": "KdcAdminPassword", "ParameterValue": "adminp2ss"},
+        {"ParameterKey": "InstanceType", "ParameterValue": "i4s"},
+        {"ParameterKey": "ReleaseLabel", "ParameterValue": "latest"},
+        {"ParameterKey": "EbsRootVolumeSize", "ParameterValue": "15"},
+        {"ParameterKey": "SubnetId", "ParameterValue": subnet.id},
+        {
+            "ParameterKey": "ElasticMapReducePrincipal",
+            "ParameterValue": "emr_principal",
+        },
+        {"ParameterKey": "Ec2Principal", "ParameterValue": "ec2_principal"},
+    ]
+
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(
+        StackName="teststack",
+        TemplateBody=json.dumps(template_with_kerberos_attrs),
+        Parameters=params,
+    )
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"]
+    cluster = [r for r in res if r["ResourceType"] == "AWS::EMR::Cluster"][0]
+    cluster_id = cluster["PhysicalResourceId"]
+
+    instance_profile = [
+        r for r in res if r["ResourceType"] == "AWS::IAM::InstanceProfile"
+    ][0]
+    instance_profile_id = instance_profile["PhysicalResourceId"]
+
+    role = [r for r in res if r["ResourceType"] == "AWS::IAM::Role"][0]
+    role_id = role["PhysicalResourceId"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Name"] == "CFNtest"
+    assert cl["Ec2InstanceAttributes"]["Ec2SubnetId"] == subnet.id
+    assert cl["Ec2InstanceAttributes"]["IamInstanceProfile"] == instance_profile_id
+    assert cl["ServiceRole"] == role_id
+
+    kerberos = cl["KerberosAttributes"]
+    assert kerberos == {
+        "Realm": "EC2.INTERNAL",
+        "KdcAdminPassword": "adminp2ss",
+        "CrossRealmTrustPrincipalPassword": "p2ss",
+    }