Repository: getmoto/moto
Pull request #8165 (2024-09-27 06:46:18)

CloudFront: add get_invalidation support

ligangty: Seems the server mode test give the error that only contains "404" for not found exception but no specified message. Just give a fix for the tests.

Patch:
diff --git a/moto/cloudfront/exceptions.py b/moto/cloudfront/exceptions.py
index 6e9104eba..28d374341 100644
--- a/moto/cloudfront/exceptions.py
+++ b/moto/cloudfront/exceptions.py
@@ -74,3 +74,12 @@ class NoSuchOriginAccessControl(CloudFrontException):
             "NoSuchOriginAccessControl",
             message="The specified origin access control does not exist.",
         )
+
+
+class NoSuchInvalidation(CloudFrontException):
+    code = 404
+
+    def __init__(self) -> None:
+        super().__init__(
+            "NoSuchInvalidation", message="The specified invalidation does not exist."
+        )
diff --git a/moto/cloudfront/models.py b/moto/cloudfront/models.py
index c720c3d8f..b00e7b688 100644
--- a/moto/cloudfront/models.py
+++ b/moto/cloudfront/models.py
@@ -14,6 +14,7 @@ from .exceptions import (
     DomainNameNotAnS3Bucket,
     InvalidIfMatchVersion,
     NoSuchDistribution,
+    NoSuchInvalidation,
     NoSuchOriginAccessControl,
     OriginDoesNotExist,
 )
