Repository: getmoto/moto
Pull request #7671 (2024-05-05 18:07:23)

Feature: EMR Serverless job runs
The existing EMR Serverless implementation has Applications implemented, but not Job Runs on those Applications. This PR adds those.

There are also some minor changes, like typos or copy/paste mistakes.


Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 8b4411847..716a66fdd 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -706,27 +706,37 @@
 
 ## bedrock
 <details>
-<summary>65% implemented</summary>
+<summary>43% implemented</summary>
 
+- [ ] create_evaluation_job
+- [ ] create_guardrail
+- [ ] create_guardrail_version
 - [X] create_model_customization_job
 - [ ] create_provisioned_model_throughput
 - [X] delete_custom_model
+- [ ] delete_guardrail
 - [X] delete_model_invocation_logging_configuration
 - [ ] delete_provisioned_model_throughput
 - [X] get_custom_model
+- [ ] get_evaluation_job
 - [ ] get_foundation_model
+- [ ] get_guardrail
 - [X] get_model_customization_job
 - [X] get_model_invocation_logging_configuration
 - [ ] get_provisioned_model_throughput
 - [X] list_custom_models
+- [ ] list_evaluation_jobs
 - [ ] list_foundation_models
+- [ ] list_guardrails
 - [X] list_model_customization_jobs
 - [ ] list_provisioned_model_throughputs
 - [X] list_tags_for_resource
 - [X] put_model_invocation_logging_configuration
+- [ ] stop_evaluation_job
 - [X] stop_model_customization_job
 - [X] tag_resource
 - [X] untag_resource
+- [ ] update_guardrail
 - [ ] update_provisioned_model_throughput
 </details>
 
@@ -1334,6 +1344,7 @@
 - [ ] put_webhook
 - [ ] register_webhook_with_third_party
 - [ ] retry_stage_execution
+- [ ] rollback_stage
 - [ ] start_pipeline_execution
 - [ ] stop_pipeline_execution
 - [X] tag_resource
@@ -2024,7 +2035,7 @@
 
 ## dynamodb
 <details>
-<summary>56% implemented</summary>
+<summary>59% implemented</summary>
 
 - [X] batch_execute_statement
 - [X] batch_get_item
@@ -2043,7 +2054,7 @@
 - [ ] describe_export
 - [ ] describe_global_table
 - [ ] describe_global_table_settings
-- [ ] describe_import
+- [X] describe_import
 - [ ] describe_kinesis_streaming_destination
 - [ ] describe_limits
 - [X] describe_table
@@ -2056,7 +2067,7 @@
 - [ ] export_table_to_point_in_time
 - [X] get_item
 - [ ] get_resource_policy
-- [ ] import_table
+- [X] import_table
 - [X] list_backups
 - [ ] list_contributor_insights
 - [ ] list_exports
@@ -2554,6 +2565,7 @@
 - [ ] get_host_reservation_purchase_preview
 - [ ] get_image_block_public_access_state
 - [ ] get_instance_metadata_defaults
+- [ ] get_instance_tpm_ek_pub
 - [ ] get_instance_types_from_instance_requirements
 - [ ] get_instance_uefi_data
 - [ ] get_ipam_address_history
@@ -3257,11 +3269,12 @@
 
 ## emr-containers
 <details>
-<summary>40% implemented</summary>
+<summary>34% implemented</summary>
 
 - [X] cancel_job_run
 - [ ] create_job_template
 - [ ] create_managed_endpoint
+- [ ] create_security_configuration
 - [X] create_virtual_cluster
 - [ ] delete_job_template
 - [ ] delete_managed_endpoint
@@ -3269,11 +3282,13 @@
 - [X] describe_job_run
 - [ ] describe_job_template
 - [ ] describe_managed_endpoint
+- [ ] describe_security_configuration
 - [X] describe_virtual_cluster
 - [ ] get_managed_endpoint_session_credentials
 - [X] list_job_runs
 - [ ] list_job_templates
 - [ ] list_managed_endpoints
+- [ ] list_security_configurations
 - [ ] list_tags_for_resource
 - [X] list_virtual_clusters
 - [X] start_job_run
@@ -3283,19 +3298,19 @@
 
 ## emr-serverless
 <details>
-<summary>46% implemented</summary>
+<summary>73% implemented</summary>
 
-- [ ] cancel_job_run
+- [X] cancel_job_run
 - [X] create_application
 - [X] delete_application
 - [X] get_application
 - [ ] get_dashboard_for_job_run
-- [ ] get_job_run
+- [X] get_job_run
 - [X] list_applications
-- [ ] list_job_runs
+- [X] list_job_runs
 - [ ] list_tags_for_resource
 - [X] start_application
-- [ ] start_job_run
+- [X] start_job_run
 - [X] stop_application
 - [ ] tag_resource
 - [ ] untag_resource
@@ -5487,6 +5502,7 @@
 - [ ] create_batch_inference_job
 - [ ] create_batch_segment_job
 - [ ] create_campaign
+- [ ] create_data_deletion_job
 - [ ] create_dataset
 - [ ] create_dataset_export_job
 - [ ] create_dataset_group
@@ -5511,6 +5527,7 @@
 - [ ] describe_batch_inference_job
 - [ ] describe_batch_segment_job
 - [ ] describe_campaign
+- [ ] describe_data_deletion_job
 - [ ] describe_dataset
 - [ ] describe_dataset_export_job
 - [ ] describe_dataset_group
@@ -5528,6 +5545,7 @@
 - [ ] list_batch_inference_jobs
 - [ ] list_batch_segment_jobs
 - [ ] list_campaigns
+- [ ] list_data_deletion_jobs
 - [ ] list_dataset_export_jobs
 - [ ] list_dataset_groups
 - [ ] list_dataset_import_jobs
@@ -5860,6 +5878,7 @@
 - [ ] update_public_sharing_settings
 - [ ] update_refresh_schedule
 - [ ] update_role_custom_permission
+- [ ] update_spice_capacity_configuration
 - [ ] update_template
 - [ ] update_template_alias
 - [ ] update_template_permissions
@@ -6789,7 +6808,7 @@
 
 ## s3control
 <details>
-<summary>9% implemented</summary>
+<summary>10% implemented</summary>
 
 - [ ] associate_access_grants_identity_center
 - [ ] create_access_grant
@@ -6864,7 +6883,7 @@
 - [ ] list_tags_for_resource
 - [ ] put_access_grants_instance_resource_policy
 - [ ] put_access_point_configuration_for_object_lambda
-- [ ] put_access_point_policy
+- [X] put_access_point_policy
 - [ ] put_access_point_policy_for_object_lambda
 - [ ] put_bucket_lifecycle_configuration
 - [ ] put_bucket_policy
@@ -7674,6 +7693,7 @@
 - [ ] describe_instance_patch_states
 - [ ] describe_instance_patch_states_for_patch_group
 - [ ] describe_instance_patches
+- [ ] describe_instance_properties
 - [ ] describe_inventory_deletions
 - [ ] describe_maintenance_window_execution_task_invocations
 - [ ] describe_maintenance_window_execution_tasks
@@ -7772,7 +7792,7 @@
 
 ## sso-admin
 <details>
-<summary>24% implemented</summary>
+<summary>27% implemented</summary>
 
 - [X] attach_customer_managed_policy_reference_to_permission_set
 - [X] attach_managed_policy_to_permission_set
@@ -7795,8 +7815,8 @@
 - [X] delete_permission_set
 - [ ] delete_permissions_boundary_from_permission_set
 - [ ] delete_trusted_token_issuer
-- [ ] describe_account_assignment_creation_status
-- [ ] describe_account_assignment_deletion_status
+- [X] describe_account_assignment_creation_status
+- [X] describe_account_assignment_deletion_status
 - [ ] describe_application
 - [ ] describe_application_assignment
 - [ ] describe_application_provider
@@ -7851,7 +7871,7 @@
 
 ## stepfunctions
 <details>
-<summary>55% implemented</summary>
+<summary>54% implemented</summary>
 
 - [ ] create_activity
 - [X] create_state_machine
@@ -7889,6 +7909,7 @@
 - [X] update_map_run
 - [X] update_state_machine
 - [ ] update_state_machine_alias
+- [ ] validate_state_machine_definition
 </details>
 
 ## sts
diff --git a/docs/docs/services/bedrock.rst b/docs/docs/services/bedrock.rst
index e86d0faf0..f9f907a8e 100644
--- a/docs/docs/services/bedrock.rst
+++ b/docs/docs/services/bedrock.rst
@@ -16,13 +16,19 @@ bedrock
 
 |start-h3| Implemented features for this service |end-h3|
 
+- [ ] create_evaluation_job
+- [ ] create_guardrail
+- [ ] create_guardrail_version
 - [X] create_model_customization_job
 - [ ] create_provisioned_model_throughput
 - [X] delete_custom_model
+- [ ] delete_guardrail
 - [X] delete_model_invocation_logging_configuration
 - [ ] delete_provisioned_model_throughput
 - [X] get_custom_model
+- [ ] get_evaluation_job
 - [ ] get_foundation_model
+- [ ] get_guardrail
 - [X] get_model_customization_job
 - [X] get_model_invocation_logging_configuration
 - [ ] get_provisioned_model_throughput
@@ -31,13 +37,17 @@ bedrock
         The foundation_model_arn_equals-argument is not yet supported
         
 
+- [ ] list_evaluation_jobs
 - [ ] list_foundation_models
+- [ ] list_guardrails
 - [X] list_model_customization_jobs
 - [ ] list_provisioned_model_throughputs
 - [X] list_tags_for_resource
 - [X] put_model_invocation_logging_configuration
+- [ ] stop_evaluation_job
 - [X] stop_model_customization_job
 - [X] tag_resource
 - [X] untag_resource
+- [ ] update_guardrail
 - [ ] update_provisioned_model_throughput
 
diff --git a/docs/docs/services/codepipeline.rst b/docs/docs/services/codepipeline.rst
index 9247933f1..4a53c967e 100644
--- a/docs/docs/services/codepipeline.rst
+++ b/docs/docs/services/codepipeline.rst
@@ -47,6 +47,7 @@ codepipeline
 - [ ] put_webhook
 - [ ] register_webhook_with_third_party
 - [ ] retry_stage_execution
+- [ ] rollback_stage
 - [ ] start_pipeline_execution
 - [ ] stop_pipeline_execution
 - [X] tag_resource
diff --git a/docs/docs/services/dynamodb.rst b/docs/docs/services/dynamodb.rst
index 10c9c57da..872802d6f 100644
--- a/docs/docs/services/dynamodb.rst
+++ b/docs/docs/services/dynamodb.rst
@@ -35,7 +35,7 @@ dynamodb
 - [ ] describe_export
 - [ ] describe_global_table
 - [ ] describe_global_table_settings
-- [ ] describe_import
+- [X] describe_import
 - [ ] describe_kinesis_streaming_destination
 - [ ] describe_limits
 - [X] describe_table
@@ -58,7 +58,13 @@ dynamodb
 - [ ] export_table_to_point_in_time
 - [X] get_item
 - [ ] get_resource_policy
-- [ ] import_table
+- [X] import_table
+  
+        Only InputFormat=DYNAMODB_JSON is supported so far.
+        InputCompressionType=ZSTD is not supported.
+        Other parameters that are not supported: InputFormatOptions, CloudWatchLogGroupArn
+        
+
 - [X] list_backups
 - [ ] list_contributor_insights
 - [ ] list_exports
diff --git a/docs/docs/services/ec2.rst b/docs/docs/services/ec2.rst
index e6c917d32..e5745efcb 100644
--- a/docs/docs/services/ec2.rst
+++ b/docs/docs/services/ec2.rst
@@ -495,6 +495,7 @@ ec2
 - [ ] get_host_reservation_purchase_preview
 - [ ] get_image_block_public_access_state
 - [ ] get_instance_metadata_defaults
+- [ ] get_instance_tpm_ek_pub
 - [ ] get_instance_types_from_instance_requirements
 - [ ] get_instance_uefi_data
 - [ ] get_ipam_address_history
diff --git a/docs/docs/services/emr-containers.rst b/docs/docs/services/emr-containers.rst
index 779f0d27f..888b7cc84 100644
--- a/docs/docs/services/emr-containers.rst
+++ b/docs/docs/services/emr-containers.rst
@@ -19,6 +19,7 @@ emr-containers
 - [X] cancel_job_run
 - [ ] create_job_template
 - [ ] create_managed_endpoint
+- [ ] create_security_configuration
 - [X] create_virtual_cluster
 - [ ] delete_job_template
 - [ ] delete_managed_endpoint
@@ -26,11 +27,13 @@ emr-containers
 - [X] describe_job_run
 - [ ] describe_job_template
 - [ ] describe_managed_endpoint
+- [ ] describe_security_configuration
 - [X] describe_virtual_cluster
 - [ ] get_managed_endpoint_session_credentials
 - [X] list_job_runs
 - [ ] list_job_templates
 - [ ] list_managed_endpoints
+- [ ] list_security_configurations
 - [ ] list_tags_for_resource
 - [X] list_virtual_clusters
 - [X] start_job_run
diff --git a/docs/docs/services/emr-serverless.rst b/docs/docs/services/emr-serverless.rst
index 4513d1743..ac63cc902 100644
--- a/docs/docs/services/emr-serverless.rst
+++ b/docs/docs/services/emr-serverless.rst
@@ -16,17 +16,17 @@ emr-serverless
 
 |start-h3| Implemented features for this service |end-h3|
 
-- [ ] cancel_job_run
+- [X] cancel_job_run
 - [X] create_application
 - [X] delete_application
 - [X] get_application
 - [ ] get_dashboard_for_job_run
-- [ ] get_job_run
+- [X] get_job_run
 - [X] list_applications
-- [ ] list_job_runs
+- [X] list_job_runs
 - [ ] list_tags_for_resource
 - [X] start_application
-- [ ] start_job_run
+- [X] start_job_run
 - [X] stop_application
 - [ ] tag_resource
 - [ ] untag_resource
diff --git a/docs/docs/services/personalize.rst b/docs/docs/services/personalize.rst
index 2a07c5726..756a25694 100644
--- a/docs/docs/services/personalize.rst
+++ b/docs/docs/services/personalize.rst
@@ -19,6 +19,7 @@ personalize
 - [ ] create_batch_inference_job
 - [ ] create_batch_segment_job
 - [ ] create_campaign
+- [ ] create_data_deletion_job
 - [ ] create_dataset
 - [ ] create_dataset_export_job
 - [ ] create_dataset_group
@@ -43,6 +44,7 @@ personalize
 - [ ] describe_batch_inference_job
 - [ ] describe_batch_segment_job
 - [ ] describe_campaign
+- [ ] describe_data_deletion_job
 - [ ] describe_dataset
 - [ ] describe_dataset_export_job
 - [ ] describe_dataset_group
@@ -60,6 +62,7 @@ personalize
 - [ ] list_batch_inference_jobs
 - [ ] list_batch_segment_jobs
 - [ ] list_campaigns
+- [ ] list_data_deletion_jobs
 - [ ] list_dataset_export_jobs
 - [ ] list_dataset_groups
 - [ ] list_dataset_import_jobs
diff --git a/docs/docs/services/quicksight.rst b/docs/docs/services/quicksight.rst
index 15ff6cd9c..67f3aaed2 100644
--- a/docs/docs/services/quicksight.rst
+++ b/docs/docs/services/quicksight.rst
@@ -193,6 +193,7 @@ quicksight
 - [ ] update_public_sharing_settings
 - [ ] update_refresh_schedule
 - [ ] update_role_custom_permission
