Repository: getmoto/moto
Pull request #7944 (2024-08-07 14:21:16)

[EMR] Put and Get BlockPublicAccessConfiguration


Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index bac09b6de..a621e2919 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -434,7 +434,7 @@
 
 ## athena
 <details>
-<summary>23% implemented</summary>
+<summary>25% implemented</summary>
 
 - [ ] batch_get_named_query
 - [ ] batch_get_prepared_statement
@@ -452,7 +452,7 @@
 - [ ] delete_named_query
 - [ ] delete_notebook
 - [ ] delete_prepared_statement
-- [ ] delete_work_group
+- [X] delete_work_group
 - [ ] export_notebook
 - [ ] get_calculation_execution
 - [ ] get_calculation_execution_code
diff --git a/docs/docs/services/athena.rst b/docs/docs/services/athena.rst
index c90ccaf82..f016139ab 100644
--- a/docs/docs/services/athena.rst
+++ b/docs/docs/services/athena.rst
@@ -30,7 +30,7 @@ athena
 - [ ] delete_named_query
 - [ ] delete_notebook
 - [ ] delete_prepared_statement
-- [ ] delete_work_group
+- [X] delete_work_group
 - [ ] export_notebook
 - [ ] get_calculation_execution
 - [ ] get_calculation_execution_code
diff --git a/moto/acmpca/exceptions.py b/moto/acmpca/exceptions.py
index 9a584d67c..b17c9b245 100644
--- a/moto/acmpca/exceptions.py
+++ b/moto/acmpca/exceptions.py
@@ -16,3 +16,13 @@ class InvalidS3ObjectAclInCrlConfiguration(JsonRESTError):
             "InvalidS3ObjectAclInCrlConfiguration",
             f"Invalid value for parameter RevocationConfiguration.CrlConfiguration.S3ObjectAcl, value: {value}, valid values: ['PUBLIC_READ', 'BUCKET_OWNER_FULL_CONTROL']",
         )
+
+
+class InvalidStateException(JsonRESTError):
+    code = 400
+
+    def __init__(self, arn: str):
+        super().__init__(
+            "InvalidStateException",
+            f"The certificate authority {arn} is not in the correct state to have a certificate signing request.",
+        )
diff --git a/moto/acmpca/models.py b/moto/acmpca/models.py
index 950c73adb..02e0413f8 100644
--- a/moto/acmpca/models.py
+++ b/moto/acmpca/models.py
@@ -4,11 +4,10 @@ import base64
 import datetime
 from typing import Any, Dict, List, Optional, Tuple
 
-import cryptography.hazmat.primitives.asymmetric.rsa
-import cryptography.x509
+from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.x509 import Certificate, NameOID, load_pem_x509_certificate
+from cryptography.hazmat.primitives.asymmetric import rsa
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
@@ -17,7 +16,11 @@ from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
 from moto.utilities.utils import get_partition
 
-from .exceptions import InvalidS3ObjectAclInCrlConfiguration, ResourceNotFoundException
+from .exceptions import (
+    InvalidS3ObjectAclInCrlConfiguration,
+    InvalidStateException,
+    ResourceNotFoundException,
+)
 
 
 class CertificateAuthority(BaseModel):
@@ -46,84 +49,193 @@ class CertificateAuthority(BaseModel):
         self.usage_mode = "SHORT_LIVED_CERTIFICATE"
         self.security_standard = security_standard or "FIPS_140_2_LEVEL_3_OR_HIGHER"
 
-        self.common_name = self.certificate_authority_configuration.get(
-            "Subject", {}
-        ).get("CommonName", "Moto.org")
-        self.key = cryptography.hazmat.primitives.asymmetric.rsa.generate_private_key(
-            public_exponent=65537, key_size=2048
-        )
+        self.key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
         self.password = str(mock_random.uuid4()).encode("utf-8")
         self.private_bytes = self.key.private_bytes(
             encoding=serialization.Encoding.PEM,
             format=serialization.PrivateFormat.TraditionalOpenSSL,
             encryption_algorithm=serialization.BestAvailableEncryption(self.password),
         )
-        self.certificate: Optional[Certificate] = None
+        self.certificate: Optional[x509.Certificate] = None
         self.certificate_chain: Optional[bytes] = None
-        self.csr = self.generate_csr(self.common_name)
-
         self.issued_certificates: Dict[str, bytes] = dict()
 
