Repository: getmoto/moto
Pull request #8158 (2024-09-26 04:44:38)

SNS: Support MessageStructure=json for publish and publish_batch
Hi! I implemented the parameter `MessageStructure` for SNS's publish and publish_batch methods.

I added unit tests. I tested those changes in the integration tests of my project, which work similarly to the AWS implementation (in my case).

I hope, my changes will be useful to others.

I will be happy to adapt my PR to Moto project standards if needed.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8158?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 94.50%. Comparing base [(`6c68326`)](https://app.codecov.io/gh/getmoto/moto/commit/6c68326479024619214119c6d8835153e5f7804a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`9e76e81`)](https://app.codecov.io/gh/getmoto/moto/commit/9e76e813598ed73e8d9a1f4f3344cdc3d9db8d63?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).
> Report is 3 commits behind head on master.

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master    #8158   +/-   ##
=======================================
  Coverage   94.50%   94.50%           
=======================================
  Files        1150     1150           
  Lines       99203    99217   +14     
=======================================
+ Hits        93752    93766   +14     
  Misses       5451     5451           
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8158/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/8158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `28.79% <7.14%> (-0.01%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.47% <100.00%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/8158?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).


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/docs/docs/services/wafv2.rst b/docs/docs/services/wafv2.rst
index d1ab9b4e1..f2f7f95f4 100644
--- a/docs/docs/services/wafv2.rst
+++ b/docs/docs/services/wafv2.rst
@@ -17,10 +17,6 @@ wafv2
 |start-h3| Implemented features for this service |end-h3|
 
 - [X] associate_web_acl
-  
-        Only APIGateway Stages can be associated at the moment.
-        
-
 - [ ] check_capacity
 - [ ] create_api_key
 - [X] create_ip_set
diff --git a/moto/ec2/models/elastic_block_store.py b/moto/ec2/models/elastic_block_store.py
index cee0558d3..e0c35e18e 100644
--- a/moto/ec2/models/elastic_block_store.py
+++ b/moto/ec2/models/elastic_block_store.py
@@ -269,6 +269,7 @@ class EBSBackend:
         self.volumes: Dict[str, Volume] = {}
         self.attachments: Dict[str, VolumeAttachment] = {}
         self.snapshots: Dict[str, Snapshot] = {}
+        self.default_kms_key_id: str = ""
 
     def create_volume(
         self,
@@ -284,7 +285,9 @@ class EBSBackend:
         if kms_key_id and not encrypted:
             raise InvalidParameterDependency("KmsKeyId", "Encrypted")
         if encrypted and not kms_key_id:
-            kms_key_id = self._get_default_encryption_key()
+            if not self.default_kms_key_id:
+                self._create_default_encryption_key()
+            kms_key_id = self.default_kms_key_id
         if volume_type in IOPS_REQUIRED_VOLUME_TYPES and not iops:
             raise InvalidParameterDependency("VolumeType", "Iops")
         elif volume_type == "gp3" and not iops:
@@ -546,7 +549,7 @@ class EBSBackend:
         else:
             snapshot.create_volume_permission_groups.difference_update(groups)  # type: ignore[arg-type]
 
-    def _get_default_encryption_key(self) -> str:
+    def _create_default_encryption_key(self) -> str:
         # https://aws.amazon.com/kms/features/#AWS_Service_Integration
         # An AWS managed CMK is created automatically when you first create
         # an encrypted resource using an AWS service integrated with KMS.
@@ -564,4 +567,18 @@ class EBSBackend:
             )
             kms.create_alias(key.id, ebs_alias)
         ebs_key = kms.describe_key(ebs_alias)
+        self.default_kms_key_id = ebs_key.arn
         return ebs_key.arn
+
+    def modify_ebs_default_kms_key_id(self, kms_key_id: str) -> str:
+        # If this parameter is not specified, the standard KMS key for Amazon EBS is used.
+        if not kms_key_id:
+            return self._create_default_encryption_key()
+
+        # Determine if the value is a key ID, alias, key ARN, or alias ARN. All are valid possibilities.
+        from moto.kms import kms_backends
+
+        kms = kms_backends[self.account_id][self.region_name]  # type: ignore[attr-defined]
+        kms_key = kms.describe_key(kms_key_id)
+        self.default_kms_key_id = kms_key.arn
+        return kms_key.arn
diff --git a/moto/ec2/responses/elastic_block_store.py b/moto/ec2/responses/elastic_block_store.py
index 968b4629a..d095c1d30 100644
--- a/moto/ec2/responses/elastic_block_store.py
+++ b/moto/ec2/responses/elastic_block_store.py
@@ -219,6 +219,13 @@ class ElasticBlockStore(EC2BaseResponse):
             "ElasticBlockStore.reset_snapshot_attribute is not yet implemented"
         )
 
+    def modify_ebs_default_kms_key_id(self) -> str:
+        self.error_on_dryrun()
+        kms_key_id = self._get_param("KmsKeyId")
+        new_default_kms_arn = self.ec2_backend.modify_ebs_default_kms_key_id(kms_key_id)
+        template = self.response_template(MODIFY_EBS_DEFAULT_KMS_KEY_ID_RESPONSE)
+        return template.render(kmsKeyId=new_default_kms_arn)
+
 
 CREATE_VOLUME_RESPONSE = """<CreateVolumeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
@@ -499,3 +506,11 @@ DESCRIBE_VOLUMES_MODIFICATIONS_RESPONSE = """
       {% endfor %}
     </volumeModificationSet>
 </DescribeVolumesModificationsResponse>"""
+
+
+MODIFY_EBS_DEFAULT_KMS_KEY_ID_RESPONSE = """
+<ModifyEbsDefaultKmsKeyIdResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
+    <requestId>666d2944-9276-4d6a-be12-1f4ada972fd8</requestId>
+    <kmsKeyId>{{ kmsKeyId }}</kmsKeyId>
+</ModifyEbsDefaultKmsKeyIdResponse>
+"""
diff --git a/moto/fsx/models.py b/moto/fsx/models.py
index 554974ec5..abceebbd1 100644
--- a/moto/fsx/models.py
+++ b/moto/fsx/models.py
@@ -1,7 +1,7 @@
 """FSxBackend class with methods for supported APIs."""
 
-from datetime import datetime
 from typing import Any, Dict, List, Optional, Tuple
+from uuid import uuid4
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
@@ -36,7 +36,7 @@ class FileSystem(BaseModel):
         ontap_configuration: Optional[Dict[str, Any]],
         open_zfs_configuration: Optional[Dict[str, Any]],
     ) -> None:
-        self.file_system_id = f"fs-moto{datetime.now().strftime('%Y%m%d%H%M%S')}"
+        self.file_system_id = f"fs-{uuid4().hex[:8]}"
         self.file_system_type = file_system_type
         if self.file_system_type not in FileSystemType.list_values():
             raise ValueError(f"Invalid FileSystemType: {self.file_system_type}")
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))
diff --git a/moto/wafv2/models.py b/moto/wafv2/models.py
index 8058cb4bf..5623e8138 100644
--- a/moto/wafv2/models.py
+++ b/moto/wafv2/models.py
@@ -54,6 +54,7 @@ class FakeWebACL(BaseModel):
         self.visibility_config = visibility_config
         self.default_action = default_action
         self.lock_token = str(mock_random.uuid4())[0:6]
+        self.associated_resources: List[str] = []
 
     def update(
         self,
@@ -168,25 +169,28 @@ class WAFV2Backend(BaseBackend):
         # TODO: self.load_balancers = OrderedDict()
 
     def associate_web_acl(self, web_acl_arn: str, resource_arn: str) -> None:
-        """
-        Only APIGateway Stages can be associated at the moment.
-        """
-        if web_acl_arn not in self.wacls:
+        web_acl = self.wacls.get(web_acl_arn)
+        if not web_acl:
             raise WAFNonexistentItemException
+        web_acl.associated_resources.append(resource_arn)
+        # Special Case - APIGateway wants to know about the WebACL it's associated to
         stage = self._find_apigw_stage(resource_arn)
         if stage:
             stage.web_acl_arn = web_acl_arn
 
     def disassociate_web_acl(self, resource_arn: str) -> None:
+        for web_acl in self.wacls.values():
+            if resource_arn in web_acl.associated_resources:
+                web_acl.associated_resources.remove(resource_arn)
+                break
         stage = self._find_apigw_stage(resource_arn)
         if stage:
             stage.web_acl_arn = None
 
     def get_web_acl_for_resource(self, resource_arn: str) -> Optional[FakeWebACL]:
-        stage = self._find_apigw_stage(resource_arn)
-        if stage and stage.web_acl_arn is not None:
-            wacl_arn = stage.web_acl_arn
-            return self.wacls.get(wacl_arn)
+        for wacl in self.wacls.values():
+            if resource_arn in wacl.associated_resources:
+                return wacl
         return None
 
     def _find_apigw_stage(self, resource_arn: str) -> Optional["Stage"]:  # type: ignore
diff --git a/other_langs/terraform/wafv2/cognito.tf b/other_langs/terraform/wafv2/cognito.tf
new file mode 100644
index 000000000..d32200e6d
--- /dev/null
+++ b/other_langs/terraform/wafv2/cognito.tf
@@ -0,0 +1,61 @@
+# Create a Cognito User Pool (this assumes you need a new one)
+resource "aws_cognito_user_pool" "example" {
+  name = "example-user-pool"
+}
+
+# Create a WAFv2 Web ACL
+resource "aws_wafv2_web_acl" "example" {
+  name        = "example-web-acl"
+  description = "Example WAFv2 Web ACL for Cognito User Pool"
+  scope       = "REGIONAL" # Since Cognito is a regional service
+  default_action {
+    allow {}
+  }
+
+  visibility_config {
+    cloudwatch_metrics_enabled = true
+    metric_name                = "exampleWebACL"
+    sampled_requests_enabled   = true
+  }
+
+  rule {
+    name     = "block-bad-ips"
+    priority = 1
+
+    action {
+      block {}
+    }
+
+    statement {
+      ip_set_reference_statement {
+        arn = aws_wafv2_ip_set.example.arn
+      }
+    }
+
+    visibility_config {
+      cloudwatch_metrics_enabled = true
+      metric_name                = "blockBadIps"
+      sampled_requests_enabled   = true
+    }
+  }
+}
+
+# Create an IP Set for the WAFv2 rule
+resource "aws_wafv2_ip_set" "example" {
+  name        = "example-ip-set"
+  description = "Example IP Set for bad actors"
+  scope       = "REGIONAL"
+
+  ip_address_version = "IPV4"
+
+  addresses = [
+    "192.0.2.0/24",  # Example IP addresses to block
+    "198.51.100.0/24"
+  ]
+}
+
+# Associate the WAFv2 Web ACL with the Cognito User Pool
+resource "aws_wafv2_web_acl_association" "example" {
+  resource_arn = aws_cognito_user_pool.example.arn
+  web_acl_arn  = aws_wafv2_web_acl.example.arn
+}
\ No newline at end of file
diff --git a/other_langs/terraform/wafv2/provider.tf b/other_langs/terraform/wafv2/provider.tf
new file mode 100644
index 000000000..ee528ec15
--- /dev/null
+++ b/other_langs/terraform/wafv2/provider.tf
@@ -0,0 +1,26 @@
+provider "aws" {
+  region                      = "us-east-1"
+  s3_use_path_style           = true
+  skip_credentials_validation = true
+  skip_metadata_api_check     = true
+  skip_requesting_account_id  = true
+
+  endpoints {
+    cognitoidp     = "http://localhost:5000"
+    iam            = "http://localhost:5000"
+    wafv2          = "http://localhost:5000"
+  }
+
+  access_key = "my-access-key"
+  secret_key = "my-secret-key"
+}
+
+terraform {
+  required_providers {
+
+    aws = {
+      source  = "hashicorp/aws"
+      version = "4.67.0"
+    }
+  }
+}
\ No newline at end of file


Test patch:
diff --git a/.github/workflows/tests_terraform_examples.yml b/.github/workflows/tests_terraform_examples.yml
index 82cf15891..672afce50 100644
--- a/.github/workflows/tests_terraform_examples.yml
+++ b/.github/workflows/tests_terraform_examples.yml
@@ -11,7 +11,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        service: ["acm", "autoscaling", "awslambda", "cloudfront", "codebuild", "ds", "elb", "iam", "rds", "route53", "sqs"]
+        service: ["acm", "autoscaling", "awslambda", "cloudfront", "codebuild", "ds", "elb", "iam", "rds", "route53", "sqs", "wafv2"]
 
     steps:
     - uses: actions/checkout@v4
diff --git a/tests/test_ec2/test_elastic_block_store.py b/tests/test_ec2/test_elastic_block_store.py
index 1fdd4d75b..a5636ed2f 100644
--- a/tests/test_ec2/test_elastic_block_store.py
+++ b/tests/test_ec2/test_elastic_block_store.py
@@ -1145,3 +1145,56 @@ def test_create_volume_with_throughput_fails():
         )
     assert ex.value.response["Error"]["Code"] == "InvalidParameterDependency"
     assert "Throughput" in ex.value.response["Error"]["Message"]
+
+
+@mock_aws
+def test_modify_ebs_default_kms_key_id():
+    new_default_alias = "alias/my-new-ebs-default"
+    kms = boto3.client("kms", region_name="us-east-1")
+    ec2 = boto3.client("ec2", region_name="us-east-1")
+
+    # Get the original default key
+    original_default_ebs_key_arn = kms.describe_key(KeyId="alias/aws/ebs")[
+        "KeyMetadata"
+    ]["Arn"]
+
+    # Create a new default key
+    new_default_key = kms.create_key(Description="test ebs key")["KeyMetadata"]
+    kms.create_alias(AliasName=new_default_alias, TargetKeyId=new_default_key["KeyId"])
+
+    # Modify the default ebs key using the KeyId
+    new_default_ebs_key_arn = ec2.modify_ebs_default_kms_key_id(
+        KmsKeyId=new_default_key["KeyId"]
+    )["KmsKeyId"]
+
+    assert new_default_ebs_key_arn != original_default_ebs_key_arn
+
+    # Creating an encrypted volume should create (and use) the new default key.
+    resource = boto3.resource("ec2", region_name="us-east-1")
+    volume = resource.create_volume(
+        AvailabilityZone="us-east-1a", Encrypted=True, Size=10
+    )
+
+    assert volume.kms_key_id == new_default_ebs_key_arn
+    assert volume.encrypted is True
+
+    # Revert to the original default key by passing empty string
+    ec2.modify_ebs_default_kms_key_id(KmsKeyId="")
+    # Creating an encrypted volume should create (and use) the original default key.
+    volume = resource.create_volume(
+        AvailabilityZone="us-east-1a", Encrypted=True, Size=10
+    )
+    assert volume.kms_key_id == original_default_ebs_key_arn
+
+    # Modify the default ebs key using a key alias
+    new_default_ebs_key_arn = ec2.modify_ebs_default_kms_key_id(
+        KmsKeyId=new_default_alias
+    )["KmsKeyId"]
+    assert new_default_ebs_key_arn != original_default_ebs_key_arn
+    ec2.modify_ebs_default_kms_key_id(KmsKeyId="")
+
+    # Modify the default ebs key using a key arn
+    new_default_ebs_key_arn = ec2.modify_ebs_default_kms_key_id(
+        KmsKeyId=new_default_key["Arn"]
+    )["KmsKeyId"]
+    assert new_default_ebs_key_arn != original_default_ebs_key_arn
diff --git a/tests/test_fsx/test_fsx.py b/tests/test_fsx/test_fsx.py
index c9ea7b716..8246d3a8b 100644
--- a/tests/test_fsx/test_fsx.py
+++ b/tests/test_fsx/test_fsx.py
@@ -1,5 +1,6 @@
 """Unit tests for fsx-supported APIs."""
 
+import re
 import time
 
 import boto3
@@ -32,7 +33,7 @@ def test_create_filesystem(client):
 
     file_system = resp["FileSystem"]
 
-    assert file_system["FileSystemId"].startswith("fs-moto")
+    assert re.match(r"^fs-[0-9a-f]{8,}$", file_system["FileSystemId"])
     assert file_system["FileSystemType"] == "LUSTRE"
 
 
@@ -90,7 +91,6 @@ def test_delete_file_system():
         SecurityGroupIds=FAKE_SECURITY_GROUP_IDS,
     )
 
-    assert fs["FileSystem"]["FileSystemId"].startswith("fs-moto")
     file_system_id = fs["FileSystem"]["FileSystemId"]
     resp = client.delete_file_system(FileSystemId=file_system_id)
     assert resp["FileSystemId"] == file_system_id
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"
+        )
diff --git a/tests/test_wafv2/test_wafv2_integration.py b/tests/test_wafv2/test_wafv2_integration.py
index bf536b594..558982e30 100644
--- a/tests/test_wafv2/test_wafv2_integration.py
+++ b/tests/test_wafv2/test_wafv2_integration.py
@@ -20,13 +20,15 @@ def test_associate_with_unknown_resource():
         WebACLArn=wacl_arn,
         ResourceArn="arn:aws:apigateway:us-east-1::/restapis/unknown/stages/unknown",
     )
-    conn.associate_web_acl(WebACLArn=wacl_arn, ResourceArn="unknownarnwithminlength20")
+    random_arn = "arn:aws:cognito-idp:us-east-1:123456789012:userpool/us-east-1_a7e89b7ced4244f48000bd5bd191d871"
+    conn.associate_web_acl(WebACLArn=wacl_arn, ResourceArn=random_arn)
+
+    wacl = conn.get_web_acl_for_resource(ResourceArn=random_arn)["WebACL"]
+    assert wacl["ARN"] == wacl_arn
 
     # We can validate if the WebACL exists
     with pytest.raises(ClientError) as exc:
-        conn.associate_web_acl(
-            WebACLArn=f"{wacl_arn}2", ResourceArn="unknownarnwithminlength20"
-        )
+        conn.associate_web_acl(WebACLArn=f"{wacl_arn}2", ResourceArn=random_arn)
     err = exc.value.response["Error"]
     assert err["Code"] == "WAFNonexistentItemException"
     assert err["Message"] == (
@@ -79,6 +81,7 @@ def test_get_web_acl_for_resource():
 @mock_aws
 def test_disassociate_unknown_resource():
     conn = boto3.client("wafv2", region_name="us-east-1")
+    conn.create_web_acl(**CREATE_WEB_ACL_BODY("John", "REGIONAL"))
     # Nothing happens
     conn.disassociate_web_acl(ResourceArn="unknownarnwithlength20")