Repository: getmoto/moto
Pull request #8162 (2024-09-26 20:10:26)

batch: fix bug in add_parameters_to_command
https://github.com/getmoto/moto/pull/7973 introduced support for parameters in job commands.
However, the `_add_parameters_to_command` function had a bug that resulted in duplicated arguments, when the number of arguments was greater than one.

This was due to the nested list comprehension logic adopted.

For instance, the  given the following job parameters `{'dossier_md5': 'abc', 'event_id': 'E12340060311'}` and the following command `['--dossier_md5', 'Ref::dossier_md5', '--event_id', 'Ref::event_id']` we were given the following output:
```
['--dossier_md5', '--dossier_md5', 'abc', 'Ref::dossier_md5', '--event_id', '--event_id', 'Ref::event_id', 'E12340060311']
```

This PR adjusts the logic accordingly and adds a simple unit test to verify the fix.



codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8162?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 [(`9e621f5`)](https://app.codecov.io/gh/getmoto/moto/commit/9e621f5a3f7db54a4d8b30c007095e088a0ee7d9?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    #8162      +/-   ##
==========================================
- Coverage   94.50%   94.50%   -0.01%     
==========================================
  Files        1150     1150              
  Lines       99203    99202       -1     
==========================================
- Hits        93752    93751       -1     
  Misses       5451     5451              
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8162/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/8162/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `28.79% <100.00%> (+<0.01%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8162/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_down: |

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/8162?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/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/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_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_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")