-    def generate_cert(self, common_name: str, subject: cryptography.x509.Name) -> bytes:
-        issuer = cryptography.x509.Name(
-            [  # C = US, O = Amazon, OU = Server CA 1B, CN = Amazon
-                cryptography.x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
-                cryptography.x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Amazon"),
-                cryptography.x509.NameAttribute(
-                    NameOID.ORGANIZATIONAL_UNIT_NAME, "Server CA 1B"
-                ),
-                cryptography.x509.NameAttribute(NameOID.COMMON_NAME, common_name),
-            ]
-        )
-        cert = (
-            cryptography.x509.CertificateBuilder()
+        subject = self.certificate_authority_configuration.get("Subject", {})
+        name_attributes = []
+        if "Country" in subject:
+            name_attributes.append(
+                x509.NameAttribute(x509.NameOID.COUNTRY_NAME, subject["Country"])
+            )
+        if "State" in subject:
+            name_attributes.append(
+                x509.NameAttribute(
+                    x509.NameOID.STATE_OR_PROVINCE_NAME, subject["State"]
+                )
+            )
+        if "Organization" in subject:
+            name_attributes.append(
+                x509.NameAttribute(
+                    x509.NameOID.ORGANIZATION_NAME, subject["Organization"]
+                )
+            )
+        if "OrganizationalUnit" in subject:
+            name_attributes.append(
+                x509.NameAttribute(
+                    x509.NameOID.ORGANIZATIONAL_UNIT_NAME, subject["OrganizationalUnit"]
+                )
+            )
+        if "CommonName" in subject:
+            name_attributes.append(
+                x509.NameAttribute(x509.NameOID.COMMON_NAME, subject["CommonName"])
+            )
+        self.issuer = x509.Name(name_attributes)
+        self.csr = self._ca_csr(self.issuer)
+
+    def generate_cert(
+        self,
+        subject: x509.Name,
+        public_key: rsa.RSAPublicKey,
+        extensions: List[Tuple[x509.ExtensionType, bool]],
+    ) -> bytes:
+        builder = (
+            x509.CertificateBuilder()
             .subject_name(subject)
-            .issuer_name(issuer)
-            .public_key(self.key.public_key())
-            .serial_number(cryptography.x509.random_serial_number())
+            .issuer_name(self.issuer)
+            .public_key(public_key)
+            .serial_number(x509.random_serial_number())
             .not_valid_before(utcnow())
             .not_valid_after(utcnow() + datetime.timedelta(days=365))
-            .sign(self.key, hashes.SHA512(), default_backend())
         )
 
+        for extension, critical in extensions:
+            builder = builder.add_extension(extension, critical)
+
+        cert = builder.sign(self.key, hashes.SHA512(), default_backend())
+
         return cert.public_bytes(serialization.Encoding.PEM)
 
-    def generate_csr(self, common_name: str) -> bytes:
+    def _ca_csr(self, issuer: x509.Name) -> bytes:
         csr = (
-            cryptography.x509.CertificateSigningRequestBuilder()
-            .subject_name(
-                cryptography.x509.Name(
-                    [
-                        cryptography.x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
-                        cryptography.x509.NameAttribute(
-                            NameOID.STATE_OR_PROVINCE_NAME, "California"
-                        ),
-                        cryptography.x509.NameAttribute(
-                            NameOID.LOCALITY_NAME, "San Francisco"
-                        ),
-                        cryptography.x509.NameAttribute(
-                            NameOID.ORGANIZATION_NAME, "My Company"
-                        ),
-                        cryptography.x509.NameAttribute(
-                            NameOID.COMMON_NAME, common_name
-                        ),
-                    ]
-                )
+            x509.CertificateSigningRequestBuilder()
+            .subject_name(issuer)
+            .add_extension(
+                x509.BasicConstraints(ca=True, path_length=None),
+                critical=True,
             )
             .sign(self.key, hashes.SHA256())
         )
         return csr.public_bytes(serialization.Encoding.PEM)
 
-    def issue_certificate(self, csr_bytes: bytes) -> str:
-        cert = cryptography.x509.load_pem_x509_csr(base64.b64decode(csr_bytes))
+    def issue_certificate(self, csr_bytes: bytes, template_arn: Optional[str]) -> str:
+        csr = x509.load_pem_x509_csr(base64.b64decode(csr_bytes))
+        extensions = self._x509_extensions(csr, template_arn)
         new_cert = self.generate_cert(
-            common_name=self.common_name, subject=cert.subject
+            subject=csr.subject,
+            public_key=csr.public_key(),  # type: ignore[arg-type]
+            extensions=extensions,
         )
+
         cert_id = str(mock_random.uuid4()).replace("-", "")
         cert_arn = f"arn:{get_partition(self.region_name)}:acm-pca:{self.region_name}:{self.account_id}:certificate-authority/{self.id}/certificate/{cert_id}"
         self.issued_certificates[cert_arn] = new_cert
         return cert_arn
 
+    def _x509_extensions(
+        self, csr: x509.CertificateSigningRequest, template_arn: Optional[str]
+    ) -> List[Tuple[x509.ExtensionType, bool]]:
+        """
+        Uses a PCA certificate template ARN to return a list of X.509 extensions.
+        These extensions are part of the constructed certificate.
+
+        See https://docs.aws.amazon.com/privateca/latest/userguide/UsingTemplates.html
+        """
+        extensions = []
+
+        if template_arn == "arn:aws:acm-pca:::template/RootCACertificate/V1":
+            extensions.extend(
+                [
+                    (
+                        x509.BasicConstraints(ca=True, path_length=None),
+                        True,
+                    ),
+                    (
+                        x509.KeyUsage(
+                            crl_sign=True,
+                            key_cert_sign=True,
+                            digital_signature=True,
+                            content_commitment=False,
+                            key_encipherment=False,
+                            data_encipherment=False,
+                            key_agreement=False,
+                            encipher_only=False,
+                            decipher_only=False,
+                        ),
+                        True,
+                    ),
+                    (
+                        x509.SubjectKeyIdentifier.from_public_key(csr.public_key()),
+                        False,
+                    ),
+                ]
+            )
+
+        elif template_arn in (
+            "arn:aws:acm-pca:::template/EndEntityCertificate/V1",
+            None,
+        ):
+            extensions.extend(
+                [
+                    (
+                        x509.BasicConstraints(ca=False, path_length=None),
+                        True,
+                    ),
+                    (
+                        x509.AuthorityKeyIdentifier.from_issuer_public_key(
+                            self.key.public_key()
+                        ),
+                        False,
+                    ),
+                    (
+                        x509.SubjectKeyIdentifier.from_public_key(csr.public_key()),
+                        False,
+                    ),
+                    (
+                        x509.KeyUsage(
+                            crl_sign=False,
+                            key_cert_sign=False,
+                            digital_signature=True,
+                            content_commitment=False,
+                            key_encipherment=True,
+                            data_encipherment=False,
+                            key_agreement=False,
+                            encipher_only=False,
+                            decipher_only=False,
+                        ),
+                        True,
+                    ),
+                    (
+                        x509.ExtendedKeyUsage(
+                            [
+                                x509.ExtendedKeyUsageOID.SERVER_AUTH,
+                                x509.ExtendedKeyUsageOID.CLIENT_AUTH,
+                            ]
+                        ),
+                        False,
+                    ),
+                ]
+            )
+
+        cn = csr.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
+        if cn:
+            extensions.append(
+                (
+                    x509.SubjectAlternativeName([x509.DNSName(cn[0].value)]),  # type: ignore[arg-type]
+                    False,
+                ),
+            )
+
+        return extensions
+
     def get_certificate(self, certificate_arn: str) -> bytes:
         return self.issued_certificates[certificate_arn]
 
@@ -171,7 +283,7 @@ class CertificateAuthority(BaseModel):
     def import_certificate_authority_certificate(
         self, certificate: bytes, certificate_chain: Optional[bytes]
     ) -> None:
-        self.certificate = load_pem_x509_certificate(certificate)
+        self.certificate = x509.load_pem_x509_certificate(certificate)
         self.certificate_chain = certificate_chain
         self.status = "ACTIVE"
         self.updated_at = unix_time()
@@ -243,6 +355,8 @@ class ACMPCABackend(BaseBackend):
         self, certificate_authority_arn: str
     ) -> Tuple[bytes, Optional[bytes]]:
         ca = self.describe_certificate_authority(certificate_authority_arn)
+        if ca.status != "ACTIVE":
+            raise InvalidStateException(certificate_authority_arn)
         return ca.certificate_bytes, ca.certificate_chain
 
     def get_certificate_authority_csr(self, certificate_authority_arn: str) -> bytes:
@@ -273,13 +387,15 @@ class ACMPCABackend(BaseBackend):
         ca = self.describe_certificate_authority(certificate_authority_arn)
         ca.status = "DELETED"
 
-    def issue_certificate(self, certificate_authority_arn: str, csr: bytes) -> str:
+    def issue_certificate(
+        self, certificate_authority_arn: str, csr: bytes, template_arn: Optional[str]
+    ) -> str:
         """
-        The following parameters are not yet implemented: ApiPassthrough, SigningAlgorithm, TemplateArn, Validity, ValidityNotBefore, IdempotencyToken
+        The following parameters are not yet implemented: ApiPassthrough, SigningAlgorithm, Validity, ValidityNotBefore, IdempotencyToken
         Some fields of the resulting certificate will have default values, instead of using the CSR
         """
         ca = self.describe_certificate_authority(certificate_authority_arn)
-        certificate_arn = ca.issue_certificate(csr)
+        certificate_arn = ca.issue_certificate(csr, template_arn)
         return certificate_arn
 
     def get_certificate(
diff --git a/moto/acmpca/responses.py b/moto/acmpca/responses.py
index d7f3c0f5b..899f23127 100644
--- a/moto/acmpca/responses.py
+++ b/moto/acmpca/responses.py
@@ -100,10 +100,12 @@ class ACMPCAResponse(BaseResponse):
     def issue_certificate(self) -> str:
         params = json.loads(self.body)
         certificate_authority_arn = params.get("CertificateAuthorityArn")
+        template_arn = params.get("TemplateArn")
         csr = params.get("Csr").encode("utf-8")
         certificate_arn = self.acmpca_backend.issue_certificate(
             certificate_authority_arn=certificate_authority_arn,
             csr=csr,
+            template_arn=template_arn,
         )
         return json.dumps(dict(CertificateArn=certificate_arn))
 
diff --git a/moto/athena/models.py b/moto/athena/models.py
index 1ca9d35bd..988cd387a 100644
--- a/moto/athena/models.py
+++ b/moto/athena/models.py
@@ -62,6 +62,20 @@ class WorkGroup(TaggableResourceMixin, BaseModel):
         self.description = description
         self.configuration = configuration
 
+        if "EnableMinimumEncryptionConfiguration" not in self.configuration:
+            self.configuration["EnableMinimumEncryptionConfiguration"] = False
+        if "EnforceWorkGroupConfiguration" not in self.configuration:
+            self.configuration["EnforceWorkGroupConfiguration"] = True
+        if "EngineVersion" not in self.configuration:
+            self.configuration["EngineVersion"] = {
+                "EffectiveEngineVersion": "Athena engine " "version 3",
+                "SelectedEngineVersion": "AUTO",
+            }
+        if "PublishCloudWatchMetricsEnabled" not in self.configuration:
+            self.configuration["PublishCloudWatchMetricsEnabled"] = False
+        if "RequesterPaysEnabled" not in self.configuration:
+            self.configuration["RequesterPaysEnabled"] = False
+
 
 class DataCatalog(TaggableResourceMixin, BaseModel):
     def __init__(
@@ -179,7 +193,13 @@ class AthenaBackend(BaseBackend):
 
         # Initialise with the primary workgroup
         self.create_work_group(
-            name="primary", description="", configuration=dict(), tags=[]
+            name="primary",
+            description="",
+            configuration={
+                "ResultConfiguration": {},
+                "EnforceWorkGroupConfiguration": False,
+            },
+            tags=[],
         )
 
     def create_work_group(
@@ -218,6 +238,9 @@ class AthenaBackend(BaseBackend):
             "CreationTime": time.time(),
         }
 
+    def delete_work_group(self, name: str) -> None:
+        self.work_groups.pop(name, None)
+
     def start_query_execution(
         self,
         query: str,
diff --git a/moto/athena/responses.py b/moto/athena/responses.py
index 75712e0f7..938f9a9ea 100644
--- a/moto/athena/responses.py
+++ b/moto/athena/responses.py
@@ -41,6 +41,11 @@ class AthenaResponse(BaseResponse):
         name = self._get_param("WorkGroup")
         return json.dumps({"WorkGroup": self.athena_backend.get_work_group(name)})
 
+    def delete_work_group(self) -> str:
+        name = self._get_param("WorkGroup")
+        self.athena_backend.delete_work_group(name)
+        return "{}"
+
     def start_query_execution(self) -> Union[Tuple[str, Dict[str, int]], str]:
         query = self._get_param("QueryString")
         context = self._get_param("QueryExecutionContext")
diff --git a/moto/cloudformation/responses.py b/moto/cloudformation/responses.py
index c1a81441c..b17187a6e 100644
--- a/moto/cloudformation/responses.py
+++ b/moto/cloudformation/responses.py
@@ -825,6 +825,9 @@ DESCRIBE_STACKS_TEMPLATE = """<DescribeStacksResponse>
           <member>
             <OutputKey>{{ output.key }}</OutputKey>
             <OutputValue>{{ output.value }}</OutputValue>
+            {% for export in stack.exports if export.value == output.value %}
+                <ExportName>{{ export.name }}</ExportName>
+            {% endfor %}
             {% if output.description %}<Description>{{ output.description }}</Description>{% endif %}
           </member>
         {% endfor %}
diff --git a/moto/core/responses.py b/moto/core/responses.py
index 8a171ffd7..4b0ae6689 100644
--- a/moto/core/responses.py
+++ b/moto/core/responses.py
@@ -1139,6 +1139,8 @@ def to_str(value: Any, spec: Dict[str, Any]) -> str:
         return utcfromtimestamp(value).replace(tzinfo=datetime.timezone.utc).isoformat()
     elif vtype == "string":
         return str(value)
+    elif vtype == "structure":
+        return ""
     elif value is None:
         return "null"
     else:
diff --git a/moto/dynamodb/models/__init__.py b/moto/dynamodb/models/__init__.py
index 407cce047..e8ab2ce99 100644
--- a/moto/dynamodb/models/__init__.py
+++ b/moto/dynamodb/models/__init__.py
@@ -107,13 +107,11 @@ class DynamoDBBackend(BaseBackend):
         return table
 
     def delete_table(self, name: str) -> Table:
-        if name not in self.tables:
-            raise ResourceNotFoundException
-        table_for_deletion = self.tables.get(name)
+        table_for_deletion = self.get_table(name)
         if isinstance(table_for_deletion, Table):
             if table_for_deletion.deletion_protection_enabled:
                 raise DeletionProtectedException(name)
-        return self.tables.pop(name)
+        return self.tables.pop(table_for_deletion.name)
 
     def describe_endpoints(self) -> List[Dict[str, Union[int, str]]]:
         return [
@@ -166,10 +164,12 @@ class DynamoDBBackend(BaseBackend):
         return tables, None
 
     def describe_table(self, name: str) -> Dict[str, Any]:
-        # We can't use get_table() here, because the error message is slightly different for this operation
-        if name not in self.tables:
+        # Error message is slightly different for this operation
+        try:
+            table = self.get_table(name)
+        except ResourceNotFoundException:
             raise ResourceNotFoundException(table_name=name)
-        return self.tables[name].describe(base_key="Table")
+        return table.describe(base_key="Table")
 
     def update_table(
         self,
@@ -185,52 +185,30 @@ class DynamoDBBackend(BaseBackend):
         if attr_definitions:
             table.attr = attr_definitions
         if global_index:
-            table = self.update_table_global_indexes(name, global_index)
+            self.update_table_global_indexes(table, global_index)
         if throughput:
-            table = self.update_table_throughput(name, throughput)
+            table.throughput = throughput
         if billing_mode:
-            table = self.update_table_billing_mode(name, billing_mode)
+            table.billing_mode = billing_mode
         if stream_spec:
-            table = self.update_table_streams(name, stream_spec)
+            self.update_table_streams(table, stream_spec)
         if deletion_protection_enabled:
-            table = self.update_table_deletion_protection_enabled(
-                name, deletion_protection_enabled
-            )
-        return table
-
-    def update_table_throughput(self, name: str, throughput: Dict[str, int]) -> Table:
-        table = self.tables[name]
-        table.throughput = throughput
-        return table
-
-    def update_table_billing_mode(self, name: str, billing_mode: str) -> Table:
-        table = self.tables[name]
-        table.billing_mode = billing_mode
-        return table
-
-    def update_table_deletion_protection_enabled(
-        self, name: str, deletion_protection_enabled: bool
-    ) -> Table:
-        table = self.tables[name]
-        table.deletion_protection_enabled = deletion_protection_enabled
+            table.deletion_protection_enabled = deletion_protection_enabled
         return table
 
     def update_table_streams(
-        self, name: str, stream_specification: Dict[str, Any]
-    ) -> Table:
-        table = self.tables[name]
+        self, table: Table, stream_specification: Dict[str, Any]
+    ) -> None:
         if (
             stream_specification.get("StreamEnabled")
             or stream_specification.get("StreamViewType")
         ) and table.latest_stream_label:
             raise StreamAlreadyEnabledException
         table.set_stream_specification(stream_specification)
-        return table
 
     def update_table_global_indexes(
-        self, name: str, global_index_updates: List[Dict[str, Any]]
-    ) -> Table:
-        table = self.tables[name]
+        self, table: Table, global_index_updates: List[Dict[str, Any]]
+    ) -> None:
         gsis_by_name = dict((i.name, i) for i in table.global_indexes)
         for gsi_update in global_index_updates:
             gsi_to_create = gsi_update.get("Create")
@@ -268,7 +246,6 @@ class DynamoDBBackend(BaseBackend):
                 )
 
         table.global_indexes = list(gsis_by_name.values())
-        return table
 
     def put_item(
         self,
@@ -347,9 +324,17 @@ class DynamoDBBackend(BaseBackend):
             return table.schema
 
     def get_table(self, table_name: str) -> Table:
-        if table_name not in self.tables:
+        table = next(
+            (
+                t
+                for t in self.tables.values()
+                if t.name == table_name or t.table_arn == table_name
+            ),
+            None,
+        )
+        if not table:
             raise ResourceNotFoundException()
-        return self.tables[table_name]
+        return table
 
     def get_item(
         self,
@@ -587,8 +572,9 @@ class DynamoDBBackend(BaseBackend):
         return table.delete_item(hash_value, range_value)
 
     def update_time_to_live(self, table_name: str, ttl_spec: Dict[str, Any]) -> None:
-        table = self.tables.get(table_name)
-        if table is None:
+        try:
+            table = self.get_table(table_name)
+        except ResourceNotFoundException:
             raise JsonRESTError("ResourceNotFound", "Table not found")
 
         if "Enabled" not in ttl_spec or "AttributeName" not in ttl_spec:
@@ -604,8 +590,9 @@ class DynamoDBBackend(BaseBackend):
         table.ttl["AttributeName"] = ttl_spec["AttributeName"]
 
     def describe_time_to_live(self, table_name: str) -> Dict[str, Any]:
-        table = self.tables.get(table_name)
-        if table is None:
+        try:
+            table = self.get_table(table_name)
+        except ResourceNotFoundException:
             raise JsonRESTError("ResourceNotFound", "Table not found")
 
         return table.ttl
@@ -786,7 +773,12 @@ class DynamoDBBackend(BaseBackend):
     def list_backups(self, table_name: str) -> List[Backup]:
         backups = list(self.backups.values())
         if table_name is not None:
-            backups = [backup for backup in backups if backup.table.name == table_name]
+            backups = [
+                backup
+                for backup in backups
+                if backup.table.name == table_name
+                or backup.table.table_arn == table_name
+            ]
         return backups
 
     def create_backup(self, table_name: str, backup_name: str) -> Backup:
diff --git a/moto/emr/models.py b/moto/emr/models.py
index 1e1381878..fc14a121a 100644
--- a/moto/emr/models.py
+++ b/moto/emr/models.py
@@ -556,6 +556,7 @@ class ElasticMapReduceBackend(BaseBackend):
         self.clusters: Dict[str, FakeCluster] = {}
         self.instance_groups: Dict[str, FakeInstanceGroup] = {}
         self.security_configurations: Dict[str, FakeSecurityConfiguration] = {}
+        self.block_public_access_configuration: Dict[str, Any] = {}  # type: ignore [misc]
 
     @property
     def ec2_backend(self) -> Any:  # type: ignore[misc]
@@ -885,5 +886,39 @@ class ElasticMapReduceBackend(BaseBackend):
             )
         del self.security_configurations[name]
 
+    def get_block_public_access_configuration(
+        self,
+    ) -> Dict[str, Any]:  # type ignore[misc]
+        return self.block_public_access_configuration
+
+    def put_block_public_access_configuration(
+        self,
+        block_public_security_group_rules: bool,
+        rule_ranges: Optional[List[Dict[str, int]]],
+    ) -> None:
+        from moto.sts import sts_backends
+
+        sts_backend = sts_backends[self.account_id]["global"]
+        _, user_arn, _ = sts_backend.get_caller_identity(
+            self.account_id, region=self.region_name
+        )
+        self.block_public_access_configuration = {
+            "block_public_access_configuration": {
+                "block_public_security_group_rules": block_public_security_group_rules,
+                "permitted_public_security_group_rule_ranges": [
+                    {
+                        "min_range": rule_range.get("MinRange"),
+                        "max_range": rule_range.get("MaxRange"),
+                    }
+                    for rule_range in rule_ranges or []
+                ],
+            },
+            "block_public_access_configuration_metadata": {
+                "creation_date_time": datetime.now(),
+                "created_by_arn": user_arn,
+            },
+        }
+        return
+
 
 emr_backends = BackendDict(ElasticMapReduceBackend, "emr")
diff --git a/moto/emr/responses.py b/moto/emr/responses.py
index 52688e171..c1618455a 100644
--- a/moto/emr/responses.py
+++ b/moto/emr/responses.py
@@ -535,6 +535,45 @@ class ElasticMapReduceResponse(BaseResponse):
         template = self.response_template(REMOVE_AUTO_SCALING_POLICY)
         return template.render()
 
+    @generate_boto3_response("GetBlockPublicAccessConfiguration")
+    def get_block_public_access_configuration(self) -> str:
+        configuration = self.backend.get_block_public_access_configuration()
+        config = configuration.get("block_public_access_configuration") or {}
+        metadata = configuration.get("block_public_access_configuration_metadata") or {}
+        template = self.response_template(
+            GET_BLOCK_PUBLIC_ACCESS_CONFIGURATION_TEMPLATE
+        )
+        return template.render(
+            block_public_security_group_rules=config.get(
+                "block_public_security_group_rules"
+            ),
+            permitted_public_security_group_rule_ranges=config.get(
+                "permitted_public_security_group_rule_ranges"
+            ),
+            creation_date_time=metadata.get("creation_date_time"),
+            created_by_arn=metadata.get("created_by_arn"),
+        )
+
+    @generate_boto3_response("PutBlockPublicAccessConfiguration")
+    def put_block_public_access_configuration(self) -> str:
+        params = self._get_params()
+        block_public_access_configuration = (
+            params.get("BlockPublicAccessConfiguration") or {}
+        )
+        self.backend.put_block_public_access_configuration(
+            block_public_security_group_rules=block_public_access_configuration.get(
+                "BlockPublicSecurityGroupRules"
+            )
+            or True,
+            rule_ranges=block_public_access_configuration.get(
+                "PermittedPublicSecurityGroupRuleRanges"
+            ),
+        )
+        template = self.response_template(
+            PUT_BLOCK_PUBLIC_ACCESS_CONFIGURATION_TEMPLATE
+        )
+        return template.render()
+
 
 ADD_INSTANCE_GROUPS_TEMPLATE = """<AddInstanceGroupsResponse xmlns="http://elasticmapreduce.amazonaws.com/doc/2009-03-31">
   <AddInstanceGroupsResult>
@@ -1418,3 +1457,36 @@ DELETE_SECURITY_CONFIGURATION_TEMPLATE = """<DeleteSecurityConfigurationResponse
     <RequestId>2690d7eb-ed86-11dd-9877-6fad448a8419</RequestId>
   </ResponseMetadata>
 </DeleteSecurityConfigurationResponse>"""
+
+PUT_BLOCK_PUBLIC_ACCESS_CONFIGURATION_TEMPLATE = """<PutBlockPublicAccessConfigurationResponse xmlns="http://elasticmapreduce.amazonaws.com/doc/2009-03-31">
+  <ResponseMetadata>
+    <RequestId>2690d7eb-ed86-11dd-9877-6fad448a8419</RequestId>
+  </ResponseMetadata>
+</PutBlockPublicAccessConfigurationResponse>"""
+
+GET_BLOCK_PUBLIC_ACCESS_CONFIGURATION_TEMPLATE = """
+  <GetBlockPublicAccessConfigurationResponse xmlns="http://elasticmapreduce.amazonaws.com/doc/2009-03-31">
+    <GetBlockPublicAccessConfigurationResult>
+      <BlockPublicAccessConfiguration>
+        <BlockPublicSecurityGroupRules>
+          {{block_public_security_group_rules}}
+        </BlockPublicSecurityGroupRules>
+        <PermittedPublicSecurityGroupRuleRanges>
+          {% for rule_range in permitted_public_security_group_rule_ranges %}
+            <member>
+              <MinRange>{{rule_range['min_range']}}</MinRange>
+              <MaxRange>{{rule_range['max_range']}}</MaxRange>      
+            </member>
+          {% endfor %}
+        </PermittedPublicSecurityGroupRuleRanges>
+      </BlockPublicAccessConfiguration>
+      <BlockPublicAccessConfigurationMetadata>
+        <CreationDateTime>{{creation_date_time}}</CreationDateTime>
+        <CreatedByArn>{{created_by_arn}}</CreatedByArn>
+      </BlockPublicAccessConfigurationMetadata>
+    </GetBlockPublicAccessConfigurationResult>
+    <ResponseMetadata>
+      <RequestId>2690d7eb-ed86-11dd-9877-6fad448a8419</RequestId>
+    </ResponseMetadata>
+  </GetBlockPublicAccessConfigurationResponse>
+"""
diff --git a/moto/kms/models.py b/moto/kms/models.py
index 43d78940f..aed7cec72 100644
--- a/moto/kms/models.py
+++ b/moto/kms/models.py
@@ -1,5 +1,6 @@
 import json
 import os
+import typing
 from collections import defaultdict
 from copy import copy
 from datetime import datetime, timedelta
@@ -78,6 +79,15 @@ class Key(CloudFormationModel):
         self.description = description or ""
         self.enabled = True
         self.multi_region = multi_region
+        if self.multi_region:
+            self.multi_region_configuration: typing.Dict[str, Any] = {
+                "MultiRegionKeyType": "PRIMARY",
+                "PrimaryKey": {
+                    "Arn": f"arn:{get_partition(region)}:kms:{region}:{account_id}:key/{self.id}",
+                    "Region": self.region,
+                },
+                "ReplicaKeys": [],
+            }
         self.key_rotation_status = False
         self.deletion_date: Optional[datetime] = None
         self.key_material = generate_master_key()
@@ -88,7 +98,6 @@ class Key(CloudFormationModel):
         self.arn = (
             f"arn:{get_partition(region)}:kms:{region}:{account_id}:key/{self.id}"
         )
-
         self.grants: Dict[str, Grant] = dict()
 
     def add_grant(
@@ -206,6 +215,10 @@ class Key(CloudFormationModel):
                 "SigningAlgorithms": self.signing_algorithms,
             }
         }
+        if key_dict["KeyMetadata"]["MultiRegion"]:
+            key_dict["KeyMetadata"]["MultiRegionConfiguration"] = (
+                self.multi_region_configuration
+            )
         if self.key_state == "PendingDeletion":
             key_dict["KeyMetadata"]["DeletionDate"] = unix_time(self.deletion_date)
         return key_dict
@@ -326,8 +339,24 @@ class KmsBackend(BaseBackend):
         replica_key = copy(self.keys[key_id])
         replica_key.region = replica_region
         replica_key.arn = replica_key.arn.replace(self.region_name, replica_region)
+
+        if replica_key.multi_region:
+            existing_replica = any(
+                replica["Region"] == replica_region
+                for replica in replica_key.multi_region_configuration["ReplicaKeys"]
+            )
+
+            if not existing_replica:
+                replica_payload = {"Arn": replica_key.arn, "Region": replica_region}
+                replica_key.multi_region_configuration["ReplicaKeys"].append(
+                    replica_payload
+                )
+
         to_region_backend = kms_backends[self.account_id][replica_region]
         to_region_backend.keys[replica_key.id] = replica_key
+
+        self.multi_region_configuration = copy(replica_key.multi_region_configuration)
+
         return replica_key
 
     def update_key_description(self, key_id: str, description: str) -> None:
@@ -348,6 +377,16 @@ class KmsBackend(BaseBackend):
         key_id = self.get_key_id(key_id)
         if r"alias/" in str(key_id).lower():
             key_id = self.get_key_id_from_alias(key_id)  # type: ignore[assignment]
+
+        key = self.keys[self.get_key_id(key_id)]
+
+        if key.multi_region:
+            if key.arn == key.multi_region_configuration["PrimaryKey"]["Arn"]:
+                return self.keys[self.get_key_id(key_id)]
+            else:
+                key.multi_region_configuration["MultiRegionKeyType"] = "REPLICA"
+                return key
+
         return self.keys[self.get_key_id(key_id)]
 
     def list_keys(self) -> Iterable[Key]:


Test patch:
diff --git a/tests/test_acmpca/test_acmpca.py b/tests/test_acmpca/test_acmpca.py
index bbcc4c071..a9c592cd5 100644
--- a/tests/test_acmpca/test_acmpca.py
+++ b/tests/test_acmpca/test_acmpca.py
@@ -7,7 +7,7 @@ import pytest
 from botocore.exceptions import ClientError
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.x509 import NameOID
+from cryptography.x509 import DNSName, NameOID
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID
@@ -121,10 +121,15 @@ def test_get_certificate_authority_certificate():
         IdempotencyToken="terraform-20221125230308947400000001",
     )["CertificateAuthorityArn"]
 
-    resp = client.get_certificate_authority_certificate(CertificateAuthorityArn=ca_arn)
-
-    # Certificate is empty for now,  until we call import_certificate_authority_certificate
-    assert resp["Certificate"] == ""
+    # CA is not usable yet
+    with pytest.raises(ClientError) as exc:
+        client.get_certificate_authority_certificate(CertificateAuthorityArn=ca_arn)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidStateException"
+    assert (
+        err["Message"]
+        == f"The certificate authority {ca_arn} is not in the correct state to have a certificate signing request."
+    )
 
 
 @mock_aws
@@ -134,15 +139,52 @@ def test_get_certificate_authority_csr():
         CertificateAuthorityConfiguration={
             "KeyAlgorithm": "RSA_4096",
             "SigningAlgorithm": "SHA512WITHRSA",
-            "Subject": {"CommonName": "yscb41lw.test"},
+            "Subject": {
+                "CommonName": "evilcorp.com",
+                "Country": "US",
+                "State": "MA",
+                "Organization": "Evil Corp",
+                "OrganizationalUnit": "Finance",
+            },
         },
-        CertificateAuthorityType="SUBORDINATE",
+        CertificateAuthorityType="ROOT",
         IdempotencyToken="terraform-20221125230308947400000001",
     )["CertificateAuthorityArn"]
 
     resp = client.get_certificate_authority_csr(CertificateAuthorityArn=ca_arn)
 
     assert "Csr" in resp
+    csr = cryptography.x509.load_pem_x509_csr(resp["Csr"].encode("utf-8"))
+    assert (
+        csr.subject.get_attributes_for_oid(cryptography.x509.NameOID.COMMON_NAME)[
+            0
+        ].value
+        == "evilcorp.com"
+    )
+    assert (
+        csr.subject.get_attributes_for_oid(cryptography.x509.NameOID.COUNTRY_NAME)[
+            0
+        ].value
+        == "US"
+    )
+    assert (
+        csr.subject.get_attributes_for_oid(
+            cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME
+        )[0].value
+        == "MA"
+    )
+    assert (
+        csr.subject.get_attributes_for_oid(cryptography.x509.NameOID.ORGANIZATION_NAME)[
+            0
+        ].value
+        == "Evil Corp"
+    )
+    assert (
+        csr.subject.get_attributes_for_oid(
+            cryptography.x509.NameOID.ORGANIZATIONAL_UNIT_NAME
+        )[0].value
+        == "Finance"
+    )
 
 
 @mock_aws
@@ -364,6 +406,67 @@ def test_import_certificate_authority_certificate():
     assert "-----BEGIN CERTIFICATE-----" in resp["Certificate"]
 
 
+@mock_aws
+def test_end_entity_certificate_issuance():
+    client = boto3.client("acm-pca", region_name="us-east-2")
+
+    # Create a CA
+    ca_arn = client.create_certificate_authority(
+        CertificateAuthorityConfiguration={
+            "KeyAlgorithm": "RSA_4096",
+            "SigningAlgorithm": "SHA512WITHRSA",
+            "Subject": {"CommonName": "evilcorp.com"},
+        },
+        CertificateAuthorityType="ROOT",
+    )["CertificateAuthorityArn"]
+
+    # Create CA certificate
+    csr = client.get_certificate_authority_csr(CertificateAuthorityArn=ca_arn)["Csr"]
+    certificate_arn = client.issue_certificate(
+        CertificateAuthorityArn=ca_arn,
+        Csr=csr,
+        SigningAlgorithm="SHA256WITHRSA",
+        TemplateArn="arn:aws:acm-pca:::template/RootCACertificate/V1",
+        Validity={"Type": "YEARS", "Value": 10},
+    )["CertificateArn"]
+
+    resp = client.get_certificate(
+        CertificateAuthorityArn=ca_arn, CertificateArn=certificate_arn
+    )
+    ca_cert = resp["Certificate"]
+
+    client.import_certificate_authority_certificate(
+        CertificateAuthorityArn=ca_arn,
+        Certificate=ca_cert,
+    )
+
+    # Create a end-entity certificate and sign it by the CA
+    private_key = cryptography.hazmat.primitives.asymmetric.rsa.generate_private_key(
+        public_exponent=65537, key_size=2048
+    )
+    ee_csr = create_csr(private_key, "US", "WA", "BezosCorp", "bezoscorp.com")
+    ee_cert_arn = client.issue_certificate(
+        CertificateAuthorityArn=ca_arn,
+        Csr=ee_csr,
+        SigningAlgorithm="SHA256WITHRSA",
+        Validity={"Type": "YEARS", "Value": 1},
+    )["CertificateArn"]
+    ee_cert = client.get_certificate(
+        CertificateAuthorityArn=ca_arn, CertificateArn=ee_cert_arn
+    )["Certificate"]
+
+    # Ensure that this is a valid single-hierarchy CA and all certificates are valid
+    store = cryptography.x509.verification.Store(
+        [cryptography.x509.load_pem_x509_certificate(ca_cert.encode("utf-8"))]
+    )
+    builder = cryptography.x509.verification.PolicyBuilder().store(store)
+    verifier = builder.build_server_verifier(DNSName("bezoscorp.com"))
+    chain = verifier.verify(
+        cryptography.x509.load_pem_x509_certificate(ee_cert.encode("utf-8")), []
+    )
+    assert len(chain) == 2
+
+
 @mock_aws
 def test_tag_certificate_authority():
     client = boto3.client("acm-pca", region_name="eu-west-1")
@@ -410,6 +513,36 @@ def test_untag_certificate_authority():
     assert resp["Tags"] == [{"Key": "t2", "Value": "v2"}]
 
 
+def create_csr(private_key, country, state, org, cn):
+    csr = (
+        cryptography.x509.CertificateSigningRequestBuilder()
+        .subject_name(
+            cryptography.x509.Name(
+                [
+                    cryptography.x509.NameAttribute(
+                        cryptography.x509.NameOID.COUNTRY_NAME, country
+                    ),
+                    cryptography.x509.NameAttribute(
+                        cryptography.x509.NameOID.STATE_OR_PROVINCE_NAME, state
+                    ),
+                    cryptography.x509.NameAttribute(
+                        cryptography.x509.NameOID.ORGANIZATION_NAME, org
+                    ),
+                    cryptography.x509.NameAttribute(
+                        cryptography.x509.NameOID.COMMON_NAME, cn
+                    ),
+                ]
+            )
+        )
+        .add_extension(
+            cryptography.x509.BasicConstraints(ca=False, path_length=None),
+            critical=True,
+        )
+        .sign(private_key, hashes.SHA256())
+    )
+    return csr.public_bytes(serialization.Encoding.PEM)
+
+
 def create_cert():
     serial_number = cryptography.x509.random_serial_number()
     subject = cryptography.x509.Name(
diff --git a/tests/test_athena/test_athena.py b/tests/test_athena/test_athena.py
index dafa7de42..c48c08f6a 100644
--- a/tests/test_athena/test_athena.py
+++ b/tests/test_athena/test_athena.py
@@ -5,6 +5,7 @@ from botocore.exceptions import ClientError
 from moto import mock_aws, settings
 from moto.athena.models import QueryResults, athena_backends
 from moto.core import DEFAULT_ACCOUNT_ID
+from tests import aws_verified
 
 
 @mock_aws
@@ -57,32 +58,56 @@ def test_create_work_group():
     assert work_group["State"] == "ENABLED"
 
 
-@mock_aws
+@aws_verified
+@pytest.mark.aws_verified
 def test_get_primary_workgroup():
     client = boto3.client("athena", region_name="us-east-1")
-    assert len(client.list_work_groups()["WorkGroups"]) == 1
+    assert len(client.list_work_groups()["WorkGroups"]) >= 1
 
     primary = client.get_work_group(WorkGroup="primary")["WorkGroup"]
     assert primary["Name"] == "primary"
-    assert primary["Configuration"] == {}
+    assert primary["Configuration"] == {
+        "ResultConfiguration": {},
+        "EnforceWorkGroupConfiguration": False,
+        "PublishCloudWatchMetricsEnabled": False,
+        "RequesterPaysEnabled": False,
+        "EngineVersion": {
+            "SelectedEngineVersion": "AUTO",
+            "EffectiveEngineVersion": "Athena engine version 3",
+        },
+        "EnableMinimumEncryptionConfiguration": False,
+    }
 
 
-@mock_aws
+@aws_verified
+@pytest.mark.aws_verified
 def test_create_and_get_workgroup():
     client = boto3.client("athena", region_name="us-east-1")
 
-    create_basic_workgroup(client=client, name="athena_workgroup")
-
-    work_group = client.get_work_group(WorkGroup="athena_workgroup")["WorkGroup"]
-    del work_group["CreationTime"]  # Were not testing creationtime atm
-    assert work_group == {
-        "Name": "athena_workgroup",
-        "State": "ENABLED",
-        "Configuration": {
-            "ResultConfiguration": {"OutputLocation": "s3://bucket-name/prefix/"}
-        },
-        "Description": "Test work group",
-    }
+    wg_name = "athena_workgroup"
+    create_basic_workgroup(client=client, name=wg_name)
+
+    try:
+        work_group = client.get_work_group(WorkGroup=wg_name)["WorkGroup"]
+        del work_group["CreationTime"]  # We're not testing creationtime atm
+        assert work_group == {
+            "Name": wg_name,
+            "State": "ENABLED",
+            "Configuration": {
+                "ResultConfiguration": {"OutputLocation": "s3://bucket-name/prefix/"},
+                "EnforceWorkGroupConfiguration": True,
+                "PublishCloudWatchMetricsEnabled": False,
+                "RequesterPaysEnabled": False,
+                "EngineVersion": {
+                    "SelectedEngineVersion": "AUTO",
+                    "EffectiveEngineVersion": "Athena engine version 3",
+                },
+                "EnableMinimumEncryptionConfiguration": False,
+            },
+            "Description": "Test work group",
+        }
+    finally:
+        client.delete_work_group(WorkGroup=wg_name)
 
 
 @mock_aws
diff --git a/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py b/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
index 4320b1126..d34e6e4d5 100644
--- a/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
+++ b/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
@@ -239,16 +239,16 @@ dummy_output_template = {
     "AWSTemplateFormatVersion": "2010-09-09",
     "Description": "Stack 1",
     "Resources": {
-        "Instance": {
-            "Type": "AWS::EC2::Instance",
-            "Properties": {"ImageId": EXAMPLE_AMI_ID},
+        "mybucket": {
+            "Type": "AWS::S3::Bucket",
+            "Properties": {"Tags": [{"Key": "type", "Value": "testbucket"}]},
         }
     },
     "Outputs": {
         "StackVPC": {
             "Description": "The ID of the VPC",
             "Value": "VPCID",
-            "Export": {"Name": "My VPC ID"},
+            "Export": {"Name": "My-VPC-ID"},
         }
     },
 }
@@ -259,7 +259,7 @@ dummy_import_template = {
         "Queue": {
             "Type": "AWS::SQS::Queue",
             "Properties": {
-                "QueueName": {"Fn::ImportValue": "My VPC ID"},
+                "QueueName": {"Fn::ImportValue": "My-VPC-ID"},
                 "VisibilityTimeout": 60,
             },
         }
@@ -2325,23 +2325,33 @@ def test_stack_events():
     assert exp_metadata.get("HTTPStatusCode") == 400
 
 
-@mock_aws
+@pytest.mark.aws_verified
+@aws_verified
 def test_list_exports():
     cf_client = boto3.client("cloudformation", region_name=REGION_NAME)
     cf_resource = boto3.resource("cloudformation", region_name=REGION_NAME)
+    stack_name = f"mototest-with-exports-{str(uuid.uuid4())[0:6]}"
     stack = cf_resource.create_stack(
-        StackName=TEST_STACK_NAME, TemplateBody=dummy_output_template_json
+        StackName=stack_name, TemplateBody=dummy_output_template_json
     )
+    waiter = cf_client.get_waiter("stack_create_complete")
+    waiter.wait(StackName=stack_name)
     output_value = "VPCID"
-    exports = cf_client.list_exports()["Exports"]
-
-    assert len(stack.outputs) == 1
-    assert stack.outputs[0]["OutputValue"] == output_value
-
-    assert len(exports) == 1
-    assert exports[0]["ExportingStackId"] == stack.stack_id
-    assert exports[0]["Name"] == "My VPC ID"
-    assert exports[0]["Value"] == output_value
+    try:
+        assert len(stack.outputs) == 1
+        assert stack.outputs[0]["OutputKey"] == "StackVPC"
+        assert stack.outputs[0]["OutputValue"] == output_value
+        assert stack.outputs[0]["Description"] == "The ID of the VPC"
+        assert stack.outputs[0]["ExportName"] == "My-VPC-ID"
+
+        exports = cf_client.list_exports()["Exports"]
+
+        assert len(exports) == 1
+        assert exports[0]["ExportingStackId"] == stack.stack_id
+        assert exports[0]["Name"] == "My-VPC-ID"
+        assert exports[0]["Value"] == output_value
+    finally:
+        cf_client.delete_stack(StackName=stack_name)
 
 
 @mock_aws
diff --git a/tests/test_dynamodb/test_dynamodb.py b/tests/test_dynamodb/test_dynamodb.py
index 7624d051d..2be6f32f1 100644
--- a/tests/test_dynamodb/test_dynamodb.py
+++ b/tests/test_dynamodb/test_dynamodb.py
@@ -74,6 +74,45 @@ def test_describe_missing_table_boto3():
     )
 
 
+@mock_aws
+def test_describe_table_using_arn():
+    name = "TestTable"
+    conn = boto3.client("dynamodb", region_name="us-west-2")
+    table_arn = conn.create_table(
+        TableName=name,
+        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )["TableDescription"]["TableArn"]
+
+    conn.describe_table(TableName=table_arn)
+
+    conn.put_item(TableName=table_arn, Item={"id": {"S": "n/a"}})
+    conn.get_item(TableName=table_arn, Key={"id": {"S": "n/a"}})
+    conn.query(
+        TableName=table_arn,
+        KeyConditionExpression="id = :val",
+        ExpressionAttributeValues={":val": {"S": "n/a"}},
+    )
+    conn.scan(TableName=table_arn)
+    conn.delete_item(TableName=table_arn, Key={"id": {"S": "n/a"}})
+
+    conn.update_table(
+        TableName=table_arn,
+        StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_IMAGE"},
+    )
+    conn.update_time_to_live(
+        TableName=table_arn,
+        TimeToLiveSpecification={"Enabled": False, "AttributeName": "a"},
+    )
+    conn.describe_time_to_live(TableName=table_arn)
+
+    conn.delete_table(TableName=table_arn)
+
+    with pytest.raises(ClientError):
+        conn.describe_table(TableName=table_arn)
+
+
 @mock_aws
 def test_list_table_tags():
     name = "TestTable"
@@ -3004,584 +3043,6 @@ def test_query_catches_when_no_filters():
     )
 
 
-@mock_aws
-def test_invalid_transact_get_items():
-    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test1",
-        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
-        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
-        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
-    )
-    table = dynamodb.Table("test1")
-    table.put_item(Item={"id": "1", "val": "1"})
-    table.put_item(Item={"id": "1", "val": "2"})
-
-    client = boto3.client("dynamodb", region_name="us-east-1")
-
-    with pytest.raises(ClientError) as ex:
-        client.transact_get_items(
-            TransactItems=[
-                {"Get": {"Key": {"id": {"S": "1"}}, "TableName": "test1"}}
-                for i in range(26)
-            ]
-        )
-
-    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
-    assert (
-        "Member must have length less than or equal to 25"
-        in ex.value.response["Error"]["Message"]
-    )
-
-    with pytest.raises(ClientError) as ex:
-        client.transact_get_items(
-            TransactItems=[
-                {"Get": {"Key": {"id": {"S": "1"}}, "TableName": "test1"}},
-                {"Get": {"Key": {"id": {"S": "1"}}, "TableName": "non_exists_table"}},
-            ]
-        )
-
-    assert ex.value.response["Error"]["Code"] == "ResourceNotFoundException"
-    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
-    assert ex.value.response["Error"]["Message"] == "Requested resource not found"
-
-
-@mock_aws
-def test_valid_transact_get_items():
-    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test1",
-        KeySchema=[
-            {"AttributeName": "id", "KeyType": "HASH"},
-            {"AttributeName": "sort_key", "KeyType": "RANGE"},
-        ],
-        AttributeDefinitions=[
-            {"AttributeName": "id", "AttributeType": "S"},
-            {"AttributeName": "sort_key", "AttributeType": "S"},
-        ],
-        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
-    )
-    table1 = dynamodb.Table("test1")
-    table1.put_item(Item={"id": "1", "sort_key": "1"})
-    table1.put_item(Item={"id": "1", "sort_key": "2"})
-
-    dynamodb.create_table(
-        TableName="test2",
-        KeySchema=[
-            {"AttributeName": "id", "KeyType": "HASH"},
-            {"AttributeName": "sort_key", "KeyType": "RANGE"},
-        ],
-        AttributeDefinitions=[
-            {"AttributeName": "id", "AttributeType": "S"},
-            {"AttributeName": "sort_key", "AttributeType": "S"},
-        ],
-        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
-    )
-    table2 = dynamodb.Table("test2")
-    table2.put_item(Item={"id": "1", "sort_key": "1"})
-
-    client = boto3.client("dynamodb", region_name="us-east-1")
-    res = client.transact_get_items(
-        TransactItems=[
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
-                    "TableName": "test1",
-                }
-            },
-            {
-                "Get": {
-                    "Key": {"id": {"S": "non_exists_key"}, "sort_key": {"S": "2"}},
-                    "TableName": "test1",
-                }
-            },
-        ]
-    )
-    assert res["Responses"][0]["Item"] == {"id": {"S": "1"}, "sort_key": {"S": "1"}}
-    assert len(res["Responses"]) == 2
-    assert res["Responses"][1] == {}
-
-    res = client.transact_get_items(
-        TransactItems=[
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
-                    "TableName": "test1",
-                }
-            },
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
-                    "TableName": "test1",
-                }
-            },
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
-                    "TableName": "test2",
-                }
-            },
-        ]
-    )
-
-    assert res["Responses"][0]["Item"] == {"id": {"S": "1"}, "sort_key": {"S": "1"}}
-
-    assert res["Responses"][1]["Item"] == {"id": {"S": "1"}, "sort_key": {"S": "2"}}
-
-    assert res["Responses"][2]["Item"] == {"id": {"S": "1"}, "sort_key": {"S": "1"}}
-
-    res = client.transact_get_items(
-        TransactItems=[
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
-                    "TableName": "test1",
-                }
-            },
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
-                    "TableName": "test1",
-                }
-            },
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
-                    "TableName": "test2",
-                }
-            },
-        ],
-        ReturnConsumedCapacity="TOTAL",
-    )
-
-    assert res["ConsumedCapacity"][0] == {
-        "TableName": "test1",
-        "CapacityUnits": 4.0,
-        "ReadCapacityUnits": 4.0,
-    }
-
-    assert res["ConsumedCapacity"][1] == {
-        "TableName": "test2",
-        "CapacityUnits": 2.0,
-        "ReadCapacityUnits": 2.0,
-    }
-
-    res = client.transact_get_items(
-        TransactItems=[
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
-                    "TableName": "test1",
-                }
-            },
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
-                    "TableName": "test1",
-                }
-            },
-            {
-                "Get": {
-                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
-                    "TableName": "test2",
-                }
-            },
-        ],
-        ReturnConsumedCapacity="INDEXES",
-    )
-
-    assert res["ConsumedCapacity"][0] == {
-        "TableName": "test1",
-        "CapacityUnits": 4.0,
-        "ReadCapacityUnits": 4.0,
-        "Table": {"CapacityUnits": 4.0, "ReadCapacityUnits": 4.0},
-    }
-
-    assert res["ConsumedCapacity"][1] == {
-        "TableName": "test2",
-        "CapacityUnits": 2.0,
-        "ReadCapacityUnits": 2.0,
-        "Table": {"CapacityUnits": 2.0, "ReadCapacityUnits": 2.0},
-    }
-
-
-@mock_aws
-def test_transact_write_items_put():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
-    )
-    # Put multiple items
-    dynamodb.transact_write_items(
-        TransactItems=[
-            {
-                "Put": {
-                    "Item": {"id": {"S": f"foo{str(i)}"}, "foo": {"S": "bar"}},
-                    "TableName": "test-table",
-                }
-            }
-            for i in range(0, 5)
-        ]
-    )
-    # Assert all are present
-    items = dynamodb.scan(TableName="test-table")["Items"]
-    assert len(items) == 5
-
-
-@pytest.mark.aws_verified
-@dynamodb_aws_verified()
-def test_transact_write_items_put_conditional_expressions(table_name=None):
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-
-    dynamodb.put_item(TableName=table_name, Item={"pk": {"S": "foo2"}})
-    # Put multiple items
-    with pytest.raises(ClientError) as ex:
-        dynamodb.transact_write_items(
-            TransactItems=[
-                {
-                    "Put": {
-                        "Item": {
-                            "pk": {"S": f"foo{i}"},
-                            "foo": {"S": "bar"},
-                        },
-                        "TableName": table_name,
-                        "ConditionExpression": "#i <> :i",
-                        "ExpressionAttributeNames": {"#i": "pk"},
-                        "ExpressionAttributeValues": {
-                            ":i": {
-                                "S": "foo2"
-                            }  # This item already exist, so the ConditionExpression should fail
-                        },
-                    }
-                }
-                for i in range(0, 5)
-            ]
-        )
-    # Assert the exception is correct
-    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
-    reasons = ex.value.response["CancellationReasons"]
-    assert len(reasons) == 5
-    assert {
-        "Code": "ConditionalCheckFailed",
-        "Message": "The conditional request failed",
-    } in reasons
-    assert {"Code": "None"} in reasons
-    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
-    # Assert all are present
-    items = dynamodb.scan(TableName=table_name)["Items"]
-    assert len(items) == 1
-    assert items[0] == {"pk": {"S": "foo2"}}
-
-
-@pytest.mark.aws_verified
-@dynamodb_aws_verified()
-def test_transact_write_items_failure__return_item(table_name=None):
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    dynamodb.put_item(TableName=table_name, Item={"pk": {"S": "foo2"}})
-    # Put multiple items
-    with pytest.raises(ClientError) as ex:
-        dynamodb.transact_write_items(
-            TransactItems=[
-                {
-                    "Put": {
-                        "Item": {
-                            "pk": {"S": f"foo{i}"},
-                            "foo": {"S": "bar"},
-                        },
-                        "TableName": table_name,
-                        "ConditionExpression": "#i <> :i",
-                        "ExpressionAttributeNames": {"#i": "pk"},
-                        # This man right here - should return item as part of error message
-                        "ReturnValuesOnConditionCheckFailure": "ALL_OLD",
-                        "ExpressionAttributeValues": {
-                            ":i": {
-                                "S": "foo2"
-                            }  # This item already exist, so the ConditionExpression should fail
-                        },
-                    }
-                }
-                for i in range(0, 5)
-            ]
-        )
-    # Assert the exception is correct
-    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
-    reasons = ex.value.response["CancellationReasons"]
-    assert len(reasons) == 5
-    assert {
-        "Code": "ConditionalCheckFailed",
-        "Message": "The conditional request failed",
-        "Item": {"pk": {"S": "foo2"}},
-    } in reasons
-    assert {"Code": "None"} in reasons
-    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
-    # Assert all are present
-    items = dynamodb.scan(TableName=table_name)["Items"]
-    assert len(items) == 1
-    assert items[0] == {"pk": {"S": "foo2"}}
-
-
-@mock_aws
-def test_transact_write_items_conditioncheck_passes():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
-    )
-    # 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_exists(#e)",
-                    "ExpressionAttributeNames": {"#e": "email_address"},
-                }
-            },
-            {
-                "Put": {
-                    "Item": {
-                        "id": {"S": "bar"},
-                        "email_address": {"S": "bar@moto.com"},
-                    },
-                    "TableName": "test-table",
-                }
-            },
-        ]
-    )
-    # Assert all are present
-    items = dynamodb.scan(TableName="test-table")["Items"]
-    assert len(items) == 2
-    assert {"email_address": {"S": "foo@moto.com"}, "id": {"S": "foo"}} in items
-    assert {"email_address": {"S": "bar@moto.com"}, "id": {"S": "bar"}} in items
-
-
-@mock_aws
-def test_transact_write_items_conditioncheck_fails():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    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"}},
-    )
-    # 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(
-            TransactItems=[
-                {
-                    "ConditionCheck": {
-                        "Key": {"id": {"S": "foo"}},
-                        "TableName": "test-table",
-                        "ConditionExpression": "attribute_exists(#e)",
-                        "ExpressionAttributeNames": {"#e": "email_address"},
-                    }
-                },
-                {
-                    "Put": {
-                        "Item": {
-                            "id": {"S": "bar"},
-                            "email_address": {"S": "bar@moto.com"},
-                        },
-                        "TableName": "test-table",
-                    }
-                },
-            ]
-        )
-    # Assert the exception is correct
-    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
-    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
-
-    # Assert the original item is still present
-    items = dynamodb.scan(TableName="test-table")["Items"]
-    assert len(items) == 1
-    assert items[0] == {"id": {"S": "foo"}}
-
-
-@mock_aws
-def test_transact_write_items_delete():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
-    )
-    # Insert an item
-    dynamodb.put_item(TableName="test-table", Item={"id": {"S": "foo"}})
-    # Delete the item
-    dynamodb.transact_write_items(
-        TransactItems=[
-            {"Delete": {"Key": {"id": {"S": "foo"}}, "TableName": "test-table"}}
-        ]
-    )
-    # Assert the item is deleted
-    items = dynamodb.scan(TableName="test-table")["Items"]
-    assert len(items) == 0
-
-
-@mock_aws
-def test_transact_write_items_delete_with_successful_condition_expression():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    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"}})
-    # ConditionExpression will pass - no email address has been specified yet
-    dynamodb.transact_write_items(
-        TransactItems=[
-            {
-                "Delete": {
-                    "Key": {"id": {"S": "foo"}},
-                    "TableName": "test-table",
-                    "ConditionExpression": "attribute_not_exists(#e)",
-                    "ExpressionAttributeNames": {"#e": "email_address"},
-                }
-            }
-        ]
-    )
-    # Assert the item is deleted
-    items = dynamodb.scan(TableName="test-table")["Items"]
-    assert len(items) == 0
-
-
-@mock_aws
-def test_transact_write_items_delete_with_failed_condition_expression():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
-    )
-    # Insert an item with email address
-    dynamodb.put_item(
-        TableName="test-table",
-        Item={"id": {"S": "foo"}, "email_address": {"S": "test@moto.com"}},
-    )
-    # Try to delete an item that does not have an email address
-    # ConditionCheck should fail
-    with pytest.raises(ClientError) as ex:
-        dynamodb.transact_write_items(
-            TransactItems=[
-                {
-                    "Delete": {
-                        "Key": {"id": {"S": "foo"}},
-                        "TableName": "test-table",
-                        "ConditionExpression": "attribute_not_exists(#e)",
-                        "ExpressionAttributeNames": {"#e": "email_address"},
-                    }
-                }
-            ]
-        )
-    # Assert the exception is correct
-    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
-    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
-    # 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"}}
-
-
-@mock_aws
-def test_transact_write_items_update():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
-    )
-    # Insert an item
-    dynamodb.put_item(TableName="test-table", Item={"id": {"S": "foo"}})
-    # Update the item
-    dynamodb.transact_write_items(
-        TransactItems=[
-            {
-                "Update": {
-                    "Key": {"id": {"S": "foo"}},
-                    "TableName": "test-table",
-                    "UpdateExpression": "SET #e = :v",
-                    "ExpressionAttributeNames": {"#e": "email_address"},
-                    "ExpressionAttributeValues": {":v": {"S": "test@moto.com"}},
-                }
-            }
-        ]
-    )
-    # Assert the item is updated
-    items = dynamodb.scan(TableName="test-table")["Items"]
-    assert len(items) == 1
-    assert items[0] == {"id": {"S": "foo"}, "email_address": {"S": "test@moto.com"}}
-
-
-@mock_aws
-def test_transact_write_items_update_with_failed_condition_expression():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
-    )
-    # Insert an item with email address
-    dynamodb.put_item(
-        TableName="test-table",
-        Item={"id": {"S": "foo"}, "email_address": {"S": "test@moto.com"}},
-    )
-    # Try to update an item that does not have an email address
-    # ConditionCheck should fail
-    with pytest.raises(ClientError) as ex:
-        dynamodb.transact_write_items(
-            TransactItems=[
-                {
-                    "Update": {
-                        "Key": {"id": {"S": "foo"}},
-                        "TableName": "test-table",
-                        "UpdateExpression": "SET #e = :v",
-                        "ConditionExpression": "attribute_not_exists(#e)",
-                        "ExpressionAttributeNames": {"#e": "email_address"},
-                        "ExpressionAttributeValues": {":v": {"S": "update@moto.com"}},
-                    }
-                }
-            ]
-        )
-    # Assert the exception is correct
-    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
-    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
-    # 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"}}
-
-
 @mock_aws
 def test_dynamodb_max_1mb_limit():
     ddb = boto3.resource("dynamodb", region_name="eu-west-1")