@@ -393,6 +394,19 @@ class CloudFrontBackend(BaseBackend):
         """
         return self.invalidations.get(dist_id) or []
 
+    def get_invalidation(self, dist_id: str, id: str) -> Invalidation:
+        if dist_id not in self.distributions:
+            raise NoSuchDistribution
+        try:
+            invalidations = self.invalidations[dist_id]
+            if invalidations:
+                for invalidation in invalidations:
+                    if invalidation.invalidation_id == id:
+                        return invalidation
+        except KeyError:
+            pass
+        raise NoSuchInvalidation
+
     def list_tags_for_resource(self, resource: str) -> Dict[str, List[Dict[str, str]]]:
         return self.tagger.list_tags_for_resource(resource)
 
diff --git a/moto/cloudfront/responses.py b/moto/cloudfront/responses.py
index 609bd76ba..8ba8f5235 100644
--- a/moto/cloudfront/responses.py
+++ b/moto/cloudfront/responses.py
@@ -103,6 +103,15 @@ class CloudFrontResponse(BaseResponse):
 
         return 200, {}, response
 
+    def get_invalidation(self) -> TYPE_RESPONSE:
+        pathItems = self.path.split("/")
+        dist_id = pathItems[-3]
+        id = pathItems[-1]
+        invalidation = self.backend.get_invalidation(dist_id, id)
+        template = self.response_template(GET_INVALIDATION_TEMPLATE)
+        response = template.render(invalidation=invalidation, xmlns=XMLNS)
+        return 200, {}, response
+
     def list_tags_for_resource(self) -> TYPE_RESPONSE:
         resource = unquote(self._get_param("Resource"))
         tags = self.backend.list_tags_for_resource(resource=resource)["Tags"]
@@ -634,6 +643,8 @@ CREATE_INVALIDATION_TEMPLATE = """<?xml version="1.0"?>
 </Invalidation>
 """
 
+GET_INVALIDATION_TEMPLATE = CREATE_INVALIDATION_TEMPLATE
+
 INVALIDATIONS_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?>
 <InvalidationList>
    <IsTruncated>false</IsTruncated>
diff --git a/moto/cloudfront/urls.py b/moto/cloudfront/urls.py
index a630a1f28..5b47398e4 100644
--- a/moto/cloudfront/urls.py
+++ b/moto/cloudfront/urls.py
@@ -11,6 +11,7 @@ url_paths = {
     "{0}/2020-05-31/distribution/(?P<distribution_id>[^/]+)$": CloudFrontResponse.dispatch,
     "{0}/2020-05-31/distribution/(?P<distribution_id>[^/]+)/config$": CloudFrontResponse.dispatch,
     "{0}/2020-05-31/distribution/(?P<distribution_id>[^/]+)/invalidation": CloudFrontResponse.dispatch,
+    "{0}/2020-05-31/distribution/(?P<distribution_id>[^/]+)/invalidation/(?P<invalidation_id>[^/]+)": CloudFrontResponse.dispatch,
     "{0}/2020-05-31/tagging$": CloudFrontResponse.dispatch,
     "{0}/2020-05-31/origin-access-control$": CloudFrontResponse.dispatch,
     "{0}/2020-05-31/origin-access-control/(?P<oac_id>[^/]+)$": CloudFrontResponse.dispatch,
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/ecs/models.py b/moto/ecs/models.py
index c256d940f..302fa2efd 100644
--- a/moto/ecs/models.py
+++ b/moto/ecs/models.py
@@ -299,10 +299,14 @@ class TaskDefinition(BaseObject, CloudFormationModel):
             properties.get("ContainerDefinitions", []), pascal_to_camelcase
         )
         volumes = remap_nested_keys(properties.get("Volumes", []), pascal_to_camelcase)
+        memory = properties.get("Memory")
 
         ecs_backend = ecs_backends[account_id][region_name]
         return ecs_backend.register_task_definition(
-            family=family, container_definitions=container_definitions, volumes=volumes
+            family=family,
+            container_definitions=container_definitions,
+            volumes=volumes,
+            memory=memory,
         )
 
     @classmethod
@@ -320,6 +324,7 @@ class TaskDefinition(BaseObject, CloudFormationModel):
         )
         container_definitions = properties["ContainerDefinitions"]
         volumes = properties.get("Volumes")
+        memory = properties.get("Memory")
         if (
             original_resource.family != family
             or original_resource.container_definitions != container_definitions
@@ -333,6 +338,7 @@ class TaskDefinition(BaseObject, CloudFormationModel):
                 family=family,
                 container_definitions=container_definitions,
                 volumes=volumes,
+                memory=memory,
             )
         else:
             # no-op when nothing changed between old and new resources


Test patch:
diff --git a/tests/test_cloudfront/test_cloudfront_invalidation.py b/tests/test_cloudfront/test_cloudfront_invalidation.py
index 655dbcb37..cf98046c8 100644
--- a/tests/test_cloudfront/test_cloudfront_invalidation.py
+++ b/tests/test_cloudfront/test_cloudfront_invalidation.py
@@ -1,4 +1,6 @@
 import boto3
+import pytest
+from botocore.errorfactory import ClientError
 
 from moto import mock_aws
 
@@ -97,3 +99,52 @@ def test_list_invalidations__no_entries():
     assert resp["InvalidationList"]["IsTruncated"] is False
     assert resp["InvalidationList"]["Quantity"] == 0
     assert "Items" not in resp["InvalidationList"]
+
+
+@mock_aws
+def test_get_invalidation():
+    client = boto3.client("cloudfront", region_name="us-west-1")
+    config = scaffold.example_distribution_config("ref")
+    resp = client.create_distribution(DistributionConfig=config)
+    dist_id = resp["Distribution"]["Id"]
+
+    createResp = client.create_invalidation(
+        DistributionId=dist_id,
+        InvalidationBatch={
+            "Paths": {"Quantity": 1, "Items": ["/path1"]},
+            "CallerReference": "ref2",
+        },
+    )
+    existed = createResp["Invalidation"]
+    resp = client.get_invalidation(DistributionId=dist_id, Id=existed["Id"])
+    assert "Invalidation" in resp
+    returned = resp["Invalidation"]
+    assert returned["Id"] == existed["Id"]
+    assert returned["Status"] == existed["Status"]
+    assert returned["CreateTime"] == existed["CreateTime"]
+    assert returned["InvalidationBatch"] == existed["InvalidationBatch"]
+
+
+@mock_aws
+def test_get_invalidation_dist_not_found():
+    client = boto3.client("cloudfront", region_name="us-west-1")
+
+    with pytest.raises(ClientError) as exc:
+        client.get_invalidation(DistributionId="notfound", Id="notfound")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "NoSuchDistribution"
+    assert err["Message"] == "The specified distribution does not exist."
+
+
+@mock_aws
+def test_get_invalidation_id_not_found():
+    client = boto3.client("cloudfront", region_name="us-west-1")
+    config = scaffold.example_distribution_config("ref")
+    resp = client.create_distribution(DistributionConfig=config)
+    dist_id = resp["Distribution"]["Id"]
+
+    with pytest.raises(ClientError) as exc:
+        client.get_invalidation(DistributionId=dist_id, Id="notfound")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "NoSuchInvalidation"
+    assert err["Message"] == "The specified invalidation does not exist."
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_ecs/test_ecs_cloudformation.py b/tests/test_ecs/test_ecs_cloudformation.py
index 292bb3d60..6617b9377 100644
--- a/tests/test_ecs/test_ecs_cloudformation.py
+++ b/tests/test_ecs/test_ecs_cloudformation.py
@@ -2,6 +2,7 @@ import json
 from copy import deepcopy
 
 import boto3
+import pytest
 
 from moto import mock_aws
 from moto.core.utils import pascal_to_camelcase, remap_nested_keys
@@ -366,3 +367,79 @@ def test_create_task_definition_through_cloudformation():
     assert task_definition["volumes"] == expected_properties["volumes"]
     for key, value in expected_properties["containerDefinitions"][0].items():
         assert task_definition["containerDefinitions"][0][key] == value
+
+
+@mock_aws
+def test_cloudformation_container_definition_validation():
+    cfn_conn = boto3.client("cloudformation", region_name="us-west-1")
+
+    # No task memory, no container memory
+    invalid_definition = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Description": "ECS Cluster Test CloudFormation",
+        "Resources": {
+            "testTaskDefinition": {
+                "Type": "AWS::ECS::TaskDefinition",
+                "Properties": {
+                    "ContainerDefinitions": [
+                        {
+                            "Name": "ecs-sample",
+                            "Image": "amazon/amazon-ecs-sample",
+                        }
+                    ],
+                },
+            }
+        },
+    }
+    invalid_json = json.dumps(invalid_definition)
+    with pytest.raises(Exception):
+        cfn_conn.create_stack(StackName="invalid-stack", TemplateBody=invalid_json)
+
+    # No task memory, has container memory
+    valid_definition = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Description": "ECS Cluster Test CloudFormation",
+        "Resources": {
+            "testTaskDefinition": {
+                "Type": "AWS::ECS::TaskDefinition",
+                "Properties": {
+                    "ContainerDefinitions": [
+                        {
+                            "Name": "ecs-sample",
+                            "Image": "amazon/amazon-ecs-sample",
+                            "Memory": 1024,
+                        }
+                    ],
+                },
+            }
+        },
+    }
+    valid_json = json.dumps(valid_definition)
+    name = "valid-stack-container-memory"
+    cfn_conn.create_stack(StackName=name, TemplateBody=valid_json)
+    cfn_conn.update_stack(StackName=name, TemplateBody=valid_json)
+
+    # Has task memory, no container memory
+    valid_definition = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Description": "ECS Cluster Test CloudFormation",
+        "Resources": {
+            "testTaskDefinition": {
+                "Type": "AWS::ECS::TaskDefinition",
+                "Properties": {
+                    "Memory": 1024,
+                    "ContainerDefinitions": [
+                        {
+                            "Name": "ecs-sample",
+                            "Image": "amazon/amazon-ecs-sample",
+                        }
+                    ],
+                },
+            }
+        },
+    }
+
+    valid_json = json.dumps(valid_definition)
+    name = "valid-stack-task-memory"
+    cfn_conn.create_stack(StackName=name, TemplateBody=valid_json)
+    cfn_conn.update_stack(StackName=name, TemplateBody=valid_json)