+- [ ] update_spice_capacity_configuration
 - [ ] update_template
 - [ ] update_template_alias
 - [ ] update_template_permissions
diff --git a/docs/docs/services/s3control.rst b/docs/docs/services/s3control.rst
index c5476c1cb..281c64ec9 100644
--- a/docs/docs/services/s3control.rst
+++ b/docs/docs/services/s3control.rst
@@ -91,7 +91,7 @@ s3control
 - [ ] list_tags_for_resource
 - [ ] put_access_grants_instance_resource_policy
 - [ ] put_access_point_configuration_for_object_lambda
-- [ ] put_access_point_policy
+- [X] put_access_point_policy
 - [ ] put_access_point_policy_for_object_lambda
 - [ ] put_bucket_lifecycle_configuration
 - [ ] put_bucket_policy
diff --git a/docs/docs/services/ssm.rst b/docs/docs/services/ssm.rst
index 24db63e85..dd733ff34 100644
--- a/docs/docs/services/ssm.rst
+++ b/docs/docs/services/ssm.rst
@@ -85,6 +85,7 @@ ssm
 - [ ] describe_instance_patch_states
 - [ ] describe_instance_patch_states_for_patch_group
 - [ ] describe_instance_patches
+- [ ] describe_instance_properties
 - [ ] describe_inventory_deletions
 - [ ] describe_maintenance_window_execution_task_invocations
 - [ ] describe_maintenance_window_execution_tasks
diff --git a/docs/docs/services/sso-admin.rst b/docs/docs/services/sso-admin.rst
index 2768397ff..600e7c99b 100644
--- a/docs/docs/services/sso-admin.rst
+++ b/docs/docs/services/sso-admin.rst
@@ -37,8 +37,8 @@ sso-admin
 - [X] delete_permission_set
 - [ ] delete_permissions_boundary_from_permission_set
 - [ ] delete_trusted_token_issuer
-- [ ] describe_account_assignment_creation_status
-- [ ] describe_account_assignment_deletion_status
+- [X] describe_account_assignment_creation_status
+- [X] describe_account_assignment_deletion_status
 - [ ] describe_application
 - [ ] describe_application_assignment
 - [ ] describe_application_provider
diff --git a/docs/docs/services/stepfunctions.rst b/docs/docs/services/stepfunctions.rst
index 067ddedee..5c01dc0fd 100644
--- a/docs/docs/services/stepfunctions.rst
+++ b/docs/docs/services/stepfunctions.rst
@@ -52,4 +52,5 @@ stepfunctions
 - [X] update_map_run
 - [X] update_state_machine
 - [ ] update_state_machine_alias
+- [ ] validate_state_machine_definition
 
diff --git a/moto/databrew/models.py b/moto/databrew/models.py
index 5dc01f198..825ff42f3 100644
--- a/moto/databrew/models.py
+++ b/moto/databrew/models.py
@@ -477,6 +477,9 @@ class FakeRecipe(BaseModel):
         del self.versions[self.latest_working.version]
         self.versions[self.latest_published.version] = self.latest_published
         self.latest_working.version = self.latest_published.version + 0.1