@@ -4090,52 +3551,6 @@ def test_update_item_add_to_non_existent_number_set():
     assert updated_item["s_i"]["NS"] == ["3"]
 
 
-@mock_aws
-def test_transact_write_items_fails_with_transaction_canceled_exception():
-    table_schema = {
-        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
-        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
-    }
-    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
-    dynamodb.create_table(
-        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
-    )
-    # Insert one item
-    dynamodb.put_item(TableName="test-table", Item={"id": {"S": "foo"}})
-    # Update two items, the one that exists and another that doesn't
-    with pytest.raises(ClientError) as ex:
-        dynamodb.transact_write_items(
-            TransactItems=[
-                {
-                    "Update": {
-                        "Key": {"id": {"S": "foo"}},
-                        "TableName": "test-table",
-                        "UpdateExpression": "SET #k = :v",
-                        "ConditionExpression": "attribute_exists(id)",
-                        "ExpressionAttributeNames": {"#k": "key"},
-                        "ExpressionAttributeValues": {":v": {"S": "value"}},
-                    }
-                },
-                {
-                    "Update": {
-                        "Key": {"id": {"S": "doesnotexist"}},
-                        "TableName": "test-table",
-                        "UpdateExpression": "SET #e = :v",
-                        "ConditionExpression": "attribute_exists(id)",
-                        "ExpressionAttributeNames": {"#e": "key"},
-                        "ExpressionAttributeValues": {":v": {"S": "value"}},
-                    }
-                },
-            ]
-        )
-    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
-    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
-    assert (
-        ex.value.response["Error"]["Message"]
-        == "Transaction cancelled, please refer cancellation reasons for specific reasons [None, ConditionalCheckFailed]"
-    )
-
-
 @mock_aws
 def test_gsi_projection_type_keys_only():
     table_schema = {
@@ -4329,31 +3744,6 @@ def test_set_attribute_is_dropped_if_empty_after_update_expression(table_name=No
     assert items == [{"pk": {"S": "item1"}}]
 
 
-@mock_aws
-def test_transact_get_items_should_return_empty_map_for_non_existent_item():
-    client = boto3.client("dynamodb", region_name="us-west-2")
-    table_name = "test-table"
-    key_schema = [{"AttributeName": "id", "KeyType": "HASH"}]
-    attribute_definitions = [{"AttributeName": "id", "AttributeType": "S"}]
-    client.create_table(
-        TableName=table_name,
-        KeySchema=key_schema,
-        AttributeDefinitions=attribute_definitions,
-        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
-    )
-    item = {"id": {"S": "1"}}
-    client.put_item(TableName=table_name, Item=item)
-    items = client.transact_get_items(
-        TransactItems=[
-            {"Get": {"Key": {"id": {"S": "1"}}, "TableName": table_name}},
-            {"Get": {"Key": {"id": {"S": "2"}}, "TableName": table_name}},
-        ]
-    ).get("Responses", [])
-    assert len(items) == 2
-    assert items[0] == {"Item": item}
-    assert items[1] == {}
-
-
 @mock_aws
 def test_dynamodb_update_item_fails_on_string_sets():
     dynamodb = boto3.resource("dynamodb", region_name="eu-west-1")
@@ -4625,6 +4015,21 @@ def test_create_backup():
     assert isinstance(details["BackupCreationDateTime"], datetime)
 
 
+@mock_aws
+def test_create_backup_using_arn():
+    client = boto3.client("dynamodb", "us-east-1")
+    table_arn = client.create_table(
+        TableName="test-table",
+        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )["TableDescription"]["TableArn"]
+    client.create_backup(TableName=table_arn, BackupName="n/a")
+
+    backups = client.list_backups(TableName=table_arn)["BackupSummaries"]
+    assert len(backups) == 1
+
+
 @mock_aws
 def test_create_multiple_backups_with_same_name():
     client = boto3.client("dynamodb", "us-east-1")
diff --git a/tests/test_dynamodb/test_dynamodb_batch_write.py b/tests/test_dynamodb/test_dynamodb_batch_write.py
index 7ee0a27ac..f530de651 100644
--- a/tests/test_dynamodb/test_dynamodb_batch_write.py
+++ b/tests/test_dynamodb/test_dynamodb_batch_write.py
@@ -77,3 +77,31 @@ def test_batch_write_item_to_multiple_tables():
             except Exception as e:
                 print(f"Failed to delete table {name}")  # noqa
                 print(e)  # noqa
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified()
+def test_batch_write_using_arn(table_name=None):
+    ddb_client = boto3.client("dynamodb", region_name="us-east-1")
+    table_arn = ddb_client.describe_table(TableName=table_name)["Table"]["TableArn"]
+
+    ddb_client.transact_write_items(
+        TransactItems=[
+            {
+                "Update": {
+                    "TableName": table_arn,
+                    "Key": {"pk": {"S": "test"}},
+                    "UpdateExpression": "SET xxx = :xxx",
+                    "ExpressionAttributeValues": {":xxx": {"S": "123"}},
+                }
+            }
+        ]
+    )
+
+    results = ddb_client.scan(TableName=table_arn)["Items"]
+    assert results == [{"pk": {"S": "test"}, "xxx": {"S": "123"}}]
+
+    items = ddb_client.batch_get_item(
+        RequestItems={table_arn: {"Keys": [{"pk": {"S": "test"}}]}}
+    )["Responses"]
+    assert items == {table_arn: [{"pk": {"S": "test"}, "xxx": {"S": "123"}}]}
diff --git a/tests/test_dynamodb/test_dynamodb_transact.py b/tests/test_dynamodb/test_dynamodb_transact.py
new file mode 100644
index 000000000..13e635302
--- /dev/null
+++ b/tests/test_dynamodb/test_dynamodb_transact.py
@@ -0,0 +1,677 @@
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+
+from moto import mock_aws
+
+from . import dynamodb_aws_verified
+
+
+@mock_aws
+def test_invalid_transact_get_items():
+    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test1",
+        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    table = dynamodb.Table("test1")
+    table.put_item(Item={"id": "1", "val": "1"})
+    table.put_item(Item={"id": "1", "val": "2"})
+
+    client = boto3.client("dynamodb", region_name="us-east-1")
+
+    with pytest.raises(ClientError) as ex:
+        client.transact_get_items(
+            TransactItems=[
+                {"Get": {"Key": {"id": {"S": "1"}}, "TableName": "test1"}}
+                for i in range(26)
+            ]
+        )
+
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    assert (
+        "Member must have length less than or equal to 25"
+        in ex.value.response["Error"]["Message"]
+    )
+
+    with pytest.raises(ClientError) as ex:
+        client.transact_get_items(
+            TransactItems=[
+                {"Get": {"Key": {"id": {"S": "1"}}, "TableName": "test1"}},
+                {"Get": {"Key": {"id": {"S": "1"}}, "TableName": "non_exists_table"}},
+            ]
+        )
+
+    assert ex.value.response["Error"]["Code"] == "ResourceNotFoundException"
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    assert ex.value.response["Error"]["Message"] == "Requested resource not found"
+
+
+@mock_aws
+def test_transact_get_items_should_return_empty_map_for_non_existent_item():
+    client = boto3.client("dynamodb", region_name="us-west-2")
+    table_name = "test-table"
+    key_schema = [{"AttributeName": "id", "KeyType": "HASH"}]
+    attribute_definitions = [{"AttributeName": "id", "AttributeType": "S"}]
+    client.create_table(
+        TableName=table_name,
+        KeySchema=key_schema,
+        AttributeDefinitions=attribute_definitions,
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    item = {"id": {"S": "1"}}
+    client.put_item(TableName=table_name, Item=item)
+    items = client.transact_get_items(
+        TransactItems=[
+            {"Get": {"Key": {"id": {"S": "1"}}, "TableName": table_name}},
+            {"Get": {"Key": {"id": {"S": "2"}}, "TableName": table_name}},
+        ]
+    ).get("Responses", [])
+    assert len(items) == 2
+    assert items[0] == {"Item": item}
+    assert items[1] == {}
+
+
+@mock_aws
+def test_valid_transact_get_items():
+    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test1",
+        KeySchema=[
+            {"AttributeName": "id", "KeyType": "HASH"},
+            {"AttributeName": "sort_key", "KeyType": "RANGE"},
+        ],
+        AttributeDefinitions=[
+            {"AttributeName": "id", "AttributeType": "S"},
+            {"AttributeName": "sort_key", "AttributeType": "S"},
+        ],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    table1 = dynamodb.Table("test1")
+    table1.put_item(Item={"id": "1", "sort_key": "1"})
+    table1.put_item(Item={"id": "1", "sort_key": "2"})
+
+    dynamodb.create_table(
+        TableName="test2",
+        KeySchema=[
+            {"AttributeName": "id", "KeyType": "HASH"},
+            {"AttributeName": "sort_key", "KeyType": "RANGE"},
+        ],
+        AttributeDefinitions=[
+            {"AttributeName": "id", "AttributeType": "S"},
+            {"AttributeName": "sort_key", "AttributeType": "S"},
+        ],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    table2 = dynamodb.Table("test2")
+    table2.put_item(Item={"id": "1", "sort_key": "1"})
+
+    client = boto3.client("dynamodb", region_name="us-east-1")
+    res = client.transact_get_items(
+        TransactItems=[
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
+                    "TableName": "test1",
+                }
+            },
+            {
+                "Get": {
+                    "Key": {"id": {"S": "non_exists_key"}, "sort_key": {"S": "2"}},
+                    "TableName": "test1",
+                }
+            },
+        ]
+    )
+    assert res["Responses"][0]["Item"] == {"id": {"S": "1"}, "sort_key": {"S": "1"}}
+    assert len(res["Responses"]) == 2
+    assert res["Responses"][1] == {}
+
+    res = client.transact_get_items(
+        TransactItems=[
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
+                    "TableName": "test1",
+                }
+            },
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
+                    "TableName": "test1",
+                }
+            },
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
+                    "TableName": "test2",
+                }
+            },
+        ]
+    )
+
+    assert res["Responses"][0]["Item"] == {"id": {"S": "1"}, "sort_key": {"S": "1"}}
+
+    assert res["Responses"][1]["Item"] == {"id": {"S": "1"}, "sort_key": {"S": "2"}}
+
+    assert res["Responses"][2]["Item"] == {"id": {"S": "1"}, "sort_key": {"S": "1"}}
+
+    res = client.transact_get_items(
+        TransactItems=[
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
+                    "TableName": "test1",
+                }
+            },
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
+                    "TableName": "test1",
+                }
+            },
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
+                    "TableName": "test2",
+                }
+            },
+        ],
+        ReturnConsumedCapacity="TOTAL",
+    )
+
+    assert res["ConsumedCapacity"][0] == {
+        "TableName": "test1",
+        "CapacityUnits": 4.0,
+        "ReadCapacityUnits": 4.0,
+    }
+
+    assert res["ConsumedCapacity"][1] == {
+        "TableName": "test2",
+        "CapacityUnits": 2.0,
+        "ReadCapacityUnits": 2.0,
+    }
+
+    res = client.transact_get_items(
+        TransactItems=[
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
+                    "TableName": "test1",
+                }
+            },
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
+                    "TableName": "test1",
+                }
+            },
+            {
+                "Get": {
+                    "Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
+                    "TableName": "test2",
+                }
+            },
+        ],
+        ReturnConsumedCapacity="INDEXES",
+    )
+
+    assert res["ConsumedCapacity"][0] == {
+        "TableName": "test1",
+        "CapacityUnits": 4.0,
+        "ReadCapacityUnits": 4.0,
+        "Table": {"CapacityUnits": 4.0, "ReadCapacityUnits": 4.0},
+    }
+
+    assert res["ConsumedCapacity"][1] == {
+        "TableName": "test2",
+        "CapacityUnits": 2.0,
+        "ReadCapacityUnits": 2.0,
+        "Table": {"CapacityUnits": 2.0, "ReadCapacityUnits": 2.0},
+    }
+
+
+@mock_aws
+def test_transact_write_items_put():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
+    )
+    # Put multiple items
+    dynamodb.transact_write_items(
+        TransactItems=[
+            {
+                "Put": {
+                    "Item": {"id": {"S": f"foo{str(i)}"}, "foo": {"S": "bar"}},
+                    "TableName": "test-table",
+                }
+            }
+            for i in range(0, 5)
+        ]
+    )
+    # Assert all are present
+    items = dynamodb.scan(TableName="test-table")["Items"]
+    assert len(items) == 5
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified()
+def test_transact_write_items_put_conditional_expressions(table_name=None):
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+
+    dynamodb.put_item(TableName=table_name, Item={"pk": {"S": "foo2"}})
+    # Put multiple items
+    with pytest.raises(ClientError) as ex:
+        dynamodb.transact_write_items(
+            TransactItems=[
+                {
+                    "Put": {
+                        "Item": {
+                            "pk": {"S": f"foo{i}"},
+                            "foo": {"S": "bar"},
+                        },
+                        "TableName": table_name,
+                        "ConditionExpression": "#i <> :i",
+                        "ExpressionAttributeNames": {"#i": "pk"},
+                        "ExpressionAttributeValues": {
+                            ":i": {
+                                "S": "foo2"
+                            }  # This item already exist, so the ConditionExpression should fail
+                        },
+                    }
+                }
+                for i in range(0, 5)
+            ]
+        )
+    # Assert the exception is correct
+    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
+    reasons = ex.value.response["CancellationReasons"]
+    assert len(reasons) == 5
+    assert {
+        "Code": "ConditionalCheckFailed",
+        "Message": "The conditional request failed",
+    } in reasons
+    assert {"Code": "None"} in reasons
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    # Assert all are present
+    items = dynamodb.scan(TableName=table_name)["Items"]
+    assert len(items) == 1
+    assert items[0] == {"pk": {"S": "foo2"}}
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified()
+def test_transact_write_items_failure__return_item(table_name=None):
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.put_item(TableName=table_name, Item={"pk": {"S": "foo2"}})
+    # Put multiple items
+    with pytest.raises(ClientError) as ex:
+        dynamodb.transact_write_items(
+            TransactItems=[
+                {
+                    "Put": {
+                        "Item": {
+                            "pk": {"S": f"foo{i}"},
+                            "foo": {"S": "bar"},
+                        },
+                        "TableName": table_name,
+                        "ConditionExpression": "#i <> :i",
+                        "ExpressionAttributeNames": {"#i": "pk"},
+                        # This man right here - should return item as part of error message
+                        "ReturnValuesOnConditionCheckFailure": "ALL_OLD",
+                        "ExpressionAttributeValues": {
+                            ":i": {
+                                "S": "foo2"
+                            }  # This item already exist, so the ConditionExpression should fail
+                        },
+                    }
+                }
+                for i in range(0, 5)
+            ]
+        )
+    # Assert the exception is correct
+    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
+    reasons = ex.value.response["CancellationReasons"]
+    assert len(reasons) == 5
+    assert {
+        "Code": "ConditionalCheckFailed",
+        "Message": "The conditional request failed",
+        "Item": {"pk": {"S": "foo2"}},
+    } in reasons
+    assert {"Code": "None"} in reasons
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    # Assert all are present
+    items = dynamodb.scan(TableName=table_name)["Items"]
+    assert len(items) == 1
+    assert items[0] == {"pk": {"S": "foo2"}}
+
+
+@mock_aws
+def test_transact_write_items_conditioncheck_passes():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
+    )
+    # 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_exists(#e)",
+                    "ExpressionAttributeNames": {"#e": "email_address"},
+                }
+            },
+            {
+                "Put": {
+                    "Item": {
+                        "id": {"S": "bar"},
+                        "email_address": {"S": "bar@moto.com"},
+                    },
+                    "TableName": "test-table",
+                }
+            },
+        ]
+    )
+    # Assert all are present
+    items = dynamodb.scan(TableName="test-table")["Items"]
+    assert len(items) == 2
+    assert {"email_address": {"S": "foo@moto.com"}, "id": {"S": "foo"}} in items
+    assert {"email_address": {"S": "bar@moto.com"}, "id": {"S": "bar"}} in items
+
+
+@mock_aws
+def test_transact_write_items_conditioncheck_fails():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    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"}},
+    )
+    # 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(
+            TransactItems=[
+                {
+                    "ConditionCheck": {
+                        "Key": {"id": {"S": "foo"}},
+                        "TableName": "test-table",
+                        "ConditionExpression": "attribute_exists(#e)",
+                        "ExpressionAttributeNames": {"#e": "email_address"},
+                    }
+                },
+                {
+                    "Put": {
+                        "Item": {
+                            "id": {"S": "bar"},
+                            "email_address": {"S": "bar@moto.com"},
+                        },
+                        "TableName": "test-table",
+                    }
+                },
+            ]
+        )
+    # Assert the exception is correct
+    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+
+    # Assert the original item is still present
+    items = dynamodb.scan(TableName="test-table")["Items"]
+    assert len(items) == 1
+    assert items[0] == {"id": {"S": "foo"}}
+
+
+@mock_aws
+def test_transact_write_items_delete():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
+    )
+    # Insert an item
+    dynamodb.put_item(TableName="test-table", Item={"id": {"S": "foo"}})
+    # Delete the item
+    dynamodb.transact_write_items(
+        TransactItems=[
+            {"Delete": {"Key": {"id": {"S": "foo"}}, "TableName": "test-table"}}
+        ]
+    )
+    # Assert the item is deleted
+    items = dynamodb.scan(TableName="test-table")["Items"]
+    assert len(items) == 0
+
+
+@mock_aws
+def test_transact_write_items_delete_with_successful_condition_expression():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    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"}})
+    # ConditionExpression will pass - no email address has been specified yet
+    dynamodb.transact_write_items(
+        TransactItems=[
+            {
+                "Delete": {
+                    "Key": {"id": {"S": "foo"}},
+                    "TableName": "test-table",
+                    "ConditionExpression": "attribute_not_exists(#e)",
+                    "ExpressionAttributeNames": {"#e": "email_address"},
+                }
+            }
+        ]
+    )
+    # Assert the item is deleted
+    items = dynamodb.scan(TableName="test-table")["Items"]
+    assert len(items) == 0
+
+
+@mock_aws
+def test_transact_write_items_delete_with_failed_condition_expression():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
+    )
+    # Insert an item with email address
+    dynamodb.put_item(
+        TableName="test-table",
+        Item={"id": {"S": "foo"}, "email_address": {"S": "test@moto.com"}},
+    )
+    # Try to delete an item that does not have an email address
+    # ConditionCheck should fail
+    with pytest.raises(ClientError) as ex:
+        dynamodb.transact_write_items(
+            TransactItems=[
+                {
+                    "Delete": {
+                        "Key": {"id": {"S": "foo"}},
+                        "TableName": "test-table",
+                        "ConditionExpression": "attribute_not_exists(#e)",
+                        "ExpressionAttributeNames": {"#e": "email_address"},
+                    }
+                }
+            ]
+        )
+    # Assert the exception is correct
+    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    # 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"}}
+
+
+@mock_aws
+def test_transact_write_items_update():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
+    )
+    # Insert an item
+    dynamodb.put_item(TableName="test-table", Item={"id": {"S": "foo"}})
+    # Update the item
+    dynamodb.transact_write_items(
+        TransactItems=[
+            {
+                "Update": {
+                    "Key": {"id": {"S": "foo"}},
+                    "TableName": "test-table",
+                    "UpdateExpression": "SET #e = :v",
+                    "ExpressionAttributeNames": {"#e": "email_address"},
+                    "ExpressionAttributeValues": {":v": {"S": "test@moto.com"}},
+                }
+            }
+        ]
+    )
+    # Assert the item is updated
+    items = dynamodb.scan(TableName="test-table")["Items"]
+    assert len(items) == 1
+    assert items[0] == {"id": {"S": "foo"}, "email_address": {"S": "test@moto.com"}}
+
+
+@mock_aws
+def test_transact_write_items_update_with_failed_condition_expression():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
+    )
+    # Insert an item with email address
+    dynamodb.put_item(
+        TableName="test-table",
+        Item={"id": {"S": "foo"}, "email_address": {"S": "test@moto.com"}},
+    )
+    # Try to update an item that does not have an email address
+    # ConditionCheck should fail
+    with pytest.raises(ClientError) as ex:
+        dynamodb.transact_write_items(
+            TransactItems=[
+                {
+                    "Update": {
+                        "Key": {"id": {"S": "foo"}},
+                        "TableName": "test-table",
+                        "UpdateExpression": "SET #e = :v",
+                        "ConditionExpression": "attribute_not_exists(#e)",
+                        "ExpressionAttributeNames": {"#e": "email_address"},
+                        "ExpressionAttributeValues": {":v": {"S": "update@moto.com"}},
+                    }
+                }
+            ]
+        )
+    # Assert the exception is correct
+    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    # 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"}}
+
+
+@mock_aws
+def test_transact_write_items_fails_with_transaction_canceled_exception():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
+    )
+    # Insert one item
+    dynamodb.put_item(TableName="test-table", Item={"id": {"S": "foo"}})
+    # Update two items, the one that exists and another that doesn't
+    with pytest.raises(ClientError) as ex:
+        dynamodb.transact_write_items(
+            TransactItems=[
+                {
+                    "Update": {
+                        "Key": {"id": {"S": "foo"}},
+                        "TableName": "test-table",
+                        "UpdateExpression": "SET #k = :v",
+                        "ConditionExpression": "attribute_exists(id)",
+                        "ExpressionAttributeNames": {"#k": "key"},
+                        "ExpressionAttributeValues": {":v": {"S": "value"}},
+                    }
+                },
+                {
+                    "Update": {
+                        "Key": {"id": {"S": "doesnotexist"}},
+                        "TableName": "test-table",
+                        "UpdateExpression": "SET #e = :v",
+                        "ConditionExpression": "attribute_exists(id)",
+                        "ExpressionAttributeNames": {"#e": "key"},
+                        "ExpressionAttributeValues": {":v": {"S": "value"}},
+                    }
+                },
+            ]
+        )
+    assert ex.value.response["Error"]["Code"] == "TransactionCanceledException"
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "Transaction cancelled, please refer cancellation reasons for specific reasons [None, ConditionalCheckFailed]"
+    )
+
+
+@mock_aws
+def test_transact_using_arns():
+    table_schema = {
+        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
+        "AttributeDefinitions": [{"AttributeName": "id", "AttributeType": "S"}],
+    }
+    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
+    table_arn = dynamodb.create_table(
+        TableName="test-table", BillingMode="PAY_PER_REQUEST", **table_schema
+    )["TableDescription"]["TableArn"]
+
+    dynamodb.transact_write_items(
+        TransactItems=[{"Put": {"Item": {"id": {"S": "foo"}}, "TableName": table_arn}}]
+    )
+
+    items = dynamodb.transact_get_items(
+        TransactItems=[{"Get": {"Key": {"id": {"S": "foo"}}, "TableName": table_arn}}]
+    )["Responses"]
+    assert items == [{"Item": {"id": {"S": "foo"}}}]
diff --git a/tests/test_emr/test_emr.py b/tests/test_emr/test_emr.py
new file mode 100644
index 000000000..aefdcac10
--- /dev/null
+++ b/tests/test_emr/test_emr.py
@@ -0,0 +1,53 @@
+"""Unit tests for emr-supported APIs."""
+
+from datetime import datetime
+
+import boto3
+import pytest
+
+from moto import mock_aws
+
+# See our Development Tips on writing tests for hints on how to write good tests:
+# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
+
+
+@pytest.fixture(name="client")
+def fixture_emr_client():
+    with mock_aws():
+        yield boto3.client("emr", region_name="us-east-2")
+
+
+@mock_aws
+def test_put_and_get_block_public_access_configuration(client):
+    client.put_block_public_access_configuration(
+        BlockPublicAccessConfiguration={
+            "BlockPublicSecurityGroupRules": True,
+            "PermittedPublicSecurityGroupRuleRanges": [
+                {"MinRange": 22, "MaxRange": 443},
+            ],
+        }
+    )
+
+    connection = client.get_block_public_access_configuration()
+    assert connection["BlockPublicAccessConfiguration"]["BlockPublicSecurityGroupRules"]
+    assert (
+        connection["BlockPublicAccessConfiguration"][
+            "PermittedPublicSecurityGroupRuleRanges"
+        ][0]["MinRange"]
+        == 22
+    )
+    assert (
+        connection["BlockPublicAccessConfiguration"][
+            "PermittedPublicSecurityGroupRuleRanges"
+        ][0]["MaxRange"]
+        == 443
+    )
+    assert isinstance(
+        connection["BlockPublicAccessConfigurationMetadata"]["CreatedByArn"], str
+    )
+
+    # botocore automatically parses timestamps into datetimes
+    assert isinstance(
+        connection["BlockPublicAccessConfigurationMetadata"]["CreationDateTime"],
+        datetime,
+    )
diff --git a/tests/test_kms/test_kms_boto3.py b/tests/test_kms/test_kms_boto3.py
index 9e4c69179..455ef368b 100644
--- a/tests/test_kms/test_kms_boto3.py
+++ b/tests/test_kms/test_kms_boto3.py
@@ -134,6 +134,86 @@ def test_create_multi_region_key():
     assert key["KeyMetadata"]["MultiRegion"] is True
 
 
