Repository: getmoto/moto
Pull request #8164 (2024-09-27 02:01:22)

Directory Service - Enable/Disable/Describe LDAPs & Create/Delete/Describe Trusts
Adding several features to Directory Service (DS)

**LDAPs**

- EnableLDAPS
- DisableLDAPS
- DescribeLDAPS

**Trust Relationships**

- CreateTrust
- DeleteTrust
- DescribeTrust


Patch:
diff --git a/docs/docs/services/sns.rst b/docs/docs/services/sns.rst
index 82b8be949..e9fca8a2c 100644
--- a/docs/docs/services/sns.rst
+++ b/docs/docs/services/sns.rst
@@ -51,7 +51,7 @@ sns
 - [X] publish
 - [X] publish_batch
   
-        The MessageStructure and MessageDeduplicationId-parameters have not yet been implemented.
+        The MessageDeduplicationId-parameter have not yet been implemented.
         
 
 - [ ] put_data_protection_policy
diff --git a/moto/batch/models.py b/moto/batch/models.py
index e0f72540f..9352c6944 100644
--- a/moto/batch/models.py
+++ b/moto/batch/models.py
@@ -638,12 +638,17 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         if not self.parameters:
             return command
 
-        new_command = [
-            command_part.replace(f"Ref::{param}", value)
+        return [
+            next(
+                (
+                    command_part.replace(f"Ref::{param}", value)
+                    for param, value in self.parameters.items()
+                    if f"Ref::{param}" in command_part
+                ),
+                command_part,
+            )
             for command_part in command
-            for param, value in self.parameters.items()
         ]
-        return new_command
 
     def run(self) -> None:
         """
diff --git a/moto/ds/exceptions.py b/moto/ds/exceptions.py
index a505a9067..e5af8920d 100644
--- a/moto/ds/exceptions.py
+++ b/moto/ds/exceptions.py
@@ -106,3 +106,12 @@ class ValidationException(JsonRESTError):
 
     def __init__(self, message: str):
         super().__init__("ValidationException", message)
+
+
+class UnsupportedOperationException(JsonRESTError):
+    """Unsupported Operation"""
+
+    code = 400
+
+    def __init__(self, message: str):
+        super().__init__("UnsupportedOperationException", message)
diff --git a/moto/ds/models.py b/moto/ds/models.py
index 0145259b8..25f0e78c2 100644
--- a/moto/ds/models.py
+++ b/moto/ds/models.py
@@ -12,6 +12,7 @@ from moto.ds.exceptions import (
     EntityDoesNotExistException,
     InvalidParameterException,
     TagLimitExceededException,
+    UnsupportedOperationException,
     ValidationException,
 )
 from moto.ds.utils import PAGINATION_MODEL
@@ -23,6 +24,61 @@ from moto.utilities.paginator import paginate
 from moto.utilities.tagging_service import TaggingService
 
 
+class LdapsSettingInfo(BaseModel):
+    def __init__(self) -> None:
+        self.last_updated_date_time = unix_time()
+        self.ldaps_status = "Enabled"
+        self.ldaps_status_reason = ""
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "LastUpdatedDateTime": self.last_updated_date_time,
+            "LDAPSStatus": self.ldaps_status,
+            "LDAPSStatusReason": self.ldaps_status_reason,
+        }
+
+
+class Trust(BaseModel):
+    def __init__(
+        self,
+        directory_id: str,
+        remote_domain_name: str,
+        trust_password: str,
+        trust_direction: str,
+        trust_type: Optional[str],
+        conditional_forwarder_ip_addrs: Optional[List[str]],
+        selective_auth: Optional[str],
+    ) -> None:
+        self.trust_id = f"t-{mock_random.get_random_hex(10)}"
+        self.created_date_time = unix_time()
+        self.last_updated_date_time = self.created_date_time
+        self.state_last_updated_date_time = self.created_date_time
+        self.trust_state = "Creating"
+        self.trust_state_reason = ""
+        self.directory_id = directory_id
+        self.remote_domain_name = remote_domain_name
+        self.trust_password = trust_password
+        self.trust_direction = trust_direction
+        self.trust_type = trust_type
+        self.conditional_forwarder_ip_addrs = conditional_forwarder_ip_addrs
+        self.selective_auth = selective_auth
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "CreatedDateTime": self.created_date_time,
+            "DirectoryId": self.directory_id,
+            "LastUpdatedDateTime": self.last_updated_date_time,
+            "RemoteDomainName": self.remote_domain_name,
+            "SelectiveAuth": self.selective_auth,
+            "StateLastUpdatedDateTime": self.state_last_updated_date_time,
+            "TrustDirection": self.trust_direction,
+            "TrustId": self.trust_id,
+            "TrustState": self.trust_state,
+            "TrustStateReason": self.trust_state_reason,
+            "TrustType": self.trust_type,
+        }
+
+
 class Directory(BaseModel):  # pylint: disable=too-many-instance-attributes
     """Representation of a Simple AD Directory.
 
@@ -82,6 +138,9 @@ class Directory(BaseModel):  # pylint: disable=too-many-instance-attributes
         self.stage = "Active"
         self.launch_time = unix_time()
         self.stage_last_updated_date_time = unix_time()
+        # add type annotation to line below
+        self.ldaps_settings_info: List[LdapsSettingInfo] = []
+        self.trusts: List[Trust] = []
 
         if self.directory_type == "ADConnector":
             self.security_group_id = self.create_security_group(
@@ -159,6 +218,22 @@ class Directory(BaseModel):  # pylint: disable=too-many-instance-attributes
         """Enable/disable sso based on whether new_state is True or False."""
         self.sso_enabled = new_state
 
+    def enable_ldaps(self, enable: bool) -> None:
+        """Enable/disable ldaps based on whether new_state is True or False.
+        This method is only for MicrosoftAD.
+        """
+        if self.directory_type != "MicrosoftAD":
+            raise UnsupportedOperationException(
+                "LDAPS operations are not supported for this Directory Type."
+            )
+        if enable and len(self.ldaps_settings_info) == 0:
+            ldaps_setting = LdapsSettingInfo()
+            ldaps_setting.ldaps_status = "Enabled"
+            self.ldaps_settings_info.append(ldaps_setting)
+        elif not enable and len(self.ldaps_settings_info) > 0:
+            for setting in self.ldaps_settings_info:
+                setting.ldaps_status = "Disabled"
+
     def to_dict(self) -> Dict[str, Any]:
         """Create a dictionary of attributes for Directory."""
         attributes = {
@@ -533,5 +608,89 @@ class DirectoryServiceBackend(BaseBackend):
         self._validate_directory_id(resource_id)
         return self.tagger.list_tags_for_resource(resource_id).get("Tags")  # type: ignore[return-value]
 
+    def create_trust(
+        self,
+        directory_id: str,
+        remote_domain_name: str,
+        trust_password: str,
+        trust_direction: str,
+        trust_type: Optional[str],
+        conditional_forwarder_ip_addrs: Optional[List[str]],
+        selective_auth: Optional[str],
+    ) -> str:
+        self._validate_directory_id(directory_id)
+        validate_args(
+            [
+                ("ssoPassword", trust_password),
+                ("trustDirection", trust_direction),
+                ("remoteDomainName", remote_domain_name),
+            ]
+        )
+        directory = self.directories[directory_id]
+        trust = Trust(
+            directory_id=directory_id,
+            remote_domain_name=remote_domain_name,
+            trust_password=trust_password,
+            trust_direction=trust_direction,
+            trust_type=trust_type,
+            conditional_forwarder_ip_addrs=conditional_forwarder_ip_addrs,
+            selective_auth=selective_auth,
+        )
+        directory.trusts.append(trust)
+        return trust.trust_id
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def describe_trusts(
+        self, directory_id: str, trust_ids: Optional[List[str]]
+    ) -> List[Trust]:
+        self._validate_directory_id(directory_id)
+        directory = self.directories[directory_id]
+        trusts = directory.trusts
+        if trust_ids:
+            trusts = [t for t in directory.trusts if t.trust_id in trust_ids]
+        else:
+            trusts = directory.trusts
+        return trusts
+
+    def delete_trust(
+        self, trust_id: str, delete_associated_conditional_forwarder: Optional[bool]
+    ) -> str:
+        # TODO: Implement handle for delete_associated_conditional_forwarder once conditional forwarder is implemented
+        delete_associated_conditional_forwarder = (
+            delete_associated_conditional_forwarder or False
+        )
+        for directory in self.directories.values():
+            for trust in directory.trusts:
+                if trust.trust_id == trust_id:
+                    directory.trusts.remove(trust)
+                    return trust_id
+        raise EntityDoesNotExistException(f"Trust {trust_id} does not exist")
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def describe_ldaps_settings(
+        self, directory_id: str, type: str
+    ) -> List[LdapsSettingInfo]:
+        """Describe LDAPS settings for a Directory"""
+        self._validate_directory_id(directory_id)
+        directory = self.directories[directory_id]
+        if directory.directory_type != "MicrosoftAD":
+            raise UnsupportedOperationException(
+                "LDAPS operations are not supported for this Directory Type."
+            )
+
+        return directory.ldaps_settings_info
+
+    def enable_ldaps(self, directory_id: str, type: str) -> None:
+        """Enable LDAPS for a Directory"""
+        self._validate_directory_id(directory_id)
+        directory = self.directories[directory_id]
+        directory.enable_ldaps(True)
+
+    def disable_ldaps(self, directory_id: str, type: str) -> None:
+        """Disable LDAPS for a Directory"""
+        self._validate_directory_id(directory_id)
+        directory = self.directories[directory_id]
+        directory.enable_ldaps(False)
+
 
 ds_backends = BackendDict(DirectoryServiceBackend, service_name="ds")
diff --git a/moto/ds/responses.py b/moto/ds/responses.py
index a51cb72d4..2a6068e8e 100644
--- a/moto/ds/responses.py
+++ b/moto/ds/responses.py
@@ -165,3 +165,79 @@ class DirectoryServiceResponse(BaseResponse):
         if next_token:
             response["NextToken"] = next_token
         return json.dumps(response)
+
+    def create_trust(self) -> str:
+        directory_id = self._get_param("DirectoryId")
+        remote_domain_name = self._get_param("RemoteDomainName")
+        trust_password = self._get_param("TrustPassword")
+        trust_direction = self._get_param("TrustDirection")
+        trust_type = self._get_param("TrustType")
+        conditional_forwarder_ip_addrs = self._get_param("ConditionalForwarderIpAddrs")
+        selective_auth = self._get_param("SelectiveAuth")
+        trust_id = self.ds_backend.create_trust(
+            directory_id=directory_id,
+            remote_domain_name=remote_domain_name,
+            trust_password=trust_password,
+            trust_direction=trust_direction,
+            trust_type=trust_type,
+            conditional_forwarder_ip_addrs=conditional_forwarder_ip_addrs,
+            selective_auth=selective_auth,
+        )
+        return json.dumps(dict(TrustId=trust_id))
+
+    def describe_trusts(self) -> str:
+        directory_id = self._get_param("DirectoryId")
+        trust_ids = self._get_param("TrustIds")
+        next_token = self._get_param("NextToken")
+        limit = self._get_param("Limit")
+        trusts, next_token = self.ds_backend.describe_trusts(
+            directory_id=directory_id,
+            trust_ids=trust_ids,
+            next_token=next_token,
+            limit=limit,
+        )
+        trust_list = [trust.to_dict() for trust in trusts]
+        return json.dumps(dict(Trusts=trust_list, nextToken=next_token))
+
+    def delete_trust(self) -> str:
+        trust_id = self._get_param("TrustId")
+        delete_associated_conditional_forwarder = self._get_param(
+            "DeleteAssociatedConditionalForwarder"
+        )
+        trust_id = self.ds_backend.delete_trust(
+            trust_id=trust_id,
+            delete_associated_conditional_forwarder=delete_associated_conditional_forwarder,
+        )
+        return json.dumps(dict(TrustId=trust_id))
+
+    def describe_ldaps_settings(self) -> str:
+        directory_id = self._get_param("DirectoryId")
+        type = self._get_param("Type")
+        next_token = self._get_param("NextToken")
+        limit = self._get_param("Limit")
+        ldaps_settings_info, next_token = self.ds_backend.describe_ldaps_settings(
+            directory_id=directory_id,
+            type=type,
+            next_token=next_token,
+            limit=limit,
+        )
+        ldaps = [ldap.to_dict() for ldap in ldaps_settings_info]
+        return json.dumps(dict(LDAPSSettingsInfo=ldaps, nextToken=next_token))
+
+    def enable_ldaps(self) -> str:
+        directory_id = self._get_param("DirectoryId")
+        type = self._get_param("Type")
+        self.ds_backend.enable_ldaps(
+            directory_id=directory_id,
+            type=type,
+        )
+        return ""
+
+    def disable_ldaps(self) -> str:
+        directory_id = self._get_param("DirectoryId")
+        type = self._get_param("Type")
+        self.ds_backend.disable_ldaps(
+            directory_id=directory_id,
+            type=type,
+        )
+        return ""
diff --git a/moto/ds/utils.py b/moto/ds/utils.py
index f297d393b..a098c14cf 100644
--- a/moto/ds/utils.py
+++ b/moto/ds/utils.py
@@ -13,4 +13,15 @@ PAGINATION_MODEL = {
         "limit_default": 50,
         "unique_attribute": "Key",
     },
+    "describe_ldaps_settings": {
+        "input_token": "next_token",
+        "limit_key": "limit",
+        "limit_default": 100,
+    },
+    "describe_trusts": {
+        "input_token": "next_token",
+        "limit_key": "limit",
+        "limit_default": 100,
+        "unique_attribute": "trust_id",
+    },
 }
diff --git a/moto/ds/validations.py b/moto/ds/validations.py
index 45cdfd322..f263a84dc 100644
--- a/moto/ds/validations.py
+++ b/moto/ds/validations.py
@@ -32,6 +32,8 @@ def validate_args(validators: Any) -> None:
         "connectSettings.customerUserName": validate_user_name,
         "userName": validate_user_name,
         "vpcSettings.subnetIds": validate_subnet_ids,
+        "trustDirection": validate_trust_direction,
+        "remoteDomainName": validate_remote_domain_name,
     }
     err_msgs = []
     # This eventually could be a switch (python 3.10), elminating the need
@@ -147,3 +149,20 @@ def validate_user_name(value: str) -> str:
     if value and not re.match(username_pattern, value):
         return rf"satisfy regular expression pattern: {username_pattern}"
     return ""
+
+
+def validate_trust_direction(value: str) -> str:
+    """Raise exception if trust direction fails to match constraints."""
+    if value not in ["One-Way: Outgoing", "One-Way: Incoming", "Two-Way"]:
+        return "satisfy enum value set: [One-Way: Outgoing, One-Way: Incoming, Two-Way]"
+    return ""
+
+
+def validate_remote_domain_name(value: str) -> str:
+    """Raise exception if remote domain name fails to match constraints."""
+    domain_name_pattern = r"^([a-zA-Z0-9]+[\\.-])+([a-zA-Z0-9])+[.]?$"
+    if not re.match(domain_name_pattern, value):
+        return rf"satisfy regular expression pattern: {domain_name_pattern}"
+    elif len(value) > 1024:
+        return "have length less than or equal to 1024"
+    return ""
diff --git a/moto/polly/resources.py b/moto/polly/resources.py
index 8bfdf2e1a..4abca05f1 100644
--- a/moto/polly/resources.py
+++ b/moto/polly/resources.py
@@ -5,6 +5,54 @@
 #
 # aws polly describe-voices | jq .Voices --indent 4
 VOICE_DATA = [
+    {
+        "Gender": "Female",
+        "Id": "Isabelle",
+        "LanguageCode": "fr-BE",
+        "LanguageName": "Belgian French",
+        "Name": "Isabelle",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Female",
+        "Id": "Danielle",
+        "LanguageCode": "en-US",
+        "LanguageName": "US English",
+        "Name": "Danielle",
+        "SupportedEngines": ["long-form", "neural"],
+    },
+    {
+        "Gender": "Male",
+        "Id": "Gregory",
+        "LanguageCode": "en-US",
+        "LanguageName": "US English",
+        "Name": "Gregory",
+        "SupportedEngines": ["long-form", "neural"],
+    },
+    {
+        "Gender": "Female",
+        "Id": "Burcu",
+        "LanguageCode": "tr-TR",
+        "LanguageName": "Turkish",
+        "Name": "Burcu",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Female",
+        "Id": "Jitka",
+        "LanguageCode": "cs-CZ",
+        "LanguageName": "Czech",
+        "Name": "Jitka",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Female",
+        "Id": "Sabrina",
+        "LanguageCode": "de-CH",
+        "LanguageName": "Swiss Standard German",
+        "Name": "Sabrina",
+        "SupportedEngines": ["neural"],
+    },
     {
         "Gender": "Male",
         "Id": "Kevin",
@@ -141,6 +189,14 @@ VOICE_DATA = [
         "Name": "Ola",
         "SupportedEngines": ["neural"],
     },
+    {
+        "Gender": "Female",
+        "Id": "Lisa",
+        "LanguageCode": "nl-BE",
+        "LanguageName": "Belgian Dutch",
+        "Name": "Lisa",
+        "SupportedEngines": ["neural"],
+    },
     {
         "Gender": "Male",
         "Id": "Ruben",
@@ -189,6 +245,22 @@ VOICE_DATA = [
         "Name": "Seoyeon",
         "SupportedEngines": ["neural", "standard"],
     },
+    {
+        "Gender": "Female",
+        "Id": "Kazuha",
+        "LanguageCode": "ja-JP",
+        "LanguageName": "Japanese",
+        "Name": "Kazuha",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Female",
+        "Id": "Tomoko",
+        "LanguageCode": "ja-JP",
+        "LanguageName": "Japanese",
+        "Name": "Tomoko",
+        "SupportedEngines": ["neural"],
+    },
     {
         "Gender": "Male",
         "Id": "Takumi",
@@ -363,7 +435,7 @@ VOICE_DATA = [
         "LanguageCode": "en-US",
         "LanguageName": "US English",
         "Name": "Matthew",
-        "SupportedEngines": ["neural", "standard"],
+        "SupportedEngines": ["generative", "neural", "standard"],
     },
     {
         "Gender": "Female",
@@ -468,7 +540,7 @@ VOICE_DATA = [
         "LanguageCode": "en-GB",
         "LanguageName": "British English",
         "Name": "Amy",
-        "SupportedEngines": ["neural", "standard"],
+        "SupportedEngines": ["generative", "neural", "standard"],
     },
     {
         "Gender": "Male",
@@ -534,6 +606,14 @@ VOICE_DATA = [
         "Name": "Mads",
         "SupportedEngines": ["standard"],
     },
+    {
+        "Gender": "Female",
+        "Id": "Sofie",
+        "LanguageCode": "da-DK",
+        "LanguageName": "Danish",
+        "Name": "Sofie",
+        "SupportedEngines": ["neural"],
+    },
     {
         "Gender": "Female",
         "Id": "Gwyneth",
@@ -564,6 +644,7 @@ VOICE_DATA = [
         "LanguageCode": "ar-AE",
         "LanguageName": "Gulf Arabic",
         "Name": "Hala",
+        "AdditionalLanguageCodes": ["arb"],
         "SupportedEngines": ["neural"],
     },
     {
@@ -578,10 +659,26 @@ VOICE_DATA = [
         "Gender": "Female",
         "Id": "Hannah",
         "LanguageCode": "de-AT",
-        "LanguageName": "German",
+        "LanguageName": "Austrian German",
         "Name": "Hannah",
         "SupportedEngines": ["neural"],
     },
+    {
+        "Gender": "Female",
+        "Id": "Ruth",
+        "LanguageCode": "en-US",
+        "LanguageName": "US English",
+        "Name": "Ruth",
+        "SupportedEngines": ["generative", "long-form", "neural"],
+    },
+    {
+        "Gender": "Male",
+        "Id": "Stephen",
+        "LanguageCode": "en-US",
+        "LanguageName": "US English",
+        "Name": "Stephen",
+        "SupportedEngines": ["neural"],
+    },
     {
         "Gender": "Female",
         "Id": "Kajal",
@@ -607,6 +704,14 @@ VOICE_DATA = [
         "Name": "Suvi",
         "SupportedEngines": ["neural"],
     },
+    {
+        "Gender": "Female",
+        "Id": "Niamh",
+        "LanguageCode": "en-IE",
+        "LanguageName": "Irish English",
+        "Name": "Niamh",
+        "SupportedEngines": ["neural"],
+    },
     {
         "Gender": "Male",
         "Id": "Arthur",
@@ -639,6 +744,55 @@ VOICE_DATA = [
         "Name": "Pedro",
         "SupportedEngines": ["neural"],
     },
+    {
+        "Gender": "Male",
+        "Id": "Sergio",
+        "LanguageCode": "es-ES",
+        "LanguageName": "Castilian Spanish",
+        "Name": "Sergio",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Male",
+        "Id": "Andres",
+        "LanguageCode": "es-MX",
+        "LanguageName": "Mexican Spanish",
+        "Name": "Andrés",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Male",
+        "Id": "Remi",
+        "LanguageCode": "fr-FR",
+        "LanguageName": "French",
+        "Name": "Rémi",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Male",
+        "Id": "Adriano",
+        "LanguageCode": "it-IT",
+        "LanguageName": "Italian",
+        "Name": "Adriano",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Male",
+        "Id": "Thiago",
+        "LanguageCode": "pt-BR",
+        "LanguageName": "Brazilian Portuguese",
+        "Name": "Thiago",
+        "SupportedEngines": ["neural"],
+    },
+    {
+        "Gender": "Male",
+        "Id": "Zayd",
+        "LanguageCode": "ar-AE",
+        "LanguageName": "Gulf Arabic",
+        "Name": "Zayd",
+        "AdditionalLanguageCodes": ["arb"],
+        "SupportedEngines": ["neural"],
+    },
 ]
 
 LANGUAGE_CODES = {voice["LanguageCode"] for voice in VOICE_DATA}
diff --git a/moto/resourcegroupstaggingapi/models.py b/moto/resourcegroupstaggingapi/models.py
index b9bad8419..8cafdb18b 100644
--- a/moto/resourcegroupstaggingapi/models.py
+++ b/moto/resourcegroupstaggingapi/models.py
@@ -276,115 +276,55 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                         continue
                     yield {"ResourceARN": f"{task.task_arn}", "Tags": tags}
 
-        # EC2 AMI, resource type ec2:image
-        if (
-            not resource_type_filters
-            or "ec2" in resource_type_filters
-            or "ec2:image" in resource_type_filters
-        ):
-            for ami in self.ec2_backend.amis.values():
-                tags = format_tags(self.ec2_backend.tags.get(ami.id, {}))
-
-                if not tags or not tag_filter(tags):
-                    # Skip if no tags, or invalid filter
-                    continue
-                yield {
-                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::image/{ami.id}",
-                    "Tags": tags,
-                }
-
-        # EC2 Instance, resource type ec2:instance
-        if (
-            not resource_type_filters
-            or "ec2" in resource_type_filters
-            or "ec2:instance" in resource_type_filters
-        ):
-            for reservation in self.ec2_backend.reservations.values():
-                for instance in reservation.instances:
-                    tags = format_tags(self.ec2_backend.tags.get(instance.id, {}))
-
-                    if not tags or not tag_filter(tags):
-                        # Skip if no tags, or invalid filter
-                        continue
-                    yield {
-                        "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::instance/{instance.id}",
-                        "Tags": tags,
-                    }
-
-        # EC2 NetworkInterface, resource type ec2:network-interface
-        if (
-            not resource_type_filters
-            or "ec2" in resource_type_filters
-            or "ec2:network-interface" in resource_type_filters
-        ):
-            for eni in self.ec2_backend.enis.values():
-                tags = format_tags(self.ec2_backend.tags.get(eni.id, {}))
-
-                if not tags or not tag_filter(tags):
-                    # Skip if no tags, or invalid filter
-                    continue
-                yield {
-                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::network-interface/{eni.id}",
-                    "Tags": tags,
-                }
-
-        # TODO EC2 ReservedInstance
-
-        # EC2 SecurityGroup, resource type ec2:security-group
-        if (
-            not resource_type_filters
-            or "ec2" in resource_type_filters
-            or "ec2:security-group" in resource_type_filters
-        ):
-            for vpc in self.ec2_backend.groups.values():
-                for sg in vpc.values():
-                    tags = format_tags(self.ec2_backend.tags.get(sg.id, {}))
+        # EC2 Resources
+        ec2_resource_types = {
+            "ec2:image": self.ec2_backend.amis.values(),
+            "ec2:instance": (
+                instance
+                for reservation in self.ec2_backend.reservations.values()
+                for instance in reservation.instances
+            ),
+            "ec2:network-interface": self.ec2_backend.enis.values(),
+            "ec2:security-group": (
+                sg for vpc in self.ec2_backend.groups.values() for sg in vpc.values()
+            ),
+            "ec2:snapshot": self.ec2_backend.snapshots.values(),
+            "ec2:volume": self.ec2_backend.volumes.values(),
+            "ec2:vpc": self.ec2_backend.vpcs.values(),
+            "ec2:subnet": (
+                subnet
+                for subnet in self.ec2_backend.subnets.values()
+                for subnet in subnet.values()
+            ),
+            "ec2:vpc-peering-connection": self.ec2_backend.vpc_pcxs.values(),
+            "ec2:transit-gateway": self.ec2_backend.transit_gateways.values(),
+            "ec2:transit-gateway-attachment": self.ec2_backend.transit_gateway_attachments.values(),
+            "ec2:route-table": self.ec2_backend.route_tables.values(),
+            "ec2:customer-gateway": self.ec2_backend.customer_gateways.values(),
+            "ec2:vpn-connection": self.ec2_backend.vpn_connections.values(),
+            "ec2:natgateway": self.ec2_backend.nat_gateways.values(),
+            "ec2:internet-gateway": self.ec2_backend.internet_gateways.values(),
+            "ec2:managed-prefix-lists": self.ec2_backend.managed_prefix_lists.values(),
+            "ec2:flow-logs": self.ec2_backend.flow_logs.values(),
+            "ec2:spot-instance-request": self.ec2_backend.spot_instance_requests.values(),
+            # TODO: "ec2:reserved-instance": ...,
+        }
 
+        for resource_type, resources in ec2_resource_types.items():
+            if (
+                not resource_type_filters
+                or "ec2" in resource_type_filters
+                or resource_type in resource_type_filters
+            ):
+                for resource in resources:
+                    tags = format_tags(self.ec2_backend.tags.get(resource.id, {}))
                     if not tags or not tag_filter(tags):
-                        # Skip if no tags, or invalid filter
                         continue
                     yield {
-                        "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::security-group/{sg.id}",
+                        "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}:{self.account_id}:{resource_type}/{resource.id}",
                         "Tags": tags,
                     }
 
-        # EC2 Snapshot, resource type ec2:snapshot
-        if (
-            not resource_type_filters
-            or "ec2" in resource_type_filters
-            or "ec2:snapshot" in resource_type_filters
-        ):
-            for snapshot in self.ec2_backend.snapshots.values():
-                tags = format_tags(self.ec2_backend.tags.get(snapshot.id, {}))
-
-                if not tags or not tag_filter(tags):
-                    # Skip if no tags, or invalid filter
-                    continue
-                yield {
-                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::snapshot/{snapshot.id}",
-                    "Tags": tags,
-                }
-
-        # TODO EC2 SpotInstanceRequest
-
-        # EC2 Volume, resource type ec2:volume
-        if (
-            not resource_type_filters
-            or "ec2" in resource_type_filters
-            or "ec2:volume" in resource_type_filters
-        ):
-            for volume in self.ec2_backend.volumes.values():
-                tags = format_tags(self.ec2_backend.tags.get(volume.id, {}))
-
-                if not tags or not tag_filter(
-                    tags
-                ):  # Skip if no tags, or invalid filter
-                    continue
-                yield {
-                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::volume/{volume.id}",
-                    "Tags": tags,
-                }
-
         # EFS, resource type elasticfilesystem:access-point
         if (
             not resource_type_filters
@@ -647,46 +587,6 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     "Tags": tags,
                 }
 
-        # VPC
-        if (
-            not resource_type_filters
-            or "ec2" in resource_type_filters
-            or "ec2:vpc" in resource_type_filters
-        ):
-            for vpc in self.ec2_backend.vpcs.values():
-                tags = format_tags(self.ec2_backend.tags.get(vpc.id, {}))
-                if not tags or not tag_filter(
-                    tags
-                ):  # Skip if no tags, or invalid filter
-                    continue
-                yield {
-                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}:{self.account_id}:vpc/{vpc.id}",
-                    "Tags": tags,
-                }
-        # VPC Subnet
-        if (
-            not resource_type_filters
-            or "ec2" in resource_type_filters
-            or "ec2:subnet" in resource_type_filters
-        ):
-            for subnet in self.ec2_backend.subnets.values():
-                for subnet_id in subnet.keys():
-                    tags = format_tags(self.ec2_backend.tags.get(subnet_id, {}))
-                    if not tags or not tag_filter(tags):
-                        continue
-                    yield {
-                        "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}:{self.account_id}:subnet/{subnet_id}",
-                        "Tags": tags,
-                    }
-
-        # VPC Customer Gateway
-        # VPC DHCP Option Set
-        # VPC Internet Gateway
-        # VPC Network ACL
-        # VPC Route Table
-        # VPC Virtual Private Gateway
-        # VPC VPN Connection
-
         # Lambda Instance
         if (
             not resource_type_filters
diff --git a/moto/sns/models.py b/moto/sns/models.py
index 74321285d..e5855df0a 100644
--- a/moto/sns/models.py
+++ b/moto/sns/models.py
@@ -85,6 +85,7 @@ class Topic(CloudFormationModel):
         message_attributes: Optional[Dict[str, Any]] = None,
         group_id: Optional[str] = None,
         deduplication_id: Optional[str] = None,
+        message_structure: Optional[str] = None,
     ) -> str:
         message_id = str(mock_random.uuid4())
         subscriptions, _ = self.sns_backend.list_subscriptions_by_topic(
@@ -98,7 +99,9 @@ class Topic(CloudFormationModel):
                 message_attributes=message_attributes,
                 group_id=group_id,
                 deduplication_id=deduplication_id,
+                message_structure=message_structure,
             )
+
         self.sent_notifications.append(
             (message_id, message, subject, message_attributes, group_id)
         )
@@ -222,14 +225,19 @@ class Subscription(BaseModel):
         message_attributes: Optional[Dict[str, Any]] = None,
         group_id: Optional[str] = None,
         deduplication_id: Optional[str] = None,
+        message_structure: Optional[str] = None,
     ) -> None:
         if self._filter_policy_matcher is not None:
             if not self._filter_policy_matcher.matches(message_attributes, message):
                 return
 
+        if message_structure == "json":
+            message = self._parse_message_structure(message, self.protocol)
+
         if self.protocol == "sqs":
             queue_name = self.endpoint.split(":")[-1]
             region = self.endpoint.split(":")[3]
+
             if self.attributes.get("RawMessageDelivery") != "true":
                 sqs_backends[self.account_id][region].send_message(
                     queue_name,
@@ -379,6 +387,23 @@ Notification
     def private_key(self) -> rsa.RSAPrivateKey:
         return rsa.generate_private_key(public_exponent=65537, key_size=2028)
 
+    def _parse_message_structure(self, message: str, protocol: str) -> str:
+        try:
+            structured_message = json.loads(message)
+            message = (
+                structured_message.get(self.protocol) or structured_message["default"]
+            )
+
+            return message
+        except json.JSONDecodeError:
+            raise InvalidParameterValue(
+                "Message is not valid JSON.",
+            )
+        except KeyError:
+            raise InvalidParameterValue(
+                f"Message does not contain {protocol} or default keys.",
+            )
+
 
 class PlatformApplication(BaseModel):
     def __init__(
@@ -657,6 +682,7 @@ class SNSBackend(BaseBackend):
         message_attributes: Optional[Dict[str, Any]] = None,
         group_id: Optional[str] = None,
         deduplication_id: Optional[str] = None,
+        message_structure: Optional[str] = None,
     ) -> str:
         if subject is not None and len(subject) > 100:
             # Note that the AWS docs around length are wrong: https://github.com/getmoto/moto/issues/1503
@@ -676,6 +702,9 @@ class SNSBackend(BaseBackend):
                 "An error occurred (InvalidParameter) when calling the Publish operation: Invalid parameter: Message too long"
             )
 
+        if message_structure is not None and message_structure != "json":
+            raise InvalidParameterValue("MessageStructure must be 'json' if provided")
+
         try:
             topic = self.get_topic(arn)  # type: ignore
 
@@ -703,6 +732,7 @@ class SNSBackend(BaseBackend):
                 message_attributes=message_attributes,
                 group_id=group_id,
                 deduplication_id=deduplication_id,
+                message_structure=message_structure,
             )
         except SNSNotFoundError:
             endpoint = self.get_endpoint(arn)  # type: ignore
@@ -1107,7 +1137,7 @@ class SNSBackend(BaseBackend):
         self, topic_arn: str, publish_batch_request_entries: List[Dict[str, Any]]
     ) -> Tuple[List[Dict[str, str]], List[Dict[str, Any]]]:
         """
-        The MessageStructure and MessageDeduplicationId-parameters have not yet been implemented.
+        The MessageDeduplicationId-parameters have not yet been implemented.
         """
         topic = self.get_topic(topic_arn)
 
@@ -1139,6 +1169,7 @@ class SNSBackend(BaseBackend):
                     message_attributes=entry.get("MessageAttributes", {}),
                     group_id=entry.get("MessageGroupId"),
                     deduplication_id=entry.get("MessageDeduplicationId"),
+                    message_structure=entry.get("MessageStructure"),
                 )
                 successful.append({"MessageId": message_id, "Id": entry["Id"]})
             except Exception as e:
diff --git a/moto/sns/responses.py b/moto/sns/responses.py
index a350e1c09..847b76db9 100644
--- a/moto/sns/responses.py
+++ b/moto/sns/responses.py
@@ -346,6 +346,7 @@ class SNSResponse(BaseResponse):
         subject = self._get_param("Subject")
         message_group_id = self._get_param("MessageGroupId")
         message_deduplication_id = self._get_param("MessageDeduplicationId")
+        message_structure = self._get_param("MessageStructure")
 
         message_attributes = self._parse_message_attributes()
 
@@ -375,6 +376,7 @@ class SNSResponse(BaseResponse):
                 message_attributes=message_attributes,
                 group_id=message_group_id,
                 deduplication_id=message_deduplication_id,
+                message_structure=message_structure,
             )
         except ValueError as err:
             error_response = self._error("InvalidParameter", str(err))


Test patch:
diff --git a/tests/test_batch/test_utils.py b/tests/test_batch/test_utils.py
index 870b2bad0..2d080f9aa 100644
--- a/tests/test_batch/test_utils.py
+++ b/tests/test_batch/test_utils.py
@@ -1,6 +1,9 @@
+from unittest.mock import patch
+
 import pytest
 
 from moto.batch.exceptions import ValidationError
+from moto.batch.models import Job
 from moto.batch.utils import JobStatus
 
 
@@ -67,3 +70,14 @@ def test_JobStatus_status_transitions():
         else:
             assert before_status == JobStatus.STARTING
             assert after_status == JobStatus.RUNNING
+
+
+@patch.object(Job, "__init__", lambda self, *args, **kwargs: None)
+def test_add_parameters_to_command():
+    _parameters = {"dossier_md5": "abc", "study_id": "T123BC00001"}
+    _command = ["--dossier_md5", "Ref::dossier_md5", "--study_id", "Ref::study_id"]
+
+    job = Job()  # noqa
+    job.parameters = _parameters
+    output = job._add_parameters_to_command(command=_command)
+    assert output == ["--dossier_md5", "abc", "--study_id", "T123BC00001"]
diff --git a/tests/test_ds/test_ds.py b/tests/test_ds/test_ds.py
index d6f6988ac..88b0a7809 100644
--- a/tests/test_ds/test_ds.py
+++ b/tests/test_ds/test_ds.py
@@ -325,3 +325,133 @@ def test_ds_disable_sso():
     result = client.describe_directories()
     directory = result["DirectoryDescriptions"][0]
     assert not directory["SsoEnabled"]
+
+
+@mock_aws
+def test_create_trust():
+    client = boto3.client("ds", region_name=TEST_REGION)
+    ec2_client = boto3.client("ec2", region_name=TEST_REGION)
+    directory_id = create_test_directory(client, ec2_client)
+
+    # Create a trust
+    trust_id = client.create_trust(
+        DirectoryId=directory_id,
+        RemoteDomainName="example.com",
+        TrustPassword="password",
+        TrustDirection="One-Way: Outgoing",
+        TrustType="External",
+    )["TrustId"]
+
+    assert trust_id is not None
+
+    # Describe the trust
+    trusts = client.describe_trusts(DirectoryId=directory_id)["Trusts"]
+    assert len(trusts) == 1
+    assert trusts[0]["TrustId"] == trust_id
+    assert trusts[0]["DirectoryId"] == directory_id
+
+
+@mock_aws
+def test_describe_trusts():
+    client = boto3.client("ds", region_name=TEST_REGION)
+    ec2_client = boto3.client("ec2", region_name=TEST_REGION)
+    directory_id = create_test_directory(client, ec2_client)
+
+    # Create several trusts
+    trust_ids = []
+    for x in range(5):
+        trust_id = client.create_trust(
+            DirectoryId=directory_id,
+            RemoteDomainName=f"example{x}.com",
+            TrustPassword="P@ssword1234!",
+            TrustDirection="One-Way: Outgoing",
+            TrustType="External",
+        )["TrustId"]
+        trust_ids.append(trust_id)
+
+    # Describe the trusts
+    trusts = client.describe_trusts(DirectoryId=directory_id)["Trusts"]
+    assert len(trusts) == 5
+    for idx, trust in enumerate(trusts):
+        assert trust["TrustId"] == trust_ids[idx]
+        assert trust["DirectoryId"] == directory_id
+
+    # Describe a single trust
+    trust = client.describe_trusts(DirectoryId=directory_id, TrustIds=[trust_ids[2]])[
+        "Trusts"
+    ][0]
+    assert trust["TrustId"] == trust_ids[2]
+
+    # Describe multiple trusts
+    trusts = client.describe_trusts(DirectoryId=directory_id, TrustIds=trust_ids[1:3])[
+        "Trusts"
+    ]
+    assert len(trusts) == 2
+    assert trusts[0]["TrustId"] == trust_ids[1]
+    assert trusts[1]["TrustId"] == trust_ids[2]
+
+
+@mock_aws
+def test_delete_trust():
+    client = boto3.client("ds", region_name=TEST_REGION)
+    ec2_client = boto3.client("ec2", region_name=TEST_REGION)
+    directory_id = create_test_directory(client, ec2_client)
+
+    # Create several trusts
+    trust_ids = []
+    for x in range(2):
+        trust_id = client.create_trust(
+            DirectoryId=directory_id,
+            RemoteDomainName=f"example{x}.com",
+            TrustPassword="P@ssword1234!",
+            TrustDirection="One-Way: Outgoing",
+            TrustType="External",
+        )["TrustId"]
+        trust_ids.append(trust_id)
+
+    # Verify the expected trusts exist
+    trusts = client.describe_trusts(DirectoryId=directory_id)["Trusts"]
+    assert len(trusts) == 2
+
+    # Delete a trust
+    resp = client.delete_trust(TrustId=trust_ids[0])
+    assert resp["TrustId"] == trust_ids[0]
+
+    # Verify the trust was deleted
+    trusts = client.describe_trusts(DirectoryId=directory_id)["Trusts"]
+    assert len(trusts) == 1
+    assert trusts[0]["TrustId"] != trust_ids[0]
+
+    # Test deleting a trust that doesn't exist
+    with pytest.raises(ClientError) as exc:
+        client.delete_trust(TrustId="t-1234567890")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "EntityDoesNotExistException"
+
+
+@mock_aws
+def test_ldaps_exceptions_non_microsoftad():
+    """Test LDAPS operations on non-Microsoft AD directories."""
+
+    client = boto3.client("ds", region_name=TEST_REGION)
+    ec2_client = boto3.client("ec2", region_name=TEST_REGION)
+
+    directory_id = create_test_directory(client, ec2_client)
+
+    # Test enabling LDAPS on a non-Microsoft AD directory.
+    with pytest.raises(ClientError) as exc:
+        client.enable_ldaps(DirectoryId=directory_id, Type="Client")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "UnsupportedOperationException"
+
+    # Test describe_ldaps_settings on a non-Microsoft AD directory.
+    with pytest.raises(ClientError) as exc:
+        client.describe_ldaps_settings(DirectoryId=directory_id, Type="Client")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "UnsupportedOperationException"
+
+    # Test disable_ldaps on a non-Microsoft AD directory.
+    with pytest.raises(ClientError) as exc:
+        client.disable_ldaps(DirectoryId=directory_id, Type="Client")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "UnsupportedOperationException"
diff --git a/tests/test_ds/test_ds_microsoft_ad.py b/tests/test_ds/test_ds_microsoft_ad.py
index 630b1e7ea..2d8c7d563 100644
--- a/tests/test_ds/test_ds_microsoft_ad.py
+++ b/tests/test_ds/test_ds_microsoft_ad.py
@@ -227,3 +227,34 @@ def test_ds_get_microsoft_ad_directory_limits():
     assert limits["CloudOnlyMicrosoftADLimitReached"]
     assert not limits["ConnectedDirectoriesLimitReached"]
     assert not limits["CloudOnlyDirectoriesCurrentCount"]
+
+
+@mock_aws
+def test_enable_describe_disable_ldaps():
+    """Test good and bad invocations of describe_directories()."""
+    client = boto3.client("ds", region_name=TEST_REGION)
+    ec2_client = boto3.client("ec2", region_name=TEST_REGION)
+
+    directory_id = create_test_microsoft_ad(client, ec2_client)
+
+    # Describe LDAPS settings for Microsoft AD without LDAPS enabled
+    ldaps = client.describe_ldaps_settings(DirectoryId=directory_id)[
+        "LDAPSSettingsInfo"
+    ]
+    assert ldaps == []
+
+    # Enable LDAPS for Microsoft AD and verify it is enabled
+    client.enable_ldaps(DirectoryId=directory_id, Type="Client")
+    ldaps = client.describe_ldaps_settings(DirectoryId=directory_id)[
+        "LDAPSSettingsInfo"
+    ]
+    assert len(ldaps) == 1
+    assert ldaps[0]["LDAPSStatus"] == "Enabled"
+
+    # Disable LDAPS for Microsoft AD and verify it is disabled
+    client.disable_ldaps(DirectoryId=directory_id, Type="Client")
+    ldaps = client.describe_ldaps_settings(DirectoryId=directory_id)[
+        "LDAPSSettingsInfo"
+    ]
+    assert len(ldaps) == 1
+    assert ldaps[0]["LDAPSStatus"] == "Disabled"
diff --git a/tests/test_sns/test_publish_batch.py b/tests/test_sns/test_publish_batch.py
index a3da85f5e..9d9aa476a 100644
--- a/tests/test_sns/test_publish_batch.py
+++ b/tests/test_sns/test_publish_batch.py
@@ -180,3 +180,64 @@ def test_publish_batch_to_sqs_raw():
 
     assert ("foo", None) in messages
     assert ("bar", {"a": {"StringValue": "v", "DataType": "String"}}) in messages
+
+
+@mock_aws
+def test_publish_with_with_message_structure_json():
+    sns = boto3.client("sns", region_name="us-east-1")
+
+    topic = sns.create_topic(Name="some-topic")
+
+    sqs = boto3.resource("sqs", region_name="us-east-1")
+    queue = sqs.create_queue(QueueName="test-queue")
+
+    sns.subscribe(
+        TopicArn=topic["TopicArn"],
+        Protocol="sqs",
+        Endpoint=queue.attributes["QueueArn"],
+    )
+
+    entries = [
+        {
+            "Id": "no-message-structure",
+            "Message": "no-message-structure",
+        },
+        {
+            "Id": "json-message-structure",
+            "Message": json.dumps(
+                {
+                    "default": "default-message",
+                    "sqs": "queue-message",
+                }
+            ),
+            "MessageStructure": "json",
+        },
+        {
+            "Id": "json-message-structure-fallback-to-default",
+            "Message": json.dumps(
+                {
+                    "default": "default-message",
+                    "email": "email-message",
+                }
+            ),
+            "MessageStructure": "json",
+        },
+    ]
+
+    sns.publish_batch(
+        TopicArn=topic["TopicArn"],
+        PublishBatchRequestEntries=entries,
+    )
+
+    queue_messages = queue.receive_messages(MaxNumberOfMessages=10)
+
+    assert len(queue_messages) == 3
+
+    first_queue_message = json.loads(queue_messages[0].body)
+    assert first_queue_message["Message"] == "no-message-structure"
+
+    second_queue_message = json.loads(queue_messages[1].body)
+    assert second_queue_message["Message"] == "queue-message"
+
+    third_queue_message = json.loads(queue_messages[2].body)
+    assert third_queue_message["Message"] == "default-message"
diff --git a/tests/test_sns/test_publishing_boto3.py b/tests/test_sns/test_publishing_boto3.py
index f29bd1abd..f1d61071f 100644
--- a/tests/test_sns/test_publishing_boto3.py
+++ b/tests/test_sns/test_publishing_boto3.py
@@ -2500,3 +2500,114 @@ def test_filtering_message_body_nested_multiple_records_match():
     assert message_attributes == [{"match": {"Type": "String", "Value": "body"}}]
     message_bodies = [json.loads(json.loads(m.body)["Message"]) for m in messages]
     assert message_bodies == [payload]
+
+
+@mock_aws
+def test_publish_with_message_structure_json():
+    sns = boto3.resource("sns", region_name="us-east-1")
+
+    topic = sns.create_topic(Name="some-topic")
+
+    sqs = boto3.resource("sqs", region_name="us-east-1")
+    queue = sqs.create_queue(QueueName="test-queue")
+
+    topic.subscribe(
+        TopicArn=topic.arn,
+        Protocol="sqs",
+        Endpoint=queue.attributes["QueueArn"],
+    )
+
+    topic.publish(
+        Message=json.dumps(
+            {
+                "default": "message",
+                "sqs": "queue-message",
+            }
+        ),
+        MessageStructure="json",
+    )
+
+    queue_message = json.loads(queue.receive_messages()[0].body)
+    assert queue_message["Message"] == "queue-message"
+
+
+@mock_aws
+def test_publish_with_message_structure_json_fallback_to_default():
+    sns = boto3.resource("sns", region_name="us-east-1")
+
+    topic = sns.create_topic(Name="some-topic")
+
+    sqs = boto3.resource("sqs", region_name="us-east-1")
+    queue = sqs.create_queue(QueueName="test-queue")
+
+    topic.subscribe(
+        TopicArn=topic.arn,
+        Protocol="sqs",
+        Endpoint=queue.attributes["QueueArn"],
+    )
+
+    topic.publish(
+        Message=json.dumps(
+            {
+                "default": "message",
+                "email": "email-message",
+            }
+        ),
+        MessageStructure="json",
+    )
+
+    queue_message = json.loads(queue.receive_messages()[0].body)
+    assert queue_message["Message"] == "message"
+
+
+@mock_aws
+def test_publish_with_message_structure_errors():
+    sns = boto3.resource("sns", region_name="us-east-1")
+
+    topic = sns.create_topic(Name="some-topic")
+
+    sqs = boto3.resource("sqs", region_name="us-east-1")
+    queue = sqs.create_queue(QueueName="test-queue")
+
+    topic.subscribe(
+        TopicArn=topic.arn,
+        Protocol="sqs",
+        Endpoint=queue.attributes["QueueArn"],
+    )
+
+    try:
+        topic.publish(
+            Message="invalid-json",
+            MessageStructure="json",
+        )
+    except ClientError as err:
+        assert err.response["Error"]["Code"] == "InvalidParameterValue"
+        assert err.response["Error"]["Message"] == "Message is not valid JSON."
+
+    try:
+        topic.publish(
+            Message=json.dumps(
+                {
+                    "email": "email-message",
+                }
+            ),
+            MessageStructure="json",
+        )
+    except ClientError as err:
+        assert err.response["Error"]["Code"] == "InvalidParameterValue"
+        assert (
+            err.response["Error"]["Message"]
+            == "Message does not contain sqs or default keys."
+        )
+
+    try:
+        topic.publish(
+            Message="no-json-structure",
+            MessageStructure="no-json",
+        )
+    except ClientError as err:
+        assert err.response["Error"]["Code"] == "InvalidParameterValue"
+        assert (
+            err.response["Error"]["Message"]
+            == "MessageStructure must be 'json' if provided"
+        )