+
+        if self.latest_published.published_date:
+            self.latest_working.created_time = self.latest_published.published_date
         self.versions[self.latest_working.version] = self.latest_working
 
     def update(
diff --git a/moto/dynamodb/models/__init__.py b/moto/dynamodb/models/__init__.py
index c6634e195..574152f60 100644
--- a/moto/dynamodb/models/__init__.py
+++ b/moto/dynamodb/models/__init__.py
@@ -33,6 +33,7 @@ from moto.dynamodb.models.table import (
     RestoredTable,
     Table,
 )
+from moto.dynamodb.models.table_import import TableImport
 from moto.dynamodb.parsing import partiql
 from moto.dynamodb.parsing.executors import UpdateExpressionExecutor
 from moto.dynamodb.parsing.expressions import UpdateExpressionParser  # type: ignore
@@ -44,6 +45,7 @@ class DynamoDBBackend(BaseBackend):
         super().__init__(region_name, account_id)
         self.tables: Dict[str, Table] = OrderedDict()
         self.backups: Dict[str, Backup] = OrderedDict()
+        self.table_imports: Dict[str, TableImport] = {}
 
     @staticmethod
     def default_vpc_endpoint_service(
@@ -60,11 +62,36 @@ class DynamoDBBackend(BaseBackend):
             base_endpoint_dns_names=[f"dynamodb.{service_region}.amazonaws.com"],
         )
 
-    def create_table(self, name: str, **params: Any) -> Table:
+    def create_table(
+        self,
+        name: str,
+        schema: List[Dict[str, str]],
+        throughput: Optional[Dict[str, int]],
+        attr: List[Dict[str, str]],
+        global_indexes: Optional[List[Dict[str, Any]]],
+        indexes: Optional[List[Dict[str, Any]]],
+        streams: Optional[Dict[str, Any]],
+        billing_mode: str,
+        sse_specification: Optional[Dict[str, Any]],
+        tags: List[Dict[str, str]],
+        deletion_protection_enabled: bool,
+    ) -> Table:
         if name in self.tables:
             raise ResourceInUseException(f"Table already exists: {name}")
         table = Table(
-            name, account_id=self.account_id, region=self.region_name, **params
+            name,
+            account_id=self.account_id,
+            region=self.region_name,
+            schema=schema,
+            throughput=throughput,
+            attr=attr,
+            global_indexes=global_indexes,
+            indexes=indexes,
+            streams=streams,
+            billing_mode=billing_mode,
+            sse_specification=sse_specification,
+            tags=tags,
+            deletion_protection_enabled=deletion_protection_enabled,
         )
         self.tables[name] = table
         return table
@@ -909,5 +936,41 @@ class DynamoDBBackend(BaseBackend):
                 responses[idx] = {"Error": {"Code": e.name, "Message": e.message}}  # type: ignore
         return responses
 
+    def import_table(
+        self,
+        s3_source: Dict[str, str],
+        input_format: Optional[str],
+        compression_type: Optional[str],
+        table_name: str,
+        billing_mode: str,
+        throughput: Optional[Dict[str, int]],
+        key_schema: List[Dict[str, str]],
+        global_indexes: Optional[List[Dict[str, Any]]],
+        attrs: List[Dict[str, str]],
+    ) -> TableImport:
+        """
+        Only InputFormat=DYNAMODB_JSON is supported so far.
+        InputCompressionType=ZSTD is not supported.
+        Other parameters that are not supported: InputFormatOptions, CloudWatchLogGroupArn
+        """
+        table_import = TableImport(
+            account_id=self.account_id,
+            s3_source=s3_source,
+            region_name=self.region_name,
+            table_name=table_name,
+            billing_mode=billing_mode,
+            throughput=throughput,
+            key_schema=key_schema,
+            global_indexes=global_indexes,
+            attrs=attrs,
+            compression_type=compression_type,
+        )
+        self.table_imports[table_import.arn] = table_import
+        table_import.start()
+        return table_import
+
+    def describe_import(self, import_arn: str) -> TableImport:
+        return self.table_imports[import_arn]
+
 
 dynamodb_backends = BackendDict(DynamoDBBackend, "dynamodb")
diff --git a/moto/dynamodb/models/table.py b/moto/dynamodb/models/table.py
index 43c55c666..c309ff925 100644
--- a/moto/dynamodb/models/table.py
+++ b/moto/dynamodb/models/table.py
@@ -386,14 +386,22 @@ class Table(CloudFormationModel):
             params["schema"] = properties["KeySchema"]
         if "AttributeDefinitions" in properties:
             params["attr"] = properties["AttributeDefinitions"]
-        if "GlobalSecondaryIndexes" in properties:
-            params["global_indexes"] = properties["GlobalSecondaryIndexes"]
-        if "ProvisionedThroughput" in properties:
-            params["throughput"] = properties["ProvisionedThroughput"]
-        if "LocalSecondaryIndexes" in properties:
-            params["indexes"] = properties["LocalSecondaryIndexes"]
-        if "StreamSpecification" in properties:
-            params["streams"] = properties["StreamSpecification"]
+        params["global_indexes"] = properties.get("GlobalSecondaryIndexes", [])
+        params["throughput"] = properties.get("ProvisionedThroughput")
+        params["indexes"] = properties.get("LocalSecondaryIndexes", [])
+        params["streams"] = properties.get("StreamSpecification")
+        params["tags"] = properties.get("Tags")
+        params["deletion_protection_enabled"] = properties.get(
+            "DeletionProtectionEnabled", False
+        )
+        params["sse_specification"] = properties.get("SSESpecification")
+
+        billing_mode = (
+            "PAY_PER_REQUEST"
+            if properties.get("BillingMode") == "PAY_PER_REQUEST"
+            else "PROVISIONED"
+        )
+        params["billing_mode"] = billing_mode
 
         table = dynamodb_backends[account_id][region_name].create_table(
             name=resource_name, **params
diff --git a/moto/dynamodb/models/table_import.py b/moto/dynamodb/models/table_import.py
new file mode 100644
index 000000000..22e720312
--- /dev/null
+++ b/moto/dynamodb/models/table_import.py
@@ -0,0 +1,138 @@
+from threading import Thread
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from uuid import uuid4
+
+from moto.core.utils import gzip_decompress
+
+if TYPE_CHECKING:
+    from moto.dynamodb.models import DynamoDBBackend
+    from moto.dynamodb.models.table import Table
+    from moto.s3.models import FakeBucket, S3Backend
+
+
+class TableImport(Thread):
+    def __init__(
+        self,
+        account_id: str,
+        s3_source: Dict[str, str],
+        region_name: str,
+        table_name: str,
+        billing_mode: str,
+        throughput: Optional[Dict[str, int]],
+        key_schema: List[Dict[str, str]],
+        global_indexes: Optional[List[Dict[str, Any]]],
+        attrs: List[Dict[str, str]],
+        compression_type: Optional[str],
+    ):
+        super().__init__()
+        self.arn = f"arn:aws:dynamodb:{region_name}:{account_id}:table/{table_name}/import/{str(uuid4()).replace('-', '')}"
+        self.status = "IN_PROGRESS"
+        self.account_id = account_id
+        self.s3_source = s3_source
+        self.region_name = region_name
+
+        self.table_name = table_name
+        self.billing_mode = billing_mode
+        self.throughput = throughput
+        self.key_schema = key_schema
+        self.global_indexes = global_indexes
+        self.attrs = attrs
+        self.compression_type = compression_type
+
+        self.failure_code: Optional[str] = None
+        self.failure_message: Optional[str] = None
+        self.table: Optional["Table"] = None
+        self.table_arn = (
+            f"arn:aws:dynamodb:{self.region_name}:{self.account_id}:table/{table_name}"
+        )
+
+        self.processed_count = 0
+        self.processed_bytes = 0
+        self.error_count = 0
+        self.imported_count = 0
+
+    def run(self) -> None:
+        s3_bucket_name = self.s3_source["S3Bucket"]
+
+        try:
+            from moto.s3.models import s3_backends
+
+            s3_backend = s3_backends[self.account_id]["global"]
+            bucket = s3_backend.buckets[s3_bucket_name]
+        except KeyError:
+            self.status = "FAILED"
+            self.failure_code = "S3NoSuchBucket"
+            self.failure_message = "The specified bucket does not exist"
+            return
+
+        try:
+            self._process_s3_files(s3_backend, bucket)
+        except Exception as e:
+            self.status = "FAILED"
+            self.failure_code = "UNKNOWN"
+            self.failure_message = str(e)
+
+    def _process_s3_files(self, s3_backend: "S3Backend", bucket: "FakeBucket") -> None:
+        # CREATE TABLE
+        from moto.dynamodb.models import dynamodb_backends
+
+        dynamo: DynamoDBBackend = dynamodb_backends[self.account_id][self.region_name]
+        self.table = dynamo.create_table(
+            name=self.table_name,
+            billing_mode=self.billing_mode,
+            throughput=self.throughput,
+            schema=self.key_schema,
+            global_indexes=self.global_indexes,
+            indexes=None,
+            attr=self.attrs,
+            sse_specification=None,
+            streams=None,
+            tags=[],
+            deletion_protection_enabled=False,
+        )
+
+        # Load data from S3
+        keys, _, _, _ = s3_backend.list_objects(
+            bucket,
+            prefix=self.s3_source.get("S3KeyPrefix"),
+            delimiter=None,
+            marker=None,
+            max_keys=None,
+        )
+
+        from py_partiql_parser._internal.json_parser import JsonParser
+
+        parser = JsonParser()
+
+        for key in keys:
+            if self.compression_type == "GZIP":
+                content = gzip_decompress(key.value).decode("utf-8")
+            else:
+                content = key.value.decode("utf-8")
+            result = parser.parse(original=content)
+            if not isinstance(result, list):
+                result = [result]
+            for json_object in result:
+                try:
+                    self.processed_count += 1
+                    self.processed_bytes += len(json_object)
+                    self.table.put_item(item_attrs=json_object["Item"])
+                    self.imported_count += 1
+                except Exception as e:
+                    self.failure_message = str(e)
+                    self.error_count += 1
+
+        self.status = "COMPLETED" if self.error_count == 0 else "FAILED"
+
+    def response(self) -> Dict[str, Any]:
+        return {
+            "ImportArn": self.arn,
+            "ImportStatus": self.status,
+            "TableArn": self.table_arn,
+            "FailureCode": self.failure_code,
+            "FailureMessage": self.failure_message,
+            "ProcessedItemCount": self.processed_count,
+            "ProcessedSizeBytes": self.processed_bytes,
+            "ErrorCount": self.error_count,
+            "ImportedItemCount": self.imported_count,
+        }
diff --git a/moto/dynamodb/responses.py b/moto/dynamodb/responses.py
index 49131ae72..fb14ba753 100644
--- a/moto/dynamodb/responses.py
+++ b/moto/dynamodb/responses.py
@@ -210,27 +210,66 @@ class DynamoHandler(BaseResponse):
 
     def create_table(self) -> str:
         body = self.body
-        # get the table name
         table_name = body["TableName"]
         # check billing mode and get the throughput
         if "BillingMode" in body.keys() and body["BillingMode"] == "PAY_PER_REQUEST":
-            if "ProvisionedThroughput" in body.keys():
-                raise MockValidationException(
-                    "ProvisionedThroughput cannot be specified when BillingMode is PAY_PER_REQUEST"
-                )
-            throughput = None
             billing_mode = "PAY_PER_REQUEST"
-        else:  # Provisioned (default billing mode)
-            throughput = body.get("ProvisionedThroughput")
-            if throughput is None:
-                raise MockValidationException(
-                    "One or more parameter values were invalid: ReadCapacityUnits and WriteCapacityUnits must both be specified when BillingMode is PROVISIONED"
-                )
-            billing_mode = "PROVISIONED"
-        # getting ServerSideEncryption details
+        else:
+            billing_mode = "PROVISIONED"  # Default
+        throughput = body.get("ProvisionedThroughput")
         sse_spec = body.get("SSESpecification")
-        # getting the schema
         key_schema = body["KeySchema"]
+        attr = body["AttributeDefinitions"]
+        global_indexes = body.get("GlobalSecondaryIndexes")
+        local_secondary_indexes = body.get("LocalSecondaryIndexes")
+        streams = body.get("StreamSpecification")
+        tags = body.get("Tags", [])
+        deletion_protection_enabled = body.get("DeletionProtectionEnabled", False)
+
+        self._validate_table_creation(
+            billing_mode=billing_mode,
+            throughput=throughput,
+            key_schema=key_schema,
+            global_indexes=global_indexes,
+            local_secondary_indexes=local_secondary_indexes,
+            attr=attr,
+        )
+
+        table = self.dynamodb_backend.create_table(
+            table_name,
+            schema=key_schema,
+            throughput=throughput,
+            attr=attr,
+            global_indexes=global_indexes or [],
+            indexes=local_secondary_indexes or [],
+            streams=streams,
+            billing_mode=billing_mode,
+            sse_specification=sse_spec,
+            tags=tags,
+            deletion_protection_enabled=deletion_protection_enabled,
+        )
+        return dynamo_json_dump(table.describe())
+
+    def _validate_table_creation(
+        self,
+        billing_mode: str,
+        throughput: Optional[Dict[str, Any]],
+        key_schema: List[Dict[str, str]],
+        global_indexes: Optional[List[Dict[str, Any]]],
+        local_secondary_indexes: Optional[List[Dict[str, Any]]],
+        attr: List[Dict[str, str]],
+    ) -> None:
+        # Validate Throughput
+        if billing_mode == "PAY_PER_REQUEST" and throughput:
+            raise MockValidationException(
+                "ProvisionedThroughput cannot be specified when BillingMode is PAY_PER_REQUEST"
+            )
+        if billing_mode == "PROVISIONED" and throughput is None:
+            raise MockValidationException(
+                "One or more parameter values were invalid: ReadCapacityUnits and WriteCapacityUnits must both be specified when BillingMode is PROVISIONED"
+            )
+
+        # Validate KeySchema
         for idx, _key in enumerate(key_schema, start=1):
             key_type = _key["KeyType"]
             if key_type not in ["HASH", "RANGE"]:
@@ -245,79 +284,50 @@ class DynamoHandler(BaseResponse):
             provided_keys = ", ".join(key_elements)
             err = f"1 validation error detected: Value '[{provided_keys}]' at 'keySchema' failed to satisfy constraint: Member must have length less than or equal to 2"
             raise MockValidationException(err)
-        # getting attribute definition
-        attr = body["AttributeDefinitions"]
 
-        # getting/validating the indexes
-        global_indexes = body.get("GlobalSecondaryIndexes")
+        # Validate Global Indexes
         if global_indexes == []:
             raise MockValidationException(
                 "One or more parameter values were invalid: List of GlobalSecondaryIndexes is empty"
             )
-        global_indexes = global_indexes or []
-        for idx, g_idx in enumerate(global_indexes, start=1):
+        for idx, g_idx in enumerate(global_indexes or [], start=1):
             for idx2, _key in enumerate(g_idx["KeySchema"], start=1):
                 key_type = _key["KeyType"]
                 if key_type not in ["HASH", "RANGE"]:
                     position = f"globalSecondaryIndexes.{idx}.member.keySchema.{idx2}.member.keyType"
                     raise UnknownKeyType(key_type=key_type, position=position)
 
-        local_secondary_indexes = body.get("LocalSecondaryIndexes")
+        # Validate Local Indexes
         if local_secondary_indexes == []:
             raise MockValidationException(
                 "One or more parameter values were invalid: List of LocalSecondaryIndexes is empty"
             )
-        local_secondary_indexes = local_secondary_indexes or []
-        for idx, g_idx in enumerate(local_secondary_indexes, start=1):
+        for idx, g_idx in enumerate(local_secondary_indexes or [], start=1):
             for idx2, _key in enumerate(g_idx["KeySchema"], start=1):
                 key_type = _key["KeyType"]
                 if key_type not in ["HASH", "RANGE"]:
                     position = f"localSecondaryIndexes.{idx}.member.keySchema.{idx2}.member.keyType"
                     raise UnknownKeyType(key_type=key_type, position=position)
 
-        # Verify AttributeDefinitions list all
+        # Validate Attributes
         expected_attrs = []
         expected_attrs.extend([key["AttributeName"] for key in key_schema])
-        expected_attrs.extend(
-            schema["AttributeName"]
-            for schema in itertools.chain(
-                *list(idx["KeySchema"] for idx in local_secondary_indexes)
-            )
+        local_key_schemas = itertools.chain(
+            *list(idx["KeySchema"] for idx in (local_secondary_indexes or []))
         )
-        expected_attrs.extend(
-            schema["AttributeName"]
-            for schema in itertools.chain(
-                *list(idx["KeySchema"] for idx in global_indexes)
-            )
+        expected_attrs.extend(schema["AttributeName"] for schema in local_key_schemas)
+
+        global_key_schemas = itertools.chain(
+            *list(idx["KeySchema"] for idx in (global_indexes or []))
         )
+        expected_attrs.extend(schema["AttributeName"] for schema in global_key_schemas)
         expected_attrs = list(set(expected_attrs))
         expected_attrs.sort()
         actual_attrs = [item["AttributeName"] for item in attr]
         actual_attrs.sort()
+        has_index = global_indexes is not None or local_secondary_indexes is not None
         if actual_attrs != expected_attrs:
-            self._throw_attr_error(
-                actual_attrs, expected_attrs, global_indexes or local_secondary_indexes
-            )
-        # get the stream specification
-        streams = body.get("StreamSpecification")
-        # Get any tags
-        tags = body.get("Tags", [])
-        deletion_protection_enabled = body.get("DeletionProtectionEnabled", False)
-
-        table = self.dynamodb_backend.create_table(
-            table_name,
-            schema=key_schema,
-            throughput=throughput,
-            attr=attr,
-            global_indexes=global_indexes,
-            indexes=local_secondary_indexes,
-            streams=streams,
-            billing_mode=billing_mode,
-            sse_specification=sse_spec,
-            tags=tags,
-            deletion_protection_enabled=deletion_protection_enabled,
-        )
-        return dynamo_json_dump(table.describe())
+            self._throw_attr_error(actual_attrs, expected_attrs, has_index)
 
     def _throw_attr_error(
         self, actual_attrs: List[str], expected_attrs: List[str], indexes: bool
@@ -1172,3 +1182,43 @@ class DynamoHandler(BaseResponse):
         stmts = self.body.get("Statements", [])
         items = self.dynamodb_backend.batch_execute_statement(stmts)
         return dynamo_json_dump({"Responses": items})
+
+    def import_table(self) -> str:
+        params = self.body
+        s3_source = params.get("S3BucketSource")
+        input_format = params.get("InputFormat") or "DYNAMODB_JSON"
+        compression_type = params.get("InputCompressionType") or "NONE"
+        table_parameters = params.get("TableCreationParameters")
+        table_name = table_parameters["TableName"]
+        table_attrs = table_parameters["AttributeDefinitions"]
+        table_schema = table_parameters["KeySchema"]
+        table_billing = table_parameters.get("BillingMode")
+        table_throughput = table_parameters.get("ProvisionedThroughput")
+        global_indexes = table_parameters.get("GlobalSecondaryIndexes")
+
+        self._validate_table_creation(
+            billing_mode=table_billing,
+            throughput=table_throughput,
+            key_schema=table_schema,
+            global_indexes=global_indexes,
+            local_secondary_indexes=None,
+            attr=table_attrs,
+        )
+
+        import_table = self.dynamodb_backend.import_table(
+            s3_source=s3_source,
+            input_format=input_format,
+            compression_type=compression_type,
+            table_name=table_name,
+            billing_mode=table_billing or "PROVISIONED",
+            throughput=table_throughput,
+            key_schema=table_schema,
+            global_indexes=global_indexes,
+            attrs=table_attrs,
+        )
+        return json.dumps({"ImportTableDescription": import_table.response()})
+
+    def describe_import(self) -> str:
+        import_arn = self.body["ImportArn"]
+        import_table = self.dynamodb_backend.describe_import(import_arn)
+        return json.dumps({"ImportTableDescription": import_table.response()})
diff --git a/moto/ec2/models/network_acls.py b/moto/ec2/models/network_acls.py
index 5e9808778..aa11829e6 100644
--- a/moto/ec2/models/network_acls.py
+++ b/moto/ec2/models/network_acls.py
@@ -1,6 +1,7 @@
 from typing import Any, Dict, List, Optional
 
 from ..exceptions import (
+    DependencyViolationError,
     InvalidNetworkAclIdError,
     InvalidRouteTableIdError,
     NetworkAclEntryAlreadyExistsError,
@@ -61,6 +62,14 @@ class NetworkAclBackend:
             )
 
     def delete_network_acl(self, network_acl_id: str) -> "NetworkAcl":
+        if any(
+            network_acl.id == network_acl_id and len(network_acl.associations) > 0
+            for network_acl in self.network_acls.values()
+        ):
+            raise DependencyViolationError(
+                f"The network ACL '{network_acl_id}' has dependencies and cannot be deleted."
+            )
+
         deleted = self.network_acls.pop(network_acl_id, None)
         if not deleted:
             raise InvalidNetworkAclIdError(network_acl_id)
diff --git a/moto/ec2/models/route_tables.py b/moto/ec2/models/route_tables.py
index c063b359c..c557b5a70 100644
--- a/moto/ec2/models/route_tables.py
+++ b/moto/ec2/models/route_tables.py
@@ -116,6 +116,12 @@ class RouteTable(TaggedEC2Resource, CloudFormationModel):
                 for route in self.routes.values()
                 if route.vpc_pcx is not None
             ]
+        elif filter_name == "route.nat-gateway-id":
+            return [
+                route.nat_gateway.id
+                for route in self.routes.values()
+                if route.nat_gateway is not None
+            ]
         else:
             return super().get_filter_value(filter_name, "DescribeRouteTables")
 
diff --git a/moto/ec2/resources/instance_types.json b/moto/ec2/resources/instance_types.json
index 3911dde23..7d9709162 100644
--- a/moto/ec2/resources/instance_types.json
+++ b/moto/ec2/resources/instance_types.json
@@ -12878,18 +12878,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -13327,18 +13327,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -47838,18 +47838,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -48335,18 +48335,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -49006,18 +49006,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -49455,18 +49455,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -69880,18 +69880,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -70377,18 +70377,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -71048,18 +71048,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
@@ -71497,18 +71497,18 @@
    "Ipv6AddressesPerInterface": 50,
    "Ipv6Supported": true,
    "MaximumNetworkCards": 2,
-   "MaximumNetworkInterfaces": 14,
+   "MaximumNetworkInterfaces": 16,
    "NetworkCards": [
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 0,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
     },
     {
      "BaselineBandwidthInGbps": 200.0,
-     "MaximumNetworkInterfaces": 7,
+     "MaximumNetworkInterfaces": 8,
      "NetworkCardIndex": 1,
      "NetworkPerformance": "Up to 170 Gigabit",
      "PeakBandwidthInGbps": 200.0
diff --git a/moto/emrserverless/exceptions.py b/moto/emrserverless/exceptions.py
index de7c7b924..46d779882 100644
--- a/moto/emrserverless/exceptions.py
+++ b/moto/emrserverless/exceptions.py
@@ -6,9 +6,9 @@ from moto.core.exceptions import JsonRESTError
 class ResourceNotFoundException(JsonRESTError):
     code = 400
 
-    def __init__(self, resource: str):
+    def __init__(self, resource: str, resource_type: str = "Application"):
         super().__init__(
-            "ResourceNotFoundException", f"Application {resource} does not exist"
+            "ResourceNotFoundException", f"{resource_type} {resource} does not exist"
         )
 
 
@@ -17,3 +17,10 @@ class ValidationException(JsonRESTError):
 
     def __init__(self, message: str):
         super().__init__("ValidationException", message)
+
+
+class AccessDeniedException(JsonRESTError):
+    code = 400
+
+    def __init__(self, message: str):
+        super().__init__("AccessDeniedException", message)
diff --git a/moto/emrserverless/models.py b/moto/emrserverless/models.py
index af5212210..493f08440 100644
--- a/moto/emrserverless/models.py
+++ b/moto/emrserverless/models.py
@@ -3,25 +3,31 @@
 import inspect
 import re
 from datetime import datetime
-from typing import Any, Dict, Iterator, List, Optional, Tuple
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.core.utils import iso_8601_datetime_without_milliseconds
 from moto.emrcontainers.utils import get_partition, paginated_list
 
-from .exceptions import ResourceNotFoundException, ValidationException
+from .exceptions import (
+    AccessDeniedException,
+    ResourceNotFoundException,
+    ValidationException,
+)
 from .utils import (
     default_auto_start_configuration,
     default_auto_stop_configuration,
     random_appplication_id,
+    random_job_id,
 )
 
-APPLICATION_ARN_TEMPLATE = "arn:{partition}:emr-containers:{region}:{account_id}:/applications/{application_id}"
+APPLICATION_ARN_TEMPLATE = "arn:{partition}:emr-serverless:{region}:{account_id}:/applications/{application_id}"
+JOB_RUN_ARN_TEMPLATE = "arn:{partition}:emr-serverless:{region}:{account_id}:/applications/{application_id}/jobruns/{job_run_id}"
 
 # Defaults used for creating an EMR Serverless application
 APPLICATION_STATUS = "STARTED"
-JOB_STATUS = "RUNNING"
+JOB_STATUS = "SUCCESS"
 
 
 class FakeApplication(BaseModel):
@@ -131,14 +137,118 @@ class FakeApplication(BaseModel):
         return response
 
 
+class FakeJobRun(BaseModel):
+    def __init__(
+        self,
+        application_id: str,
+        client_token: str,
+        execution_role_arn: str,
+        account_id: str,
+        region_name: str,
+        release_label: str,
+        application_type: str,
+        job_driver: Optional[Dict[str, Dict[str, Union[str, List[str]]]]],
+        configuration_overrides: Optional[Dict[str, Union[List[Any], Dict[str, Any]]]],
+        tags: Optional[Dict[str, str]],
+        network_configuration: Optional[Dict[str, List[str]]],
+        execution_timeout_minutes: Optional[int],
+        name: Optional[str],
+    ):
+        self.name = name
+        self.application_id = application_id
+        self.client_token = client_token
+        self.execution_role_arn = execution_role_arn
+        self.job_driver = job_driver
+        self.configuration_overrides = configuration_overrides
+        self.network_configuration = network_configuration
+        self.execution_timeout_minutes = execution_timeout_minutes or 720
+
+        # Service-generated-parameters
+        self.id = random_job_id()
+
+        self.arn = JOB_RUN_ARN_TEMPLATE.format(
+            partition="aws",
+            account_id=account_id,
+            application_id=self.application_id,
+            region=region_name,
+            job_run_id=self.id,
+        )
+
+        self.release_label = release_label
+        self.application_type = application_type
+
+        self.state = JOB_STATUS
+        self.state_details: Optional[str] = None
+
+        self.created_by: Optional[str] = None
+
+        self.created_at: str = iso_8601_datetime_without_milliseconds(
+            datetime.today().replace(hour=0, minute=0, second=0, microsecond=0)
+        )
+        self.updated_at: str = self.created_at
+
+        self.total_execution_duration_seconds: int = 0
+        self.billed_resource_utilization: Dict[str, float] = {
+            "vCPUHour": 0.0,
+            "memoryGBHour": 0.0,
+            "storageGBHour": 0.0,
+        }
+
+        self.tags = tags
+
+    def to_dict(self, caller_methods_type: str) -> Dict[str, Any]:
+        # The response structure is different for get/update and list
+        if caller_methods_type in ["get", "update"]:
+            response = {
+                "applicationId": self.application_id,
+                "jobRunId": self.id,
+                "name": self.name,
+                "arn": self.arn,
+                "createdBy": self.created_by,
+                "createdAt": self.created_at,
+                "updatedAt": self.updated_at,
+                "executionRole": self.execution_role_arn,
+                "state": self.state,
+                "stateDetails": self.state_details,
+                "releaseLabel": self.release_label,
+                "configurationOverrides": self.configuration_overrides,
+                "jobDriver": self.job_driver,
+                "tags": self.tags,
+                "networkConfiguration": self.network_configuration,
+                "totalExecutionDurationSeconds": self.total_execution_duration_seconds,
+                "executionTimeoutMinutes": self.execution_timeout_minutes,
+                "billedResourceUtilization": self.billed_resource_utilization,
+            }
+        else:
+            response = {
+                "applicationId": self.application_id,
+                "id": self.id,
+                "name": self.name,
+                "arn": self.arn,
+                "createdBy": self.created_by,
+                "createdAt": self.created_at,
+                "updatedAt": self.updated_at,
+                "executionRole": self.execution_role_arn,
+                "state": self.state,
+                "stateDetails": self.state_details,
+                "releaseLabel": self.release_label,
+                "type": self.application_type,
+            }
+
+        return response
+
+
 class EMRServerlessBackend(BaseBackend):
     """Implementation of EMRServerless APIs."""
 
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.region_name = region_name
-        self.applications: Dict[str, FakeApplication] = dict()
         self.partition = get_partition(region_name)
+        self.applications: Dict[str, FakeApplication] = dict()
+        self.job_runs: Dict[str, List[FakeJobRun]] = (
+            dict()
+        )  # {application_id: [job_run1, job_run2]}
 
     def create_application(
         self,
@@ -267,5 +377,114 @@ class EMRServerlessBackend(BaseBackend):
 
         return self.applications[application_id].to_dict()
 
+    def start_job_run(
+        self,
+        application_id: str,
+        client_token: str,
+        execution_role_arn: str,
+        job_driver: Optional[Dict[str, Dict[str, Union[str, List[str]]]]],
+        configuration_overrides: Optional[Dict[str, Union[List[Any], Dict[str, Any]]]],
+        tags: Optional[Dict[str, str]],
+        execution_timeout_minutes: Optional[int],
+        name: Optional[str],
+    ) -> FakeJobRun:
+        role_account_id = execution_role_arn.split(":")[4]
+        if role_account_id != self.account_id:
+            raise AccessDeniedException("Cross-account pass role is not allowed.")
+
+        if execution_timeout_minutes and execution_timeout_minutes < 5:
+            raise ValidationException("RunTimeout must be at least 5 minutes.")
+
+        application_resp = self.get_application(application_id)
+        job_run = FakeJobRun(
+            application_id=application_id,
+            client_token=client_token,
+            execution_role_arn=execution_role_arn,
+            account_id=self.account_id,
+            region_name=self.region_name,
+            release_label=application_resp["releaseLabel"],
+            application_type=application_resp["type"],
+            job_driver=job_driver,
+            configuration_overrides=configuration_overrides,
+            tags=tags,
+            network_configuration=application_resp.get("networkConfiguration"),
+            execution_timeout_minutes=execution_timeout_minutes,
+            name=name,
+        )
+
+        if application_resp["state"] == "TERMINATED":
+            raise ValidationException(
+                f"Application {application_id} is terminated. Cannot start job run."
+            )
+
+        if application_id not in self.job_runs:
+            self.job_runs[application_id] = []
+        self.job_runs[application_id].append(job_run)
+
+        return job_run
+
+    def get_job_run(self, application_id: str, job_run_id: str) -> FakeJobRun:
+        if application_id not in self.job_runs.keys():
+            raise ResourceNotFoundException(application_id, "Application")
+        job_run_ids = [job_run.id for job_run in self.job_runs[application_id]]
+        if job_run_id not in job_run_ids:
+            raise ResourceNotFoundException(job_run_id, "JobRun")
+
+        filtered_job_runs = [
+            job_run
+            for job_run in self.job_runs[application_id]
+            if job_run.id == job_run_id
+        ]
+        assert len(filtered_job_runs) == 1
+        job_run: FakeJobRun = filtered_job_runs[0]
+
+        return job_run
+
+    def cancel_job_run(self, application_id: str, job_run_id: str) -> Tuple[str, str]:
+        # implement here
+        if application_id not in self.job_runs.keys():
+            raise ResourceNotFoundException(application_id, "Application")
+        job_run_ids = [job_run.id for job_run in self.job_runs[application_id]]
+        if job_run_id not in job_run_ids:
+            raise ResourceNotFoundException(job_run_id, "JobRun")
+
+        self.job_runs[application_id][job_run_ids.index(job_run_id)].state = "CANCELLED"
+
+        return application_id, job_run_id
+
+    def list_job_runs(
+        self,
+        application_id: str,
+        max_results: int,
+        next_token: Optional[str],
+        created_at_after: Optional[str],
+        created_at_before: Optional[str],
+        states: Optional[List[str]],
+    ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
+        if application_id not in self.job_runs.keys():
+            raise ResourceNotFoundException(application_id, "Application")
+        job_runs = self.job_runs[application_id]
+        if states:
+            job_runs = [job_run for job_run in job_runs if job_run.state in states]
+        if created_at_after:
+            job_runs = [
+                job_run
+                for job_run in job_runs
+                if datetime.strptime(job_run.created_at, "%Y-%m-%dT%H:%M:%SZ")
+                > datetime.strptime(created_at_after, "%Y-%m-%dT%H:%M:%SZ")
+            ]
+        if created_at_before:
+            job_runs = [
+                job_run
+                for job_run in job_runs
+                if datetime.strptime(job_run.created_at, "%Y-%m-%dT%H:%M:%SZ")
+                < datetime.strptime(created_at_before, "%Y-%m-%dT%H:%M:%SZ")
+            ]
+
+        job_run_dicts = [job_run.to_dict("list") for job_run in job_runs]
+
+        sort_key = "createdAt"
+        return paginated_list(job_run_dicts, sort_key, max_results, next_token)
+
 
 emrserverless_backends = BackendDict(EMRServerlessBackend, "emr-serverless")
diff --git a/moto/emrserverless/responses.py b/moto/emrserverless/responses.py
index b83fc121e..409927f1a 100644
--- a/moto/emrserverless/responses.py
+++ b/moto/emrserverless/responses.py
@@ -11,21 +11,21 @@ DEFAULT_MAX_RESULTS = 100
 DEFAULT_NEXT_TOKEN = ""
 
 """
-These are the available methos:
+These are the available methods:
     can_paginate()
-    cancel_job_run()
+    cancel_job_run() -> DONE
     close()
     create_application() -> DONE
     delete_application() -> DONE
     get_application() -> DONE
-    get_job_run()
+    get_job_run() -> DONE
     get_paginator()
     get_waiter()
     list_applications() -> DONE
-    list_job_runs()
+    list_job_runs() -> DONE
     list_tags_for_resource()
     start_application() -> DONE
-    start_job_run()
+    start_job_run() -> DONE
     stop_application() -> DONE
     tag_resource()
     untag_resource()
@@ -128,3 +128,77 @@ class EMRServerlessResponse(BaseResponse):
         )
         response = {"application": application}
         return 200, {}, json.dumps(response)
+
+    def start_job_run(self) -> TYPE_RESPONSE:
+        # params = self._get_params()
+        application_id = self._get_param(
+            "applicationId"
+        )  # params["application_id"] #.get()
+        client_token = self._get_param("clientToken")
+        execution_role_arn = self._get_param("executionRoleArn")
+        job_driver = self._get_param("jobDriver")
+        configuration_overrides = self._get_param("configurationOverrides")
+        tags = self._get_param("tags")
+        execution_timeout_minutes = self._get_int_param("executionTimeoutMinutes")
+        name = self._get_param("name")
+        job_run = self.emrserverless_backend.start_job_run(
+            application_id=application_id,
+            client_token=client_token,
+            execution_role_arn=execution_role_arn,
+            job_driver=job_driver,
+            configuration_overrides=configuration_overrides,
+            tags=tags,
+            execution_timeout_minutes=execution_timeout_minutes,
+            name=name,
+        )
+        return (
+            200,
+            {},
+            json.dumps(
+                {
+                    "applicationId": job_run.application_id,
+                    "jobRunId": job_run.id,
+                    "arn": job_run.arn,
+                }
+            ),
+        )
+
+    def get_job_run(self) -> TYPE_RESPONSE:
+        application_id = self._get_param("applicationId")
+        job_run_id = self._get_param("jobRunId")
+        job_run = self.emrserverless_backend.get_job_run(
+            application_id=application_id, job_run_id=job_run_id
+        )
+        # TODO: adjust response
+        return 200, {}, json.dumps({"jobRun": job_run.to_dict("get")})
+
+    def cancel_job_run(self) -> TYPE_RESPONSE:
+        application_id = self._get_param("applicationId")
+        job_run_id = self._get_param("jobRunId")
+        application_id, job_run_id = self.emrserverless_backend.cancel_job_run(
+            application_id=application_id,
+            job_run_id=job_run_id,
+        )
+
+        return (
+            200,
+            {},
+            json.dumps(dict(applicationId=application_id, jobRunId=job_run_id)),
+        )
+
+    def list_job_runs(self) -> TYPE_RESPONSE:
+        application_id = self._get_param("applicationId")
+        next_token = self._get_param("nextToken", DEFAULT_NEXT_TOKEN)
+        max_results = self._get_int_param("maxResults", DEFAULT_MAX_RESULTS)
+        created_at_after = self._get_param("createdAtAfter")
+        created_at_before = self._get_param("createdAtBefore")
+        states = self.querystring.get("states", [])
+        job_runs, next_token = self.emrserverless_backend.list_job_runs(
+            application_id=application_id,
+            next_token=next_token,
+            max_results=max_results,
+            created_at_after=created_at_after,
+            created_at_before=created_at_before,
+            states=states,
+        )
+        return 200, {}, json.dumps({"jobRuns": job_runs, "nextToken": next_token})
diff --git a/moto/ssm/resources/regions.json b/moto/ssm/resources/regions.json
index dda6a1fb1..4c2dea712 100644
--- a/moto/ssm/resources/regions.json
+++ b/moto/ssm/resources/regions.json
@@ -10451,6 +10451,9 @@
                   "Value": "HTTPS"
                 }
               },
+              "bedrock": {
+                "Value": "bedrock"
+              },
               "budgets": {
                 "Value": "budgets",
                 "endpoint": {
@@ -24438,6 +24441,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "controltower": {
+                "Value": "controltower",
+                "endpoint": {
+                  "Value": "controltower.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "datasync": {
                 "Value": "datasync",
                 "endpoint": {
@@ -24630,6 +24642,60 @@
                   "Value": "HTTPS"
                 }
               },
+              "fms": {
+                "Value": "fms",
+                "endpoint": {
+                  "Value": "fms.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "fsx": {
+                "Value": "fsx",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "fsx-lustre": {
+                "Value": "fsx-lustre",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "fsx-ontap": {
+                "Value": "fsx-ontap",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "fsx-openzfs": {
+                "Value": "fsx-openzfs",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "fsx-windows": {
+                "Value": "fsx-windows",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "globalaccelerator": {
                 "Value": "globalaccelerator",
                 "endpoint": {
@@ -25068,6 +25134,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "transfer": {
+                "Value": "transfer",
+                "endpoint": {
+                  "Value": "transfer.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "transitgateway": {
                 "Value": "transitgateway",
                 "endpoint": {
@@ -25101,6 +25176,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "waf": {
+                "Value": "waf",
+                "endpoint": {
+                  "Value": "wafv2.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "xray": {
                 "Value": "xray",
                 "endpoint": {
@@ -37263,6 +37347,9 @@
                   "Value": "HTTPS"
                 }
               },
+              "bedrock": {
+                "Value": "bedrock"
+              },
               "budgets": {
                 "Value": "budgets",
                 "endpoint": {
@@ -45446,6 +45533,12 @@
                   "Value": "HTTPS"
                 }
               },
+              "snowball": {
+                "Value": "snowball",
+                "endpoint": {
+                  "Value": "snowball.il-central-1.amazonaws.com"
+                }
+              },
               "snowcone": {
                 "Value": "snowcone"
               },
@@ -53156,6 +53249,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "qbusiness": {
+                "Value": "qbusiness",
+                "endpoint": {
+                  "Value": "qbusiness.us-east-1.api.aws"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "qldb": {
                 "Value": "qldb",
                 "endpoint": {
@@ -53597,9 +53699,6 @@
               "snowcone": {
                 "Value": "snowcone"
               },
-              "snowmobile": {
-                "Value": "snowmobile"
-              },
               "sns": {
                 "Value": "sns",
                 "endpoint": {
@@ -56276,9 +56375,6 @@
               "snowcone": {
                 "Value": "snowcone"
               },
-              "snowmobile": {
-                "Value": "snowmobile"
-              },
               "sns": {
                 "Value": "sns",
                 "endpoint": {
@@ -57813,9 +57909,6 @@
                   "Value": "snowball.us-gov-east-1.amazonaws.com"
                 }
               },
-              "snowmobile": {
-                "Value": "snowmobile"
-              },
               "sns": {
                 "Value": "sns",
                 "endpoint": {
@@ -59509,9 +59602,6 @@
                   "Value": "snowball.us-gov-west-1.amazonaws.com"
                 }
               },
-              "snowmobile": {
-                "Value": "snowmobile"
-              },
               "sns": {
                 "Value": "sns",
                 "endpoint": {
@@ -61562,9 +61652,6 @@
               "snowcone": {
                 "Value": "snowcone"
               },
-              "snowmobile": {
-                "Value": "snowmobile"
-              },
               "sns": {
                 "Value": "sns",
                 "endpoint": {
@@ -61830,6 +61917,9 @@
               "usw2-den1-az1": {
                 "Value": "usw2-den1-az1"
               },
+              "usw2-hnl1-az1": {
+                "Value": "usw2-hnl1-az1"
+              },
               "usw2-las1-az1": {
                 "Value": "usw2-las1-az1"
               },
@@ -64037,6 +64127,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "qbusiness": {
+                "Value": "qbusiness",
+                "endpoint": {
+                  "Value": "qbusiness.us-west-2.api.aws"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "qldb": {
                 "Value": "qldb",
                 "endpoint": {
@@ -64499,9 +64598,6 @@
               "snowcone": {
                 "Value": "snowcone"
               },
-              "snowmobile": {
-                "Value": "snowmobile"
-              },
               "sns": {
                 "Value": "sns",
                 "endpoint": {
diff --git a/moto/ssm/resources/services.json b/moto/ssm/resources/services.json
index 86565ab15..458a76cda 100644
--- a/moto/ssm/resources/services.json
+++ b/moto/ssm/resources/services.json
@@ -7842,6 +7842,9 @@
               "ap-northeast-1": {
                 "Value": "ap-northeast-1"
               },
+              "ap-south-1": {
+                "Value": "ap-south-1"
+              },
               "ap-southeast-1": {
                 "Value": "ap-southeast-1"
               },
@@ -7851,6 +7854,9 @@
               "eu-central-1": {
                 "Value": "eu-central-1"
               },
+              "eu-west-1": {
+                "Value": "eu-west-1"
+              },
               "eu-west-3": {
                 "Value": "eu-west-3"
               },
@@ -15738,6 +15744,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "controltower.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "eu-central-1": {
                 "Value": "eu-central-1",
                 "endpoint": {
@@ -26005,6 +26020,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "fms.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -26759,6 +26783,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -27058,6 +27091,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -27357,6 +27399,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -27647,6 +27698,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -27946,6 +28006,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "fsx.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -33720,6 +33789,9 @@
             "longName": {
               "Value": "AWS IoT FleetWise"
             },
+            "marketingHomeURL": {
+              "Value": "https://aws.amazon.com/iot-fleetwise/"
+            },
             "regions": {
               "eu-central-1": {
                 "Value": "eu-central-1",
@@ -48068,6 +48140,32 @@
               }
             }
           },
+          "qbusiness": {
+            "Value": "qbusiness",
+            "longName": {
+              "Value": "QBusiness"
+            },
+            "regions": {
+              "us-east-1": {
+                "Value": "us-east-1",
+                "endpoint": {
+                  "Value": "qbusiness.us-east-1.api.aws"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "us-west-2": {
+                "Value": "us-west-2",
+                "endpoint": {
+                  "Value": "qbusiness.us-west-2.api.aws"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              }
+            }
+          },
           "qldb": {
             "Value": "qldb",
             "longName": {
@@ -59229,6 +59327,12 @@
                   "Value": "snowball.eu-west-3.amazonaws.com"
                 }
               },
+              "il-central-1": {
+                "Value": "il-central-1",
+                "endpoint": {
+                  "Value": "snowball.il-central-1.amazonaws.com"
+                }
+              },
               "me-central-1": {
                 "Value": "me-central-1",
                 "endpoint": {
@@ -59335,35 +59439,6 @@
               }
             }
           },
-          "snowmobile": {
-            "Value": "snowmobile",
-            "longName": {
-              "Value": "AWS Snowmobile"
-            },
-            "marketingHomeURL": {
-              "Value": "https://aws.amazon.com/snowmobile/"
-            },
-            "regions": {
-              "us-east-1": {
-                "Value": "us-east-1"
-              },
-              "us-east-2": {
-                "Value": "us-east-2"
-              },
-              "us-gov-east-1": {
-                "Value": "us-gov-east-1"
-              },
-              "us-gov-west-1": {
-                "Value": "us-gov-west-1"
-              },
-              "us-west-1": {
-                "Value": "us-west-1"
-              },
-              "us-west-2": {
-                "Value": "us-west-2"
-              }
-            }
-          },
           "sns": {
             "Value": "sns",
             "longName": {
@@ -63798,6 +63873,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "transfer.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -65848,6 +65932,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "wafv2.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
diff --git a/moto/wafv2/exceptions.py b/moto/wafv2/exceptions.py
index 218816d94..6a9810eeb 100644
--- a/moto/wafv2/exceptions.py
+++ b/moto/wafv2/exceptions.py
@@ -19,3 +19,11 @@ class WAFNonexistentItemException(WAFv2ClientError):
             "WAFNonexistentItemException",
             "AWS WAF couldn’t perform the operation because your resource doesn’t exist.",
         )
+
+
+class WAFOptimisticLockException(WAFv2ClientError):
+    def __init__(self) -> None:
+        super().__init__(
+            "WAFOptimisticLockException",
+            "AWS WAF couldn’t save your changes because someone changed the resource after you started to edit it. Reapply your changes.",
+        )
diff --git a/moto/wafv2/models.py b/moto/wafv2/models.py
index f3b24076e..6bead554e 100644
--- a/moto/wafv2/models.py
+++ b/moto/wafv2/models.py
@@ -8,8 +8,12 @@ from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
 
-from .exceptions import WAFNonexistentItemException, WAFV2DuplicateItemException
-from .utils import make_arn_for_wacl
+from .exceptions import (
+    WAFNonexistentItemException,
+    WAFOptimisticLockException,
+    WAFV2DuplicateItemException,
+)
+from .utils import make_arn_for_ip_set, make_arn_for_wacl
 
 if TYPE_CHECKING:
     from moto.apigateway.models import Stage
@@ -79,6 +83,75 @@ class FakeWebACL(BaseModel):
         }
 
 
+class FakeIPSet(BaseModel):
+    """
+    https://docs.aws.amazon.com/waf/latest/APIReference/API_IPSet.html
+    """
+
+    def __init__(
+        self,
+        arn: str,
+        ip_set_id: str,
+        ip_address_version: str,
+        addresses: List[str],
+        name: str,
+        description: str,
+        scope: str,
+    ):
+        self.name = name
+        self.ip_set_id = ip_set_id
+        self.arn = arn
+        self.addresses = addresses
+        self.description = description
+        self.ip_address_version = ip_address_version
+        self.scope = scope
+
+        self.lock_token = str(mock_random.uuid4())[0:6]
+
+    def update(self, description: Optional[str], addresses: List[str]) -> None:
+        if description is not None:
+            self.description = description
+        self.addresses = addresses
+
+        self.lock_token = str(mock_random.uuid4())[0:6]
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "Name": self.name,
+            "Id": self.ip_set_id,
+            "ARN": self.arn,
+            "Description": self.description,
+            "IPAddressVersion": self.ip_address_version,
+            "Addresses": self.addresses,
+            "LockToken": self.lock_token,
+        }
+
+
+class FakeLoggingConfiguration:
+    def __init__(
+        self,
+        arn: str,
+        log_destination_configs: List[str],
+        redacted_fields: Optional[Dict[str, Any]],
+        managed_gy_firewall_manager: Optional[bool],
+        logging_filter: Optional[Dict[str, Any]],
+    ):
+        self.arn = arn
+        self.log_destination_configs = log_destination_configs
+        self.redacted_fields = redacted_fields
+        self.managed_by_firewall_manager = managed_gy_firewall_manager or False
+        self.logging_filter = logging_filter
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "ResourceArn": self.arn,
+            "LogDestinationConfigs": self.log_destination_configs,
+            "RedactedFields": self.redacted_fields,
+            "ManagedByFirewallManager": self.managed_by_firewall_manager,
+            "LoggingFilter": self.logging_filter,
+        }
+
+
 class WAFV2Backend(BaseBackend):
     """
     https://docs.aws.amazon.com/waf/latest/APIReference/API_Operations_AWS_WAFV2.html
@@ -87,6 +160,8 @@ class WAFV2Backend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.wacls: Dict[str, FakeWebACL] = OrderedDict()
+        self.ip_sets: Dict[str, FakeIPSet] = OrderedDict()
+        self.logging_configurations: Dict[str, FakeLoggingConfiguration] = OrderedDict()
         self.tagging_service = TaggingService()
         # TODO: self.load_balancers = OrderedDict()
 
@@ -210,6 +285,140 @@ class WAFV2Backend(BaseBackend):
         acl.update(default_action, rules, description, visibility_config)
         return acl.lock_token
 
+    def create_ip_set(
+        self,
+        name: str,
+        scope: str,
+        description: str,
+        ip_address_version: str,
+        addresses: List[str],
+        tags: List[Dict[str, str]],
+    ) -> FakeIPSet:
+        ip_set_id = str(mock_random.uuid4())
+        arn = make_arn_for_ip_set(
+            name=name,
+            account_id=self.account_id,
+            region_name=self.region_name,
+            _id=ip_set_id,
+            scope=scope,
+        )
+
+        new_ip_set = FakeIPSet(
+            arn,
+            ip_set_id,
+            ip_address_version,
+            addresses,
+            name,
+            description,
+            scope,
+        )
+        self.ip_sets[arn] = new_ip_set
+        self.tag_resource(arn, tags)
+        return new_ip_set
+
+    def delete_ip_set(self, name: str, scope: str, _id: str, lock_token: str) -> None:
+        arn = make_arn_for_ip_set(
+            name=name,
+            account_id=self.account_id,
+            region_name=self.region_name,
+            _id=_id,
+            scope=scope,
+        )
+
+        if arn not in self.ip_sets:
+            raise WAFNonexistentItemException()
+
+        if lock_token != self.ip_sets[arn].lock_token:
+            raise WAFOptimisticLockException()
+
+        self.ip_sets.pop(arn)
+
+    def list_ip_sets(self, scope: str) -> List[FakeIPSet]:
+        ip_sets = [
+            ip_set for arn, ip_set in self.ip_sets.items() if ip_set.scope == scope
+        ]
+        return ip_sets
+
+    def get_ip_set(self, name: str, scope: str, _id: str) -> FakeIPSet:
+        arn = make_arn_for_ip_set(
+            name=name,
+            account_id=self.account_id,
+            region_name=self.region_name,
+            _id=_id,
+            scope=scope,
+        )
+        if arn not in self.ip_sets:
+            raise WAFNonexistentItemException()
+
+        return self.ip_sets[arn]
+
+    def update_ip_set(
+        self,
+        name: str,
+        scope: str,
+        _id: str,
+        description: Optional[str],
+        addresses: List[str],
+        lock_token: str,
+    ) -> FakeIPSet:
+        arn = make_arn_for_ip_set(
+            name=name,
+            account_id=self.account_id,
+            region_name=self.region_name,
+            _id=_id,
+            scope=scope,
+        )
+
+        if not (ip_set := self.ip_sets.get(arn)):
+            raise WAFNonexistentItemException()
+
+        if ip_set.lock_token != lock_token:
+            raise WAFOptimisticLockException()
+
+        ip_set.update(description, addresses)
+
+        return ip_set
+
+    def put_logging_configuration(
+        self,
+        arn: str,
+        log_destination_configs: List[str],
+        redacted_fields: Optional[Dict[str, Any]],
+        managed_gy_firewall_manager: bool,
+        logging_filter: Dict[str, Any],
+    ) -> FakeLoggingConfiguration:
+        logging_configuration = FakeLoggingConfiguration(
+            arn,
+            log_destination_configs,
+            redacted_fields,
+            managed_gy_firewall_manager,
+            logging_filter,
+        )
+        self.logging_configurations[arn] = logging_configuration
+        return logging_configuration
+
+    def delete_logging_configuration(self, arn: str) -> None:
+        if not self.logging_configurations.get(arn):
+            raise WAFNonexistentItemException()
+        self.logging_configurations.pop(arn)
+
+    def get_logging_configuration(self, arn: str) -> FakeLoggingConfiguration:
+        if not (logging_configuration := self.logging_configurations.get(arn)):
+            raise WAFNonexistentItemException()
+        return logging_configuration
+
+    def list_logging_configurations(self, scope: str) -> List[FakeLoggingConfiguration]:
+        if scope == "CLOUDFRONT":
+            scope = "global"
+        else:
+            scope = self.region_name
+
+        return [
+            logging_configuration
+            for arn, logging_configuration in self.logging_configurations.items()
+            if f":{scope}:" in arn
+        ]
+
 
 wafv2_backends = BackendDict(
     WAFV2Backend, "waf-regional", additional_regions=["global"]
diff --git a/moto/wafv2/responses.py b/moto/wafv2/responses.py
index ea81e45c7..68b496059 100644
--- a/moto/wafv2/responses.py
+++ b/moto/wafv2/responses.py
@@ -3,6 +3,7 @@ import json
 from moto.core.common_types import TYPE_RESPONSE
 from moto.core.responses import BaseResponse
 
+from ..moto_api._internal import mock_random
 from .models import GLOBAL_REGION, WAFV2Backend, wafv2_backends
 
 
@@ -140,6 +141,181 @@ class WAFV2Response(BaseResponse):
         )
         return 200, {}, json.dumps({"NextLockToken": lock_token})
 
+    def create_ip_set(self) -> TYPE_RESPONSE:
+        body = json.loads(self.body)
+
+        name = body.get("Name")
+        scope = body.get("Scope")
+
+        if scope == "CLOUDFRONT":
+            self.region = GLOBAL_REGION
+
+        description = body.get("Description")
+        ip_address_version = body.get("IPAddressVersion")
+        addresses = body.get("Addresses")
+        tags = body.get("Tags")
+
+        ip_set = self.wafv2_backend.create_ip_set(
+            name, scope, description, ip_address_version, addresses, tags
+        )
+        return (
+            200,
+            {},
+            json.dumps(
+                {
+                    "Summary": {
+                        "Name": ip_set.name,
+                        "Id": ip_set.ip_set_id,
+                        "Description": ip_set.description,
+                        "LockToken": ip_set.lock_token,
+                        "ARN": ip_set.arn,
+                    }
+                }
+            ),
+        )
+
+    def delete_ip_set(self) -> TYPE_RESPONSE:
+        body = json.loads(self.body)
+
+        name = body.get("Name")
+        scope = body.get("Scope")
+        if scope == "CLOUDFRONT":
+            self.region = GLOBAL_REGION
+
+        _id = body.get("Id")
+        lock_token = body.get("LockToken")
+
+        self.wafv2_backend.delete_ip_set(name, scope, _id, lock_token)
+
+        return 200, {}, "{}"
+
+    def list_ip_sets(self) -> TYPE_RESPONSE:
+        scope = self._get_param("Scope")
+        if scope == "CLOUDFRONT":
+            self.region = GLOBAL_REGION
+        ip_sets = self.wafv2_backend.list_ip_sets(scope)
+
+        formatted_ip_sets = [
+            {
+                "Name": ip_set.name,
+                "Id": ip_set.ip_set_id,
+                "Description": ip_set.description,
+                "LockToken": ip_set.lock_token,
+                "ARN": ip_set.arn,
+            }
+            for ip_set in ip_sets
+        ]
+
+        return (
+            200,
+            {},
+            json.dumps(
+                {"NextMarker": str(mock_random.uuid4()), "IPSets": formatted_ip_sets}
+            ),
+        )
+
+    def get_ip_set(self) -> TYPE_RESPONSE:
+        scope = self._get_param("Scope")
+        if scope == "CLOUDFRONT":
+            self.region = GLOBAL_REGION
+
+        ip_set = self.wafv2_backend.get_ip_set(
+            name=self._get_param("Name"), scope=scope, _id=self._get_param("Id")
+        )
+
+        dict_ip = ip_set.to_dict()
+        lock_token = dict_ip.pop("LockToken")
+        return 200, {}, json.dumps({"IPSet": dict_ip, "LockToken": lock_token})
+
+    def update_ip_set(self) -> TYPE_RESPONSE:
+        body = json.loads(self.body)
+
+        name = body.get("Name")
+        scope = body.get("Scope")
+        _id = body.get("Id")
+
+        if scope == "CLOUDFRONT":
+            self.region = GLOBAL_REGION
+
+        description = body.get("Description")
+        addresses = body.get("Addresses")
+        lock_token = body.get("LockToken")
+        updated_ip_set = self.wafv2_backend.update_ip_set(
+            name, scope, _id, description, addresses, lock_token
+        )
+
+        return 200, {}, json.dumps({"NextLockToken": updated_ip_set.lock_token})
+
+    def put_logging_configuration(self) -> TYPE_RESPONSE:
+        body = json.loads(self.body)
+        logging_configuration_parameter = body["LoggingConfiguration"]
+        resource_arn = logging_configuration_parameter["ResourceArn"]
+        log_destination_configs = logging_configuration_parameter[
+            "LogDestinationConfigs"
+        ]
+        redacted_fields = logging_configuration_parameter.get("RedactedFields")
+        managed_by_firewall_manager = logging_configuration_parameter.get(
+            "ManagedByFirewallManager"
+        )
+        logging_filter = logging_configuration_parameter.get("LoggingFilter")
+        logging_configuration = self.wafv2_backend.put_logging_configuration(
+            resource_arn,
+            log_destination_configs,
+            redacted_fields,
+            managed_by_firewall_manager,
+            logging_filter,
+        )
+        return (
+            200,
+            {},
+            json.dumps(
+                {
+                    "LoggingConfiguration": {
+                        k: v
+                        for k, v in logging_configuration.to_dict().items()
+                        if v is not None
+                    }
+                }
+            ),
+        )
+
+    def get_logging_configuration(self) -> TYPE_RESPONSE:
+        body = json.loads(self.body)
+        resource_arn = body["ResourceArn"]
+        logging_configuration = self.wafv2_backend.get_logging_configuration(
+            resource_arn
+        )
+        return (
+            200,
+            {},
+            json.dumps(
+                {
+                    "LoggingConfiguration": {
+                        k: v
+                        for k, v in logging_configuration.to_dict().items()
+                        if v is not None
+                    }
+                }
+            ),
+        )
+
+    def list_logging_configurations(self) -> TYPE_RESPONSE:
+        body = json.loads(self.body)
+        scope = body.get("Scope")
+        log_configs = self.wafv2_backend.list_logging_configurations(scope)
+
+        formatted = [
+            {k: v for k, v in config.to_dict().items() if v is not None}
+            for config in log_configs
+        ]
+        return 200, {}, json.dumps({"LoggingConfigurations": formatted})
+
+    def delete_logging_configuration(self) -> TYPE_RESPONSE:
+        body = json.loads(self.body)
+        resource_arn = body["ResourceArn"]
+        self.wafv2_backend.delete_logging_configuration(resource_arn)
+        return 200, {}, "{}"
+
 
 # notes about region and scope
 # --scope = CLOUDFRONT is ALWAYS us-east-1 (but we use "global" instead to differentiate between REGIONAL us-east-1)
diff --git a/moto/wafv2/utils.py b/moto/wafv2/utils.py
index c746ff614..a893ec28a 100644
--- a/moto/wafv2/utils.py
+++ b/moto/wafv2/utils.py
@@ -2,9 +2,27 @@ def make_arn_for_wacl(
     name: str, account_id: str, region_name: str, wacl_id: str, scope: str
 ) -> str:
     """https://docs.aws.amazon.com/waf/latest/developerguide/how-aws-waf-works.html - explains --scope (cloudfront vs regional)"""
+    return make_arn(name, account_id, region_name, wacl_id, scope, "webacl")
 
+
+def make_arn_for_ip_set(
+    name: str, account_id: str, region_name: str, _id: str, scope: str
+) -> str:
+    return make_arn(name, account_id, region_name, _id, scope, "ipset")
+
+
+def make_arn_for_logging_configuration(
+    name: str, account_id: str, region_name: str, _id: str, scope: str
+) -> str:
+    return make_arn(name, account_id, region_name, _id, scope, "logingconfiguration")
+
+
+def make_arn(
+    name: str, account_id: str, region_name: str, _id: str, scope: str, resource: str
+) -> str:
     if scope == "REGIONAL":
         scope = "regional"
     elif scope == "CLOUDFRONT":
         scope = "global"
-    return f"arn:aws:wafv2:{region_name}:{account_id}:{scope}/webacl/{name}/{wacl_id}"
+
+    return f"arn:aws:wafv2:{region_name}:{account_id}:{scope}/{resource}/{name}/{_id}"
diff --git a/moto/workspaces/models.py b/moto/workspaces/models.py
index 3b8e70861..86cdb5338 100644
--- a/moto/workspaces/models.py
+++ b/moto/workspaces/models.py
@@ -58,7 +58,8 @@ class Workspace(BaseModel):
         self.modification_states: List[
             Dict[str, str]
         ] = []  # modify_workspace_properties
-        self.related_workspaces: List[Dict[str, str]] = []  # create_standy_workspace
+        # create_standy_workspace
+        self.related_workspaces: List[Dict[str, str]] = []
         self.data_replication_settings: Dict[str, Any] = {}
         # The properties of the standby WorkSpace related to related_workspaces
         self.standby_workspaces_properties: List[Dict[str, Any]] = []
@@ -141,7 +142,13 @@ class WorkSpaceDirectory(BaseModel):
         self.dns_ip_addresses = directory.dns_ip_addrs
         self.customer_username = "Administrator"
         self.iam_rold_id = f"arn:aws:iam::{account_id}:role/workspaces_DefaultRole"
-        self.directory_type = directory.directory_type
+        dir_type = directory.directory_type
+        if dir_type == "ADConnector":
+            self.directory_type = "AD_CONNECTOR"
+        elif dir_type == "SimpleAD":
+            self.directory_type = "SIMPLE_AD"
+        else:
+            self.directory_type = dir_type
         self.workspace_security_group_id = security_group_id
         self.state = "REGISTERED"
         # Default values for workspace_creation_properties


Test patch:
diff --git a/.github/workflows/tests_sdk_ruby.yml b/.github/workflows/tests_sdk_ruby.yml
index fea7721c7..749ab8763 100644
--- a/.github/workflows/tests_sdk_ruby.yml
+++ b/.github/workflows/tests_sdk_ruby.yml
@@ -13,7 +13,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Set up Ruby ${{ matrix.ruby-version }}
-        uses: ruby/setup-ruby@1198b074305f9356bd56dd4b311757cc0dab2f1c
+        uses: ruby/setup-ruby@cacc9f1c0b3f4eb8a16a6bb0ed10897b43b9de49
         with:
           ruby-version: ${{ matrix.ruby-version }}
       - name: Set up Python 3.8
diff --git a/tests/test_databrew/test_databrew_recipes.py b/tests/test_databrew/test_databrew_recipes.py
index 3de39fb10..6808694fe 100644
--- a/tests/test_databrew/test_databrew_recipes.py
+++ b/tests/test_databrew/test_databrew_recipes.py
@@ -374,6 +374,12 @@ def test_publish_recipe(recipe_name):
     assert recipe["PublishedDate"] > dt_before_publish
     first_published_date = recipe["PublishedDate"]
 
+    # Latest Working should have created date == publish date
+    working_response = client.describe_recipe(
+        Name=recipe_name, RecipeVersion="LATEST_WORKING"
+    )
+    assert working_response["CreateDate"] == first_published_date
+
     # Publish the recipe a 2nd time
     publish_response = client.publish_recipe(Name=recipe_name, Description="2nd desc")
     assert publish_response["Name"] == recipe_name
@@ -383,6 +389,12 @@ def test_publish_recipe(recipe_name):
     assert recipe["RecipeVersion"] == "2.0"
     assert recipe["PublishedDate"] >= first_published_date
 
+    # Check working Created Date
+    working_response = client.describe_recipe(
+        Name=recipe_name, RecipeVersion="LATEST_WORKING"
+    )
+    assert recipe["PublishedDate"] == working_response["CreateDate"]
+
 
 @mock_aws
 def test_publish_recipe_that_does_not_exist():
diff --git a/tests/test_dynamodb/test_dynamodb_cloudformation.py b/tests/test_dynamodb/test_dynamodb_cloudformation.py
index 8cac53669..c406fe21c 100644
--- a/tests/test_dynamodb/test_dynamodb_cloudformation.py
+++ b/tests/test_dynamodb/test_dynamodb_cloudformation.py
@@ -1,4 +1,5 @@
 import json
+from copy import deepcopy
 
 import boto3
 
@@ -7,7 +8,7 @@ from moto import mock_aws
 template_create_table = {
     "AWSTemplateFormatVersion": "2010-09-09",
     "Resources": {
-        "myDynamoDBTable": {
+        "table": {
             "Type": "AWS::DynamoDB::Table",
             "Properties": {
                 "AttributeDefinitions": [
@@ -30,7 +31,56 @@ template_create_table = {
 
 
 @mock_aws
-def test_delete_stack_dynamo_template_boto3():
+def test_create_stack_pay_per_request():
+    conn = boto3.client("cloudformation", region_name="us-east-1")
+    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
+    template = deepcopy(template_create_table)
+    template["Resources"]["table"]["Properties"]["BillingMode"] = "PAY_PER_REQUEST"
+    del template["Resources"]["table"]["Properties"]["ProvisionedThroughput"]
+
+    conn.create_stack(StackName="test", TemplateBody=json.dumps(template))
+    table_desc = dynamodb_client.list_tables()
+    assert len(table_desc.get("TableNames")) == 1
+
+    table = dynamodb_client.describe_table(TableName=table_desc["TableNames"][0])[
+        "Table"
+    ]
+    assert table["BillingModeSummary"] == {"BillingMode": "PAY_PER_REQUEST"}
+
+
+@mock_aws
+def test_create_stack_with_indexes():
+    conn = boto3.client("cloudformation", region_name="us-east-1")
+    dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
+    template = deepcopy(template_create_table)
+    template["Resources"]["table"]["Properties"]["GlobalSecondaryIndexes"] = [
+        {
+            "IndexName": "gsi",
+            "KeySchema": [{"AttributeName": "gsipk", "KeyType": "S"}],
+            "Projection": {"ProjectionType": "ALL"},
+        }
+    ]
+    template["Resources"]["table"]["Properties"]["LocalSecondaryIndexes"] = [
+        {
+            "IndexName": "lsi",
+            "KeySchema": [{"AttributeName": "lsipk", "KeyType": "S"}],
+            "Projection": {"ProjectionType": "ALL"},
+        }
+    ]
+
+    conn.create_stack(StackName="test", TemplateBody=json.dumps(template))
+    table_desc = dynamodb_client.list_tables()
+    assert len(table_desc.get("TableNames")) == 1
+
+    table = dynamodb_client.describe_table(TableName=table_desc["TableNames"][0])[
+        "Table"
+    ]
+    assert len(table["GlobalSecondaryIndexes"]) == 1
+    assert len(table["LocalSecondaryIndexes"]) == 1
+
+
+@mock_aws
+def test_delete_stack_dynamo_template():
     conn = boto3.client("cloudformation", region_name="us-east-1")
     dynamodb_client = boto3.client("dynamodb", region_name="us-east-1")
 
@@ -40,6 +90,19 @@ def test_delete_stack_dynamo_template_boto3():
     table_desc = dynamodb_client.list_tables()
     assert len(table_desc.get("TableNames")) == 1
 
+    table = dynamodb_client.describe_table(TableName=table_desc["TableNames"][0])[
+        "Table"
+    ]
+    assert table["ProvisionedThroughput"] == {
+        "NumberOfDecreasesToday": 0,
+        "ReadCapacityUnits": 5,
+        "WriteCapacityUnits": 5,
+    }
+    assert table["BillingModeSummary"] == {"BillingMode": "PROVISIONED"}
+    assert table["LocalSecondaryIndexes"] == []
+    assert table["GlobalSecondaryIndexes"] == []
+    assert table["DeletionProtectionEnabled"] is False
+
     conn.delete_stack(StackName="test_stack")
     table_desc = dynamodb_client.list_tables()
     assert len(table_desc.get("TableNames")) == 0
diff --git a/tests/test_dynamodb/test_dynamodb_import_table.py b/tests/test_dynamodb/test_dynamodb_import_table.py
new file mode 100644
index 000000000..69f159f3e
--- /dev/null
+++ b/tests/test_dynamodb/test_dynamodb_import_table.py
@@ -0,0 +1,456 @@
+import gzip
+import json
+from time import sleep
+from uuid import uuid4
+
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+
+from . import dynamodb_aws_verified
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_from_missing_s3_table(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-1")
+
+    table_name = "t" + str(uuid4())[0:6]
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": f"{uuid4()}"},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "FAILED"
+    assert import_details["TableArn"].endswith(f":table/{table_name}")
+    assert import_details["FailureCode"] == "S3NoSuchBucket"
+    assert "The specified bucket does not exist" in import_details["FailureMessage"]
+
+    assert table_name not in client.list_tables()["TableNames"]
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_has_regular_validation(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-1")
+
+    table_name = "t" + str(uuid4())[0:6]
+
+    with pytest.raises(ClientError) as exc:
+        client.import_table(
+            S3BucketSource={"S3Bucket": f"{uuid4()}"},
+            InputFormat="DYNAMODB_JSON",
+            InputCompressionType="NONE",
+            TableCreationParameters={
+                "TableName": table_name,
+                "AttributeDefinitions": [
+                    {"AttributeName": "pk", "AttributeType": "S"},
+                ],
+                "KeySchema": [
+                    {"AttributeName": "pk", "KeyType": "HASH"},
+                ],
+                "BillingMode": "PROVISIONED",
+            },
+        )
+
+    err = exc.value.response["Error"]
+    assert (
+        err["Message"]
+        == "One or more parameter values were invalid: ReadCapacityUnits and WriteCapacityUnits must both be specified when BillingMode is PROVISIONED"
+    )
+
+    assert table_name not in client.list_tables()["TableNames"]
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_from_empty_s3_bucket(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-1")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 0
+    assert import_details["ImportedItemCount"] == 0
+    assert import_details["ProcessedSizeBytes"] == 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert client.scan(TableName=table_name)["Items"] == []
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_table_single_file_with_multiple_items():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    data = ""
+    for i in range(5):
+        data += (
+            json.dumps({"Item": {"pk": {"S": f"msg{i}"}, "data": {"S": f"{uuid4()}"}}})
+            + "\n"
+        )
+    for i in range(10, 15):
+        data += json.dumps(
+            {"Item": {"pk": {"S": f"msg{i}"}, "data": {"S": f"{uuid4()}"}}}
+        )
+    filename1 = "data.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=data,
+        Key=filename1,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 10
+    assert import_details["ImportedItemCount"] == 10
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+    expected = [
+        "msg0",
+        "msg1",
+        "msg10",
+        "msg11",
+        "msg12",
+        "msg13",
+        "msg14",
+        "msg2",
+        "msg3",
+        "msg4",
+    ]
+    assert (
+        sorted([i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]])
+        == expected
+    )
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_import_table_multiple_files():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    items_file1 = {"Item": {"pk": {"S": "msg1"}, "data": {"S": f"{uuid4()}"}}}
+    filename1 = "data.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file1),
+        Key=filename1,
+    )
+
+    items_file2 = {"Item": {"pk": {"S": "msg2"}, "data": {"S": f"{uuid4()}"}}}
+    filename2 = "completely_random_filename_without_extension"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file2),
+        Key=filename2,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 2
+    assert import_details["ImportedItemCount"] == 2
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert [i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]] == [
+        "msg1",
+        "msg2",
+    ]
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename2)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_some_successfull_files_and_some_with_unknown_data():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    items_file1 = {"Item": {"pk": {"S": "msg1"}, "data": {"S": f"{uuid4()}"}}}
+    filename1 = "data.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file1),
+        Key=filename1,
+    )
+
+    items_file2 = {"pk": {"S": "msg2"}, "data": {"S": f"{uuid4()}"}}
+    filename2 = "invaliddata"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file2),
+        Key=filename2,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "FAILED"
+    assert import_details["ErrorCount"] == 1
+    assert import_details["ProcessedItemCount"] == 2
+    assert import_details["ImportedItemCount"] == 1
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert [i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]] == [
+        "msg1"
+    ]
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename2)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_only_process_file_with_prefix():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    items_file1 = {"Item": {"pk": {"S": "msg1"}, "data": {"S": f"{uuid4()}"}}}
+    filename1 = "yesdata.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file1),
+        Key=filename1,
+    )
+
+    items_file2 = {"Item": {"pk": {"S": "msg2"}, "data": {"S": f"{uuid4()}"}}}
+    filename2 = "nodata.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=json.dumps(items_file2),
+        Key=filename2,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name, "S3KeyPrefix": "yes"},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="NONE",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 1
+    assert import_details["ImportedItemCount"] == 1
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert [i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]] == [
+        "msg1"
+    ]
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename2)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_process_gzipped_file():
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+
+    items_file1 = {"Item": {"pk": {"S": "msg1"}, "data": {"S": f"{uuid4()}"}}}
+    filename1 = "data.json"
+    s3.put_object(
+        Bucket=s3_bucket_name,
+        Body=gzip.compress(json.dumps(items_file1).encode("utf-8")),
+        Key=filename1,
+    )
+
+    import_description = client.import_table(
+        S3BucketSource={"S3Bucket": s3_bucket_name},
+        InputFormat="DYNAMODB_JSON",
+        InputCompressionType="GZIP",
+        TableCreationParameters={
+            "TableName": table_name,
+            "AttributeDefinitions": [
+                {"AttributeName": "pk", "AttributeType": "S"},
+            ],
+            "KeySchema": [
+                {"AttributeName": "pk", "KeyType": "HASH"},
+            ],
+            "BillingMode": "PAY_PER_REQUEST",
+        },
+    )["ImportTableDescription"]
+
+    import_details = wait_for_import(client, import_description)
+
+    assert import_details["ImportStatus"] == "COMPLETED"
+    assert import_details["ErrorCount"] == 0
+    assert import_details["ProcessedItemCount"] == 1
+    assert import_details["ImportedItemCount"] == 1
+    assert import_details["ProcessedSizeBytes"] > 0
+
+    assert table_name in client.list_tables()["TableNames"]
+
+    assert [i["pk"]["S"] for i in client.scan(TableName=table_name)["Items"]] == [
+        "msg1"
+    ]
+
+    client.delete_table(TableName=table_name)
+
+    s3.delete_object(Bucket=s3_bucket_name, Key=filename1)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+def wait_for_import(client, import_description):
+    import_details = client.describe_import(ImportArn=import_description["ImportArn"])
+    status = import_details["ImportTableDescription"]["ImportStatus"]
+    while status == "IN_PROGRESS":
+        sleep(0.1)
+        import_details = client.describe_import(
+            ImportArn=import_description["ImportArn"]
+        )
+        status = import_details["ImportTableDescription"]["ImportStatus"]
+    return import_details["ImportTableDescription"]
diff --git a/tests/test_ec2/test_network_acls.py b/tests/test_ec2/test_network_acls.py
index f861974f2..bde8c726c 100644
--- a/tests/test_ec2/test_network_acls.py
+++ b/tests/test_ec2/test_network_acls.py
@@ -421,6 +421,94 @@ def test_describe_network_acls():
     )
 
 
+@mock_aws
+def test_describe_network_acls_associations_by_subnet():
+    ec2_client = boto3.client("ec2", region_name="us-west-2")
+    vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
+    vpc_id = vpc["Vpc"]["VpcId"]
+    all_acls = ec2_client.describe_network_acls(
+        Filters=[{"Name": "vpc-id", "Values": [vpc_id]}]
+    )["NetworkAcls"]
+    subnet_1_id = ec2_client.create_subnet(VpcId=vpc_id, CidrBlock="10.0.1.0/24")[
+        "Subnet"
+    ]["SubnetId"]
+    subnet_2_id = ec2_client.create_subnet(VpcId=vpc_id, CidrBlock="10.0.2.0/24")[
+        "Subnet"
+    ]["SubnetId"]
+    subnet_3_id = ec2_client.create_subnet(VpcId=vpc_id, CidrBlock="10.0.3.0/24")[
+        "Subnet"
+    ]["SubnetId"]
+    default_acl_id = next(acl["NetworkAclId"] for acl in all_acls if acl["IsDefault"])
+    custom_acl_id = ec2_client.create_network_acl(VpcId=vpc_id)["NetworkAcl"][
+        "NetworkAclId"
+    ]
+
+    # Subnets should be associated with the default ACL by default
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_1_id, default_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_3_id, default_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_2_id, default_acl_id)
+
+    associate_subnet_to_acl(ec2_client, subnet_1_id, custom_acl_id)
+    associate_subnet_to_acl(ec2_client, subnet_3_id, custom_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_1_id, custom_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_3_id, custom_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_2_id, default_acl_id)
+
+    # Re-associate Subnet 1 with the default ACL to test re-association
+    associate_subnet_to_acl(ec2_client, subnet_1_id, default_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_1_id, default_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_2_id, default_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_3_id, custom_acl_id)
+
+    # Attempting to associate a non-existent ACL with a subnet should raise an error
+    with pytest.raises(ClientError):
+        associate_subnet_to_acl(ec2_client, subnet_1_id, "non-existent-acl")
+
+    # Attempting to delete a custom network ACL should fail if it has any subnets associated
+    with pytest.raises(ClientError) as exc:
+        ec2_client.delete_network_acl(NetworkAclId=custom_acl_id)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "DependencyViolation"
+    assert (
+        err["Message"]
+        == f"The network ACL '{custom_acl_id}' has dependencies and cannot be deleted."
+    )
+
+    # Re-associate Subnet 3 with the default ACL to test deletion of custom ACL
+    associate_subnet_to_acl(ec2_client, subnet_3_id, default_acl_id)
+    ec2_client.delete_network_acl(NetworkAclId=custom_acl_id)
+    assert_subnet_is_associated_to_acl(ec2_client, subnet_3_id, default_acl_id)
+
+
+def associate_subnet_to_acl(ec2_client, subnet_id: str, acl_id: str):
+    assoc_id = get_subnet_acl_association_id(ec2_client, subnet_id)
+    ec2_client.replace_network_acl_association(
+        NetworkAclId=acl_id, AssociationId=assoc_id
+    )
+
+
+def get_subnet_acl_association_id(ec2_client, subnet_id: str):
+    network_acls_description = ec2_client.describe_network_acls(
+        Filters=[{"Name": "association.subnet-id", "Values": [subnet_id]}]
+    )
+    associations = network_acls_description["NetworkAcls"][0]["Associations"]
+    return next(
+        assoc["NetworkAclAssociationId"]
+        for assoc in associations
+        if assoc["SubnetId"] == subnet_id
+    )
+
+
+def assert_subnet_is_associated_to_acl(ec2_client, subnet_id: str, acl_id: str):
+    network_acl_description = ec2_client.describe_network_acls(
+        Filters=[{"Name": "association.subnet-id", "Values": [subnet_id]}]
+    )
+    acls_subnet = network_acl_description["NetworkAcls"]
+    # There should be only one ACL associated with the subnet at any given time
+    assert len(acls_subnet) == 1
+    assert acls_subnet[0]["NetworkAclId"] == acl_id
+
+
 @mock_aws
 def test_create_network_acl_with_tags():
     conn = boto3.client("ec2", region_name="us-west-2")
diff --git a/tests/test_ec2/test_route_tables.py b/tests/test_ec2/test_route_tables.py
index dc13d84f3..4187103c4 100644
--- a/tests/test_ec2/test_route_tables.py
+++ b/tests/test_ec2/test_route_tables.py
@@ -919,6 +919,47 @@ def test_describe_route_tables_with_nat_gateway():
     assert nat_gw_routes[0]["State"] == "active"
 
 
+@mock_aws
+def test_describe_route_tables_filter_with_nat_gateway_id():
+    ec2 = boto3.client("ec2", region_name="us-west-1")
+    vpc_id = ec2.create_vpc(CidrBlock="192.168.0.0/23")["Vpc"]["VpcId"]
+    az = ec2.describe_availability_zones()["AvailabilityZones"][0]["ZoneName"]
+    sn_id = ec2.create_subnet(
+        AvailabilityZone=az, CidrBlock="192.168.0.0/24", VpcId=vpc_id
+    )["Subnet"]["SubnetId"]
+    route_table_id = ec2.create_route_table(VpcId=vpc_id)["RouteTable"]["RouteTableId"]
+    ec2.associate_route_table(SubnetId=sn_id, RouteTableId=route_table_id)
+    alloc_id = ec2.allocate_address(Domain="vpc")["AllocationId"]
+    nat_gw_id = ec2.create_nat_gateway(SubnetId=sn_id, AllocationId=alloc_id)[
+        "NatGateway"
+    ]["NatGatewayId"]
+
+    route_tables_by_nat_gateway_id = ec2.describe_route_tables(
+        Filters=[{"Name": "route.nat-gateway-id", "Values": [nat_gw_id]}]
+    )["RouteTables"]
+    assert len(route_tables_by_nat_gateway_id) == 0
+
+    ec2.create_route(
+        DestinationCidrBlock="0.0.0.0/0",
+        NatGatewayId=nat_gw_id,
+        RouteTableId=route_table_id,
+    )
+
+    route_tables_by_nat_gateway_id = ec2.describe_route_tables(
+        Filters=[{"Name": "route.nat-gateway-id", "Values": [nat_gw_id]}]
+    )["RouteTables"][0]
+    nat_gw_routes = [
+        route
+        for route in route_tables_by_nat_gateway_id["Routes"]
+        if route["DestinationCidrBlock"] == "0.0.0.0/0"
+    ]
+
+    assert len(nat_gw_routes) == 1
+    assert nat_gw_routes[0]["DestinationCidrBlock"] == "0.0.0.0/0"
+    assert nat_gw_routes[0]["NatGatewayId"] == nat_gw_id
+    assert nat_gw_routes[0]["State"] == "active"
+
+
 @mock_aws
 def test_create_vpc_end_point():
     ec2 = boto3.client("ec2", region_name="us-west-1")
diff --git a/tests/test_emrserverless/test_emrserverless.py b/tests/test_emrserverless/test_emrserverless.py
index c32f5b6d3..9a9ba6f4d 100644
--- a/tests/test_emrserverless/test_emrserverless.py
+++ b/tests/test_emrserverless/test_emrserverless.py
@@ -70,6 +70,106 @@ def fixture_application_factory(client):
     yield application_list
 
 
+@pytest.fixture(scope="function", name="available_application")
+def fixture_available_application(client):
+    resp = client.create_application(
+        name="test-emr-serverless-application",
+        type="SPARK",
+        releaseLabel=DEFAULT_RELEASE_LABEL,
+    )
+
+    yield resp["applicationId"]
+
+
+@pytest.fixture(scope="function", name="job_run_factory")
+def fixture_job_run_factory(client):
+    app_1_resp = client.create_application(
+        name="test-emr-serverless-application-1",
+        type="SPARK",
+        releaseLabel=DEFAULT_RELEASE_LABEL,
+    )
+    app_1_id = app_1_resp["applicationId"]
+
+    job_1_resp = client.start_job_run(
+        name="Test Job Run 1",
+        applicationId=app_1_id,
+        executionRoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/emr-serverless-role",
+        jobDriver={
+            "sparkSubmit": {
+                "entryPoint": "test.jar",
+                "entryPointArguments": ["-h"],
+                "sparkSubmitParameters": "--num-executors 1",
+            }
+        },
+    )
+
+    job_2_resp = client.start_job_run(
+        name="Test Job Run 2",
+        applicationId=app_1_id,
+        executionRoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/emr-serverless-role",
+        jobDriver={
+            "sparkSubmit": {
+                "entryPoint": "test.jar",
+                "entryPointArguments": ["-h"],
+                "sparkSubmitParameters": "--num-executors 1",
+            }
+        },
+        configurationOverrides={
+            "monitoringConfiguration": {
+                "s3MonitoringConfiguration": {"logUri": "s3://DOC-EXAMPLE-BUCKET/logs"}
+            }
+        },
+        tags={"tag1": "tag1_val"},
+        executionTimeoutMinutes=5,
+    )
+
+    app_2_resp = client.create_application(
+        name="test-emr-serverless-application-2",
+        type="SPARK",
+        releaseLabel=DEFAULT_RELEASE_LABEL,
+        tags={"tag1": "tag1_val"},
+    )
+    app_2_id = app_2_resp["applicationId"]
+
+    job_3_resp = client.start_job_run(
+        name="Test Job Run 3",
+        applicationId=app_2_id,
+        executionRoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/emr-serverless-role",
+        jobDriver={
+            "sparkSubmit": {
+                "entryPoint": "test.jar",
+                "entryPointArguments": ["-h"],
+                "sparkSubmitParameters": "--num-executors 1",
+            }
+        },
+    )
+
+    job_4_resp = client.start_job_run(
+        name="Test Job Run 4",
+        applicationId=app_2_id,
+        executionRoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/emr-serverless-role",
+        jobDriver={
+            "sparkSubmit": {
+                "entryPoint": "test.jar",
+                "entryPointArguments": ["-h"],
+                "sparkSubmitParameters": "--num-executors 1",
+            }
+        },
+        configurationOverrides={
+            "monitoringConfiguration": {
+                "s3MonitoringConfiguration": {"logUri": "s3://DOC-EXAMPLE-BUCKET/logs"}
+            }
+        },
+        tags={"tag1": "tag1_val"},
+        executionTimeoutMinutes=5,
+    )
+
+    yield {
+        app_1_id: [job_1_resp["jobRunId"], job_2_resp["jobRunId"]],
+        app_2_id: [job_3_resp["jobRunId"], job_4_resp["jobRunId"]],
+    }
+
+
 class TestCreateApplication:
     @staticmethod
     @mock_aws
@@ -83,7 +183,7 @@ class TestCreateApplication:
         assert resp["name"] == "test-emr-serverless-application"
         assert re.match(r"[a-z,0-9]{16}", resp["applicationId"])
         assert resp["arn"] == (
-            f"arn:aws:emr-containers:us-east-1:{ACCOUNT_ID}"
+            f"arn:aws:emr-serverless:{DEFAULT_REGION}:{ACCOUNT_ID}"
             f":/applications/{resp['applicationId']}"
         )
 
@@ -183,7 +283,7 @@ class TestGetApplication:
         response = {
             "applicationId": application_id,
             "name": "test-emr-serverless-application",
-            "arn": f"arn:aws:emr-containers:us-east-1:123456789012:/applications/{application_id}",
+            "arn": f"arn:aws:emr-serverless:{DEFAULT_REGION}:{ACCOUNT_ID}:/applications/{application_id}",
             "releaseLabel": "emr-6.6.0",
             "type": "Spark",
             "state": "STARTED",
@@ -292,7 +392,7 @@ class TestListApplication:
             "id": self.application_ids[0],
             "name": "test-emr-serverless-application-STARTED",
             "arn": (
-                f"arn:aws:emr-containers:us-east-1:123456789012"
+                f"arn:aws:emr-serverless:{DEFAULT_REGION}:{ACCOUNT_ID}"
                 f":/applications/{self.application_ids[0]}"
             ),
             "releaseLabel": "emr-6.6.0",
@@ -407,7 +507,7 @@ class TestUpdateApplication:
         response = {
             "applicationId": application_id,
             "name": "test-emr-serverless-application-STOPPED",
-            "arn": f"arn:aws:emr-containers:us-east-1:123456789012:/applications/{application_id}",
+            "arn": f"arn:aws:emr-serverless:{DEFAULT_REGION}:{ACCOUNT_ID}:/applications/{application_id}",
             "releaseLabel": "emr-6.6.0",
             "type": "Spark",
             "state": "STOPPED",
@@ -547,3 +647,258 @@ class TestUpdateApplication:
         err = exc.value.response["Error"]
         assert err["Code"] == "ResourceNotFoundException"
         assert err["Message"] == "Application fake_application_id does not exist"
+
+
+class TestStartJobRun:
+    @pytest.fixture(autouse=True)
+    def _setup_environment(
+        self, client, application_factory, available_application, job_run_factory
+    ):
+        self.client = client
+        self.application_ids: list[str] = application_factory
+        self.job_run_lookup: dict[str, list[dict]] = job_run_factory
+        self.available_application: str = available_application
+
+    def test_start_job_run(self):
+        application_id = self.available_application
+        resp = self.client.start_job_run(
+            applicationId=application_id,
+            executionRoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/emr-serverless-role",
+            jobDriver={
+                "sparkSubmit": {
+                    "entryPoint": "test.jar",
+                    "entryPointArguments": ["-h"],
+                    "sparkSubmitParameters": "--num-executors 1",
+                }
+            },
+            configurationOverrides={
+                "monitoringConfiguration": {
+                    "s3MonitoringConfiguration": {
+                        "logUri": "s3://DOC-EXAMPLE-BUCKET/logs"
+                    }
+                }
+            },
+            tags={"tag1": "tag1_val"},
+            executionTimeoutMinutes=10,
+            name="Test Job Run",
+        )
+
+        assert isinstance(resp, dict)
+        assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+        assert resp["applicationId"] == application_id
+        assert resp["arn"].startswith(
+            f"arn:aws:emr-serverless:{DEFAULT_REGION}:{ACCOUNT_ID}:/applications/{application_id}/jobruns/"
+        )
+        assert re.match(r"[a-z,0-9]{16}", resp["jobRunId"])
+
+    def test_invalid_application_id(self):
+        with pytest.raises(ClientError) as exc:
+            self.client.start_job_run(
+                applicationId="fake_application_id",
+                executionRoleArn="arn:aws:iam::{ACCOUNT_ID}:role/emr-serverless-role",
+                jobDriver={
+                    "sparkSubmit": {
+                        "entryPoint": "test.jar",
+                        "entryPointArguments": ["-h"],
+                        "sparkSubmitParameters": "--num-executors 1",
+                    }
+                },
+            )
+
+            err = exc.value.response["Error"]
+            assert err["Code"] == "ResourceNotFoundException"
+            assert err["Message"] == "Application fake_application_id does not exist"
+
+    def test_cross_account_role(self):
+        with pytest.raises(ClientError) as exc:
+            different_account_id = "999999999999"
+            self.client.start_job_run(
+                applicationId=self.application_ids[0],
+                executionRoleArn=f"arn:aws:iam::{different_account_id}:role/emr-serverless-role",
+                jobDriver={
+                    "sparkSubmit": {
+                        "entryPoint": "test.jar",
+                        "entryPointArguments": ["-h"],
+                        "sparkSubmitParameters": "--num-executors 1",
+                    }
+                },
+            )
+        err = exc.value.response["Error"]
+        assert err["Code"] == "AccessDeniedException"
+        assert err["Message"] == "Cross-account pass role is not allowed."
+
+    def test_run_timeout(self):
+        with pytest.raises(ClientError) as exc:
+            self.client.start_job_run(
+                applicationId=self.application_ids[0],
+                executionRoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/emr-serverless-role",
+                jobDriver={
+                    "sparkSubmit": {
+                        "entryPoint": "test.jar",
+                        "entryPointArguments": ["-h"],
+                        "sparkSubmitParameters": "--num-executors 1",
+                    }
+                },
+                executionTimeoutMinutes=4,
+            )
+        err = exc.value.response["Error"]
+        assert err["Code"] == "ValidationException"
+        assert err["Message"] == "RunTimeout must be at least 5 minutes."
+
+
+class TestGetJobRun:
+    @pytest.fixture(autouse=True)
+    def _setup_environment(
+        self, client, application_factory, available_application, job_run_factory
+    ):
+        self.client = client
+        self.application_ids: list[str] = application_factory
+        self.job_run_lookup: dict[str, list[dict]] = job_run_factory
+        self.available_application: str = available_application
+
+    def test_job_not_belongs_to_other_application(self):
+        app_1_id, app_2_id, *_ = self.job_run_lookup.keys()
+        app_2_job_run_ids = self.job_run_lookup[app_2_id]
+        for run_id in app_2_job_run_ids:
+            # Use application 1 ID and job run from application 2
+            with pytest.raises(ClientError) as exc:
+                _ = self.client.get_job_run(applicationId=app_1_id, jobRunId=run_id)
+            err = exc.value.response["Error"]
+            assert err["Code"] == "ResourceNotFoundException"
+
+    def test_get_job_run(self):
+        for app_id, run_ids in self.job_run_lookup.items():
+            for run_id in run_ids:
+                resp = self.client.get_job_run(applicationId=app_id, jobRunId=run_id)
+                assert resp is not None
+                assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+                assert resp["jobRun"]["applicationId"] == app_id
+                assert resp["jobRun"]["jobRunId"] == run_id
+
+    def test_invalid_application_id(self):
+        for _, run_ids in self.job_run_lookup.items():
+            fake_app_id = "fakeapp"
+            for run_id in run_ids:
+                with pytest.raises(ClientError) as exc:
+                    self.client.get_job_run(applicationId=fake_app_id, jobRunId=run_id)
+                err = exc.value.response["Error"]
+                assert err["Code"] == "ResourceNotFoundException"
+
+    def test_invalid_job_run_id(self):
+        for app_id, _ in self.job_run_lookup.items():
+            job_run_id = "fakejobrun"
+            with pytest.raises(ClientError) as exc:
+                self.client.get_job_run(applicationId=app_id, jobRunId=job_run_id)
+            err = exc.value.response["Error"]
+            assert err["Code"] == "ResourceNotFoundException"
+
+
+class TestListJobRun:
+    @pytest.fixture(autouse=True)
+    def _setup_environment(
+        self, client, application_factory, available_application, job_run_factory
+    ):
+        self.client = client
+        self.application_ids: list[str] = application_factory
+        self.job_run_lookup: dict[str, list[dict]] = job_run_factory
+        self.available_application: str = available_application
+
+    def test_list_job_runs(self):
+        for app_id, run_ids in self.job_run_lookup.items():
+            resp = self.client.list_job_runs(applicationId=app_id)
+            assert resp is not None
+            assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+            assert len(resp["jobRuns"]) == len(run_ids)
+            assert all(run["applicationId"] == app_id for run in resp["jobRuns"])
+            assert all(run["id"] in run_ids for run in resp["jobRuns"])
+
+    def test_invalid_application_id(self):
+        fake_app_id = "fakeapp"
+        with pytest.raises(ClientError) as exc:
+            self.client.list_job_runs(applicationId=fake_app_id)
+        err = exc.value.response["Error"]
+        assert err["Code"] == "ResourceNotFoundException"
+
+    def test_application_states(self):
+        for app_id, run_ids in self.job_run_lookup.items():
+            resp = self.client.list_job_runs(applicationId=app_id, states=["COMPLETED"])
+            assert resp is not None
+            assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+            assert all(run["applicationId"] == app_id for run in resp["jobRuns"])
+            assert all(
+                run["id"] in run_ids
+                for run in resp["jobRuns"]
+                if run["state"] == "COMPLETED"
+            )
+
+    def test_created_filters(self):
+        for app_id, _ in self.job_run_lookup.items():
+            resp = self.client.list_job_runs(
+                applicationId=app_id,
+                createdAtAfter=datetime(2024, 1, 1),
+                createdAtBefore=datetime(2024, 1, 2),
+            )
+            assert resp is not None
+            assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+            assert len(resp["jobRuns"]) == 0
+
+    def test_created_after(self):
+        for app_id, run_ids in self.job_run_lookup.items():
+            resp = self.client.list_job_runs(
+                applicationId=app_id, createdAtAfter=datetime(2024, 1, 1)
+            )
+            assert resp is not None
+            assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+            assert len(resp["jobRuns"]) == len(run_ids)
+
+    def test_max_results(self):
+        for app_id, _ in self.job_run_lookup.items():
+            resp = self.client.list_job_runs(applicationId=app_id, maxResults=1)
+            assert resp is not None
+            assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+            assert len(resp["jobRuns"]) == 1
+
+    def test_invalid_job_run_id(self):
+        for app_id, _ in self.job_run_lookup.items():
+            job_run_id = "fakejobrun"
+            with pytest.raises(ClientError) as exc:
+                self.client.cancel_job_run(applicationId=app_id, jobRunId=job_run_id)
+            err = exc.value.response["Error"]
+            assert err["Code"] == "ResourceNotFoundException"
+
+
+class TestCancelJobRun:
+    @pytest.fixture(autouse=True)
+    def _setup_environment(
+        self, client, application_factory, available_application, job_run_factory
+    ):
+        self.client = client
+        self.application_ids: list[str] = application_factory
+        self.job_run_lookup: dict[str, list[dict]] = job_run_factory
+        self.available_application: str = available_application
+
+    def test_cancel_job_run(self):
+        for app_id, run_ids in self.job_run_lookup.items():
+            for run_id in run_ids:
+                resp = self.client.cancel_job_run(applicationId=app_id, jobRunId=run_id)
+                assert resp is not None
+                assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+                assert resp["applicationId"] == app_id
+                assert resp["jobRunId"] == run_id
+
+    def test_invalid_application_id(self):
+        for _, run_ids in self.job_run_lookup.items():
+            fake_app_id = "fakeapp"
+            for run_id in run_ids:
+                with pytest.raises(ClientError) as exc:
+                    self.client.get_job_run(applicationId=fake_app_id, jobRunId=run_id)
+                err = exc.value.response["Error"]
+                assert err["Code"] == "ResourceNotFoundException"
+
+    def test_invalid_job_run_id(self):
+        for app_id, _ in self.job_run_lookup.items():
+            job_run_id = "fakejobrun"
+            with pytest.raises(ClientError) as exc:
+                self.client.cancel_job_run(applicationId=app_id, jobRunId=job_run_id)
+            err = exc.value.response["Error"]
+            assert err["Code"] == "ResourceNotFoundException"
diff --git a/tests/test_wafv2/test_wafv2.py b/tests/test_wafv2/test_wafv2.py
index 56aacd9f3..54ba198be 100644
--- a/tests/test_wafv2/test_wafv2.py
+++ b/tests/test_wafv2/test_wafv2.py
@@ -197,3 +197,150 @@ def test_update_web_acl():
         "CloudWatchMetricsEnabled": True,
         "MetricName": "updated",
     }
+
+
+@mock_aws
+def test_ip_set_crud():
+    client = boto3.client("wafv2", region_name="us-east-1")
+
+    create_response = client.create_ip_set(
+        Name="test-ip-set",
+        Scope="CLOUDFRONT",
+        Description="Test IP set",
+        IPAddressVersion="IPV4",
+        Addresses=["192.168.0.1/32", "10.0.0.0/8"],
+        Tags=[{"Key": "Environment", "Value": "Test"}],
+    )
+
+    assert "Summary" in create_response
+    summary = create_response["Summary"]
+
+    assert summary["Id"]
+    assert summary["LockToken"]
+    assert summary["ARN"]
+    assert summary["Name"] == "test-ip-set"
+    assert "Test IP set" in summary["Description"]
+
+    get_response = client.get_ip_set(
+        Name=summary["Name"], Scope="CLOUDFRONT", Id=summary["Id"]
+    )
+    assert "IPSet" in get_response
+    assert "LockToken" in get_response
+
+    ip_set = get_response["IPSet"]
+    assert ip_set["IPAddressVersion"] == "IPV4"
+
+    with pytest.raises(ClientError) as e:
+        client.update_ip_set(
+            Name="test-ip-set",
+            Scope="CLOUDFRONT",
+            Id=summary["Id"],
+            Addresses=["10.0.0.0/8"],
+            LockToken="aaaaaaaaaaaaaaaaaa",  # invalid lock token
+        )
+    e.value.response["Error"]["Code"] == "WAFOptimisticLockException"
+    e.value.response["Error"][
+        "Message"
+    ] == "AWS WAF couldn’t save your changes because someone changed the resource after you started to edit it. Reapply your changes."
+
+    update_response = client.update_ip_set(
+        Name="test-ip-set",
+        Scope="CLOUDFRONT",
+        Id=summary["Id"],
+        Description="Updated test IP set",
+        Addresses=["192.168.1.0/24", "10.0.0.0/8"],
+        LockToken=get_response["LockToken"],
+    )
+
+    assert "NextLockToken" in update_response
+
+    updated_get_response = client.get_ip_set(
+        Name=summary["Name"], Scope="CLOUDFRONT", Id=summary["Id"]
+    )
+
+    updated_ip_set = updated_get_response["IPSet"]
+    assert updated_ip_set["Description"] == "Updated test IP set"
+    assert updated_get_response["LockToken"] == update_response["NextLockToken"]
+    assert all(
+        addr in ["192.168.1.0/24", "10.0.0.0/8"] for addr in updated_ip_set["Addresses"]
+    )
+
+    list_response = client.list_ip_sets(Scope="CLOUDFRONT")
+    assert len(list_response["IPSets"]) == 1
+
+    assert all(
+        [
+            key in list_response["IPSets"][0]
+            for key in ["ARN", "Description", "Id", "LockToken", "Name"]
+        ]
+    )
+    assert "NextMarker" in list_response
+
+    client.delete_ip_set(
+        Name=summary["Name"],
+        Scope="CLOUDFRONT",
+        Id=summary["Id"],
+        LockToken=updated_get_response["LockToken"],
+    )
+
+    with pytest.raises(ClientError) as e:
+        client.get_ip_set(Name=summary["Name"], Scope="CLOUDFRONT", Id=summary["Id"])
+    e.value.response["Error"]["Code"] == "WAFNonexistentItemException"
+    e.value.response["Error"][
+        "Message"
+    ] == "AWS WAF couldn’t perform the operation because your resource doesn’t exist."
+
+
+@mock_aws
+def test_logging_configuration_crud():
+    wafv2_client = boto3.client("wafv2", region_name="us-east-1")
+    create_web_acl_response = wafv2_client.create_web_acl(
+        Name="TestWebACL",
+        Scope="REGIONAL",
+        DefaultAction={"Allow": {}},
+        VisibilityConfig={
+            "SampledRequestsEnabled": True,
+            "CloudWatchMetricsEnabled": True,
+            "MetricName": "TestWebACLMetric",
+        },
+        Rules=[],
+    )
+    web_acl_arn = create_web_acl_response["Summary"]["ARN"]
+
+    # Create log groups
+    logs_client = boto3.client("logs", region_name="us-east-1")
+    logs_client.create_log_group(logGroupName="aws-waf-logs-test")
+    log_group = logs_client.describe_log_groups(logGroupNamePrefix="aws-waf-logs-test")[
+        "logGroups"
+    ][0]
+
+    create_response = wafv2_client.put_logging_configuration(
+        LoggingConfiguration={
+            "ResourceArn": web_acl_arn,
+            "LogDestinationConfigs": [log_group["arn"]],
+        }
+    )
+
+    assert "LoggingConfiguration" in create_response
+    logging_configuration = create_response["LoggingConfiguration"]
+    assert logging_configuration["ResourceArn"]
+
+    get_response = wafv2_client.get_logging_configuration(
+        ResourceArn=logging_configuration["ResourceArn"]
+    )
+    assert "LoggingConfiguration" in get_response
+
+    list_response = wafv2_client.list_logging_configurations(
+        Scope="REGIONAL",
+    )
+    assert len(list_response["LoggingConfigurations"]) > 0
+
+    wafv2_client.delete_logging_configuration(
+        ResourceArn=logging_configuration["ResourceArn"],
+    )
+
+    with pytest.raises(ClientError) as e:
+        wafv2_client.get_logging_configuration(
+            ResourceArn=logging_configuration["ResourceArn"]
+        )
+    e.value.response["Error"]["Code"] == "WAFNonexistentItemException"