+@mock_aws
+def test_create_multi_region_configuration_key():
+    region_to_replicate_from = "us-east-1"
+    region_to_replicate_to = "us-west-1"
+    region_to_replicate_to_second_region = "us-west-2"
+
+    from_region_client = boto3.client("kms", region_name=region_to_replicate_from)
+    to_region_client_us_west_1 = boto3.client("kms", region_name=region_to_replicate_to)
+    to_region_client_us_west_2 = boto3.client(
+        "kms", region_name=region_to_replicate_to_second_region
+    )
+
+    response = from_region_client.create_key(
+        Policy="my policy",
+        Description="my key",
+        KeyUsage="ENCRYPT_DECRYPT",
+        MultiRegion=True,
+        Tags=[{"TagKey": "project", "TagValue": "moto"}],
+    )
+    key_id = response["KeyMetadata"]["KeyId"]
+
+    first_replica_response = from_region_client.replicate_key(
+        KeyId=key_id, ReplicaRegion=region_to_replicate_to
+    )
+
+    second_replica_response = from_region_client.replicate_key(
+        KeyId=key_id, ReplicaRegion=region_to_replicate_to_second_region
+    )
+
+    third_replica_response = from_region_client.replicate_key(
+        KeyId=key_id, ReplicaRegion=region_to_replicate_to_second_region
+    )
+
+    replica_key_id = first_replica_response["ReplicaKeyMetadata"]["KeyId"]
+    replica_key_second_id = second_replica_response["ReplicaKeyMetadata"]["KeyId"]
+    replica_key_third_id = third_replica_response["ReplicaKeyMetadata"]["KeyId"]
+
+    from_region_describe_key = from_region_client.describe_key(KeyId=key_id)
+    to_region_describe_key = to_region_client_us_west_1.describe_key(
+        KeyId=replica_key_id
+    )
+    to_region_describe_key_second_replica = to_region_client_us_west_2.describe_key(
+        KeyId=replica_key_second_id
+    )
+    to_region_describe_key_third_replica = to_region_client_us_west_2.describe_key(
+        KeyId=replica_key_third_id
+    )
+
+    assert from_region_describe_key["KeyMetadata"]["MultiRegion"] is True
+    assert to_region_describe_key["KeyMetadata"]["MultiRegion"] is True
+    assert to_region_describe_key_second_replica["KeyMetadata"]["MultiRegion"] is True
+    assert (
+        from_region_describe_key["KeyMetadata"]["MultiRegionConfiguration"][
+            "MultiRegionKeyType"
+        ]
+        == "PRIMARY"
+    )
+    assert (
+        to_region_describe_key["KeyMetadata"]["MultiRegionConfiguration"][
+            "MultiRegionKeyType"
+        ]
+        == "REPLICA"
+    )
+    assert (
+        to_region_describe_key_second_replica["KeyMetadata"][
+            "MultiRegionConfiguration"
+        ]["MultiRegionKeyType"]
+        == "REPLICA"
+    )
+    # This test below tests the expected failure of making a replica in an already existing region.
+    assert (
+        len(
+            to_region_describe_key_third_replica["KeyMetadata"][
+                "MultiRegionConfiguration"
+            ]["ReplicaKeys"]
+        )
+        == 2
+    )
+
+
 @mock_aws
 def test_non_multi_region_keys_should_not_have_multi_region_properties():
     conn = boto3.client("kms", region_name="us-east-1")