Repository: getmoto/moto
Pull request #8532 (2025-01-28 19:04:45)

CloudHSMv2 Integration
Resolves https://github.com/getmoto/moto/issues/8571

Singha22: Hey @bblommers do you know why I keep on getting these errors in the server tests?

```
E           botocore.exceptions.ClientError: An error occurred (500) when calling the TagResource operation (reached max retries: 4): <!doctype html>
E           <html lang=en>
E           <title>500 Internal Server Error</title>
E           <h1>Internal Server Error</h1>
E           <p>The server encountered an internal error and was unable to complete your request. Either the server is overloaded or there is an error in the application.</p>
```

Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 68d5289fc..2bbb9454f 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -995,7 +995,7 @@
 
 ## cloudformation
 <details>
-<summary>39% implemented</summary>
+<summary>37% implemented</summary>
 
 - [ ] activate_organizations_access
 - [ ] activate_type
@@ -1006,6 +1006,7 @@
 - [ ] create_generated_template
 - [X] create_stack
 - [X] create_stack_instances
+- [ ] create_stack_refactor
 - [X] create_stack_set
 - [ ] deactivate_organizations_access
 - [ ] deactivate_type
@@ -1025,6 +1026,7 @@
 - [ ] describe_stack_drift_detection_status
 - [X] describe_stack_events
 - [X] describe_stack_instance
+- [ ] describe_stack_refactor
 - [X] describe_stack_resource
 - [ ] describe_stack_resource_drifts
 - [X] describe_stack_resources
@@ -1038,6 +1040,7 @@
 - [ ] detect_stack_set_drift
 - [ ] estimate_template_cost
 - [X] execute_change_set
+- [ ] execute_stack_refactor
 - [ ] get_generated_template
 - [X] get_stack_policy
 - [X] get_template
@@ -1053,6 +1056,8 @@
 - [ ] list_resource_scans
 - [ ] list_stack_instance_resource_drifts
 - [X] list_stack_instances
+- [ ] list_stack_refactor_actions
+- [ ] list_stack_refactors
 - [X] list_stack_resources
 - [ ] list_stack_set_auto_deployment_targets
 - [X] list_stack_set_operation_results
@@ -1210,6 +1215,30 @@
 - [ ] update_vpc_origin
 </details>
 
+## cloudhsmv2
+<details>
+<summary>50% implemented</summary>
+
+- [ ] copy_backup_to_region
+- [X] create_cluster
+- [ ] create_hsm
+- [ ] delete_backup
+- [X] delete_cluster
+- [ ] delete_hsm
+- [ ] delete_resource_policy
+- [X] describe_backups
+- [X] describe_clusters
+- [X] get_resource_policy
+- [ ] initialize_cluster
+- [X] list_tags
+- [ ] modify_backup_attributes
+- [ ] modify_cluster
+- [X] put_resource_policy
+- [ ] restore_backup
+- [X] tag_resource
+- [X] untag_resource
+</details>
+
 ## cloudtrail
 <details>
 <summary>28% implemented</summary>
@@ -5309,21 +5338,21 @@
 
 ## logs
 <details>
-<summary>40% implemented</summary>
+<summary>57% implemented</summary>
 
 - [ ] associate_kms_key
 - [X] cancel_export_task
-- [ ] create_delivery
+- [X] create_delivery
 - [X] create_export_task
 - [ ] create_log_anomaly_detector
 - [X] create_log_group
 - [X] create_log_stream
 - [ ] delete_account_policy
 - [ ] delete_data_protection_policy
-- [ ] delete_delivery
-- [ ] delete_delivery_destination
-- [ ] delete_delivery_destination_policy
-- [ ] delete_delivery_source
+- [X] delete_delivery
+- [X] delete_delivery_destination
+- [X] delete_delivery_destination_policy
+- [X] delete_delivery_source
 - [X] delete_destination
 - [ ] delete_index_policy
 - [ ] delete_integration
@@ -5338,9 +5367,9 @@
 - [ ] delete_transformer
 - [ ] describe_account_policies
 - [ ] describe_configuration_templates
-- [ ] describe_deliveries
-- [ ] describe_delivery_destinations
-- [ ] describe_delivery_sources
+- [X] describe_deliveries
+- [X] describe_delivery_destinations
+- [X] describe_delivery_sources
 - [X] describe_destinations
 - [X] describe_export_tasks
 - [ ] describe_field_indexes
@@ -5355,10 +5384,10 @@
 - [ ] disassociate_kms_key
 - [X] filter_log_events
 - [ ] get_data_protection_policy
-- [ ] get_delivery
-- [ ] get_delivery_destination
-- [ ] get_delivery_destination_policy
-- [ ] get_delivery_source
+- [X] get_delivery
+- [X] get_delivery_destination
+- [X] get_delivery_destination_policy
+- [X] get_delivery_source
 - [ ] get_integration
 - [ ] get_log_anomaly_detector
 - [X] get_log_events
@@ -5374,9 +5403,9 @@
 - [X] list_tags_log_group
 - [ ] put_account_policy
 - [ ] put_data_protection_policy
-- [ ] put_delivery_destination
-- [ ] put_delivery_destination_policy
-- [ ] put_delivery_source
+- [X] put_delivery_destination
+- [X] put_delivery_destination_policy
+- [X] put_delivery_source
 - [X] put_destination
 - [X] put_destination_policy
 - [ ] put_index_policy
@@ -9532,7 +9561,6 @@
 - clouddirectory
 - cloudfront-keyvaluestore
 - cloudhsm
-- cloudhsmv2
 - cloudsearch
 - cloudsearchdomain
 - cloudtrail-data
diff --git a/docs/docs/services/cloudhsmv2.rst b/docs/docs/services/cloudhsmv2.rst
new file mode 100644
index 000000000..273b16c71
--- /dev/null
+++ b/docs/docs/services/cloudhsmv2.rst
@@ -0,0 +1,37 @@
+.. _implementedservice_cloudhsmv2:
+
+.. |start-h3| raw:: html
+
+    <h3>
+
+.. |end-h3| raw:: html
+
+    </h3>
+
+==========
+cloudhsmv2
+==========
+
+.. autoclass:: moto.cloudhsmv2.models.CloudHSMV2Backend
+
+|start-h3| Implemented features for this service |end-h3|
+
+- [ ] copy_backup_to_region
+- [X] create_cluster
+- [ ] create_hsm
+- [ ] delete_backup
+- [X] delete_cluster
+- [ ] delete_hsm
+- [ ] delete_resource_policy
+- [X] describe_backups
+- [X] describe_clusters
+- [X] get_resource_policy
+- [ ] initialize_cluster
+- [X] list_tags
+- [ ] modify_backup_attributes
+- [ ] modify_cluster
+- [X] put_resource_policy
+- [ ] restore_backup
+- [X] tag_resource
+- [X] untag_resource
+
diff --git a/docs/docs/services/logs.rst b/docs/docs/services/logs.rst
index b26cbcdaa..114ad3d64 100644
--- a/docs/docs/services/logs.rst
+++ b/docs/docs/services/logs.rst
@@ -16,17 +16,17 @@ logs
 
 - [ ] associate_kms_key
 - [X] cancel_export_task
-- [ ] create_delivery
+- [X] create_delivery
 - [X] create_export_task
 - [ ] create_log_anomaly_detector
 - [X] create_log_group
 - [X] create_log_stream
 - [ ] delete_account_policy
 - [ ] delete_data_protection_policy
-- [ ] delete_delivery
-- [ ] delete_delivery_destination
-- [ ] delete_delivery_destination_policy
-- [ ] delete_delivery_source
+- [X] delete_delivery
+- [X] delete_delivery_destination
+- [X] delete_delivery_destination_policy
+- [X] delete_delivery_source
 - [X] delete_destination
 - [ ] delete_index_policy
 - [ ] delete_integration
@@ -45,9 +45,9 @@ logs
 - [ ] delete_transformer
 - [ ] describe_account_policies
 - [ ] describe_configuration_templates
-- [ ] describe_deliveries
-- [ ] describe_delivery_destinations
-- [ ] describe_delivery_sources
+- [X] describe_deliveries
+- [X] describe_delivery_destinations
+- [X] describe_delivery_sources
 - [X] describe_destinations
 - [X] describe_export_tasks
   
@@ -83,10 +83,10 @@ logs
         
 
 - [ ] get_data_protection_policy
-- [ ] get_delivery
-- [ ] get_delivery_destination
-- [ ] get_delivery_destination_policy
-- [ ] get_delivery_source
+- [X] get_delivery
+- [X] get_delivery_destination
+- [X] get_delivery_destination_policy
+- [X] get_delivery_source
 - [ ] get_integration
 - [ ] get_log_anomaly_detector
 - [X] get_log_events
@@ -106,9 +106,9 @@ logs
 - [X] list_tags_log_group
 - [ ] put_account_policy
 - [ ] put_data_protection_policy
-- [ ] put_delivery_destination
-- [ ] put_delivery_destination_policy
-- [ ] put_delivery_source
+- [X] put_delivery_destination
+- [X] put_delivery_destination_policy
+- [X] put_delivery_source
 - [X] put_destination
 - [X] put_destination_policy
 - [ ] put_index_policy
diff --git a/moto/backend_index.py b/moto/backend_index.py
index c0480579c..81bc73472 100644
--- a/moto/backend_index.py
+++ b/moto/backend_index.py
@@ -29,6 +29,8 @@ backend_url_patterns = [
     ("cloudformation", re.compile("https?://cloudformation\\.(.+)\\.amazonaws\\.com")),
     ("cloudfront", re.compile("https?://cloudfront\\.amazonaws\\.com")),
     ("cloudfront", re.compile("https?://cloudfront\\.(.+)\\.amazonaws\\.com")),
+    ("cloudhsmv2", re.compile("https?://cloudhsm\\.(.+)\\.amazonaws\\.com")),
+    ("cloudhsmv2", re.compile("https?://cloudhsmv2\\.(.+)\\.amazonaws\\.com")),
     ("cloudtrail", re.compile("https?://cloudtrail\\.(.+)\\.amazonaws\\.com")),
     ("cloudwatch", re.compile("https?://monitoring\\.(.+)\\.amazonaws.com")),
     ("codebuild", re.compile("https?://codebuild\\.(.+)\\.amazonaws\\.com")),
diff --git a/moto/cloudhsmv2/__init__.py b/moto/cloudhsmv2/__init__.py
new file mode 100644
index 000000000..f64d2192a
--- /dev/null
+++ b/moto/cloudhsmv2/__init__.py
@@ -0,0 +1,1 @@
+from .models import cloudhsmv2_backends  # noqa: F401
diff --git a/moto/cloudhsmv2/exceptions.py b/moto/cloudhsmv2/exceptions.py
new file mode 100644
index 000000000..02e7271ab
--- /dev/null
+++ b/moto/cloudhsmv2/exceptions.py
@@ -0,0 +1,24 @@
+"""Exceptions raised by the cloudhsmv2 service."""
+
+from moto.core.exceptions import JsonRESTError
+
+
+class CloudHSMv2ClientError(JsonRESTError):
+    """Base class for CloudHSMv2 errors."""
+
+    code = 400
+
+
+class ResourceNotFoundException(CloudHSMv2ClientError):
+    def __init__(self, message: str):
+        super().__init__("ResourceNotFoundException", message)
+
+
+class InvalidRequestException(CloudHSMv2ClientError):
+    def __init__(self, message: str):
+        super().__init__("InvalidRequestException", message)
+
+
+class ClientError(CloudHSMv2ClientError):
+    def __init__(self, error_type: str, message: str):
+        super().__init__(error_type, message)
diff --git a/moto/cloudhsmv2/models.py b/moto/cloudhsmv2/models.py
new file mode 100644
index 000000000..ffcda110d
--- /dev/null
+++ b/moto/cloudhsmv2/models.py
@@ -0,0 +1,300 @@
+"""CloudHSMV2Backend class with methods for supported APIs."""
+
+import uuid
+from typing import Any, Dict, List, Optional, Tuple
+
+from moto.core.base_backend import BackendDict, BaseBackend
+from moto.core.utils import utcnow
+from moto.utilities.paginator import Paginator, paginate
+
+from .exceptions import ResourceNotFoundException
+
+
+class Cluster:
+    def __init__(
+        self,
+        backup_retention_policy: Optional[Dict[str, str]],
+        hsm_type: str,
+        source_backup_id: Optional[str],
+        subnet_ids: List[str],
+        network_type: str = "IPV4",
+        tag_list: Optional[List[Dict[str, str]]] = None,
+        mode: str = "DEFAULT",
+        region_name: str = "us-east-1",
+    ):
+        self.cluster_id = str(uuid.uuid4())
+        self.backup_policy = "DEFAULT"
+        self.backup_retention_policy = backup_retention_policy
+        self.create_timestamp = utcnow()
+        self.hsms: List[Dict[str, Any]] = []
+        self.hsm_type = hsm_type
+        self.source_backup_id = source_backup_id
+        self.state = "ACTIVE"
+        self.state_message = "The cluster is ready for use."
+        self.subnet_mapping = {subnet_id: region_name for subnet_id in subnet_ids}
+        self.vpc_id = "vpc-" + str(uuid.uuid4())[:8]
+        self.network_type = network_type
+        self.certificates = {
+            "ClusterCsr": "",
+            "HsmCertificate": "",
+            "AwsHardwareCertificate": "",
+            "ManufacturerHardwareCertificate": "",
+            "ClusterCertificate": "",
+        }
+        self.tag_list = tag_list or []
+        self.mode = mode
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "BackupPolicy": self.backup_policy,
+            "BackupRetentionPolicy": self.backup_retention_policy,
+            "ClusterId": self.cluster_id,
+            "CreateTimestamp": self.create_timestamp,
+            "Hsms": self.hsms,
+            "HsmType": self.hsm_type,
+            "SourceBackupId": self.source_backup_id,
+            "State": self.state,
+            "StateMessage": self.state_message,
+            "SubnetMapping": self.subnet_mapping,
+            "VpcId": self.vpc_id,
+            "NetworkType": self.network_type,
+            "Certificates": self.certificates,
+            "TagList": self.tag_list,
+            "Mode": self.mode,
+        }
+
+
+class Backup:
+    def __init__(
+        self,
+        cluster_id: str,
+        hsm_type: str,
+        mode: str,
+        tag_list: Optional[List[Dict[str, str]]],
+        source_backup: Optional[str] = None,
+        source_cluster: Optional[str] = None,
+        source_region: Optional[str] = None,
+        never_expires: bool = False,
+        region_name: str = "us-east-1",
+    ):
+        self.backup_id = str(uuid.uuid4())
+        self.backup_arn = (
+            f"arn:aws:cloudhsm:{region_name}:123456789012:backup/{self.backup_id}"
+        )
+        self.backup_state = "READY"
+        self.cluster_id = cluster_id
+        self.create_timestamp = utcnow()
+        self.copy_timestamp = utcnow() if source_backup else None
+        self.never_expires = never_expires
+        self.source_region = source_region
+        self.source_backup = source_backup
+        self.source_cluster = source_cluster
+        self.delete_timestamp = None
+        self.tag_list = tag_list or []
+        self.hsm_type = hsm_type
+        self.mode = mode
+
+    def to_dict(self) -> Dict[str, Any]:
+        result = {
+            "BackupId": self.backup_id,
+            "BackupArn": self.backup_arn,
+            "BackupState": self.backup_state,
+            "ClusterId": self.cluster_id,
+            "CreateTimestamp": self.create_timestamp,
+            "NeverExpires": self.never_expires,
+            "TagList": self.tag_list,
+            "HsmType": self.hsm_type,
+            "Mode": self.mode,
+        }
+
+        if self.copy_timestamp:
+            result["CopyTimestamp"] = self.copy_timestamp
+        if self.source_region:
+            result["SourceRegion"] = self.source_region
+        if self.source_backup:
+            result["SourceBackup"] = self.source_backup
+        if self.source_cluster:
+            result["SourceCluster"] = self.source_cluster
+        if self.delete_timestamp:
+            result["DeleteTimestamp"] = self.delete_timestamp
+
+        return result
+
+
+class CloudHSMV2Backend(BaseBackend):
+    """Implementation of CloudHSMV2 APIs."""
+
+    PAGINATION_MODEL = {
+        "describe_clusters": {
+            "input_token": "next_token",
+            "limit_key": "max_results",
+            "limit_default": 100,
+            "unique_attribute": "ClusterId",
+            "fail_on_invalid_token": False,
+        }
+    }
+
+    def __init__(self, region_name: str, account_id: str) -> None:
+        super().__init__(region_name, account_id)
+        self.tags: Dict[str, List[Dict[str, str]]] = {}
+        self.clusters: Dict[str, Cluster] = {}
+        self.resource_policies: Dict[str, str] = {}
+        self.backups: Dict[str, Backup] = {}
+
+    def list_tags(
+        self, resource_id: str, next_token: str, max_results: int
+    ) -> Tuple[List[Dict[str, str]], Optional[str]]:
+        """NEED TO IMPLEMENT PAGINATION"""
+        if resource_id not in self.tags:
+            return [], None
+
+        tags = sorted(self.tags.get(resource_id, []), key=lambda x: x["Key"])
+
+        return tags, None
+
+    def tag_resource(
+        self, resource_id: str, tag_list: List[Dict[str, str]]
+    ) -> Dict[str, Any]:
+        if resource_id not in self.tags:
+            self.tags[resource_id] = []
+
+        for new_tag in tag_list:
+            tag_exists = False
+            for existing_tag in self.tags[resource_id]:
+                if existing_tag["Key"] == new_tag["Key"]:
+                    existing_tag["Value"] = new_tag["Value"]
+                    tag_exists = True
+                    break
+            if not tag_exists:
+                self.tags[resource_id].append(new_tag)
+
+        return {}
+
+    def untag_resource(
+        self, resource_id: str, tag_key_list: List[str]
+    ) -> Dict[str, Any]:
+        if resource_id in self.tags:
+            self.tags[resource_id] = [
+                tag for tag in self.tags[resource_id] if tag["Key"] not in tag_key_list
+            ]
+
+        return {}
+
+    def create_cluster(
+        self,
+        backup_retention_policy: Optional[Dict[str, str]],
+        hsm_type: str,
+        source_backup_id: Optional[str],
+        subnet_ids: List[str],
+        network_type: Optional[str],
+        tag_list: Optional[List[Dict[str, str]]],
+        mode: Optional[str],
+    ) -> Dict[str, Any]:
+        cluster = Cluster(
+            backup_retention_policy=backup_retention_policy,
+            hsm_type=hsm_type,
+            source_backup_id=source_backup_id,
+            subnet_ids=subnet_ids,
+            network_type=network_type or "IPV4",
+            tag_list=tag_list,
+            mode=mode or "DEFAULT",
+            region_name=self.region_name,
+        )
+        self.clusters[cluster.cluster_id] = cluster
+
+        backup = Backup(
+            cluster_id=cluster.cluster_id,
+            hsm_type=hsm_type,
+            mode=mode or "DEFAULT",
+            tag_list=tag_list,
+            region_name=self.region_name,
+        )
+        self.backups[backup.backup_id] = backup
+
+        return cluster.to_dict()
+
+    def delete_cluster(self, cluster_id: str) -> Dict[str, Any]:
+        if cluster_id not in self.clusters:
+            raise ResourceNotFoundException(f"Cluster {cluster_id} not found")
+
+        cluster = self.clusters[cluster_id]
+        cluster.state = "DELETED"
+        cluster.state_message = "Cluster deleted"
+        del self.clusters[cluster_id]
+        return cluster.to_dict()
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def describe_clusters(
+        self, filters: Optional[Dict[str, List[str]]] = None
+    ) -> List[Dict[str, str]]:
+        """List all clusters with optional filtering.
+
+        Args:
+            filters: Optional dictionary of filters to apply
+
+        Returns:
+            List of cluster dictionaries
+        """
+        clusters = list(self.clusters.values())
+
+        if filters:
+            for key, values in filters.items():
+                if key == "clusterIds":
+                    clusters = [c for c in clusters if c.cluster_id in values]
+                elif key == "states":
+                    clusters = [c for c in clusters if c.state in values]
+                elif key == "vpcIds":
+                    clusters = [c for c in clusters if c.vpc_id in values]
+
+        clusters = sorted(clusters, key=lambda x: x.create_timestamp)
+        return [c.to_dict() for c in clusters]
+
+    def get_resource_policy(self, resource_arn: str) -> Optional[str]:
+        return self.resource_policies.get(resource_arn)
+
+    def describe_backups(
+        self,
+        next_token: Optional[str],
+        max_results: Optional[int],
+        filters: Optional[Dict[str, List[str]]],
+        shared: Optional[bool],
+        sort_ascending: Optional[bool],
+    ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
+        backups = list(self.backups.values())
+
+        if filters:
+            for key, values in filters.items():
+                if key == "backupIds":
+                    backups = [b for b in backups if b.backup_id in values]
+                elif key == "sourceBackupIds":
+                    backups = [b for b in backups if b.source_backup in values]
+                elif key == "clusterIds":
+                    backups = [b for b in backups if b.cluster_id in values]
+                elif key == "states":
+                    backups = [b for b in backups if b.backup_state in values]
+                elif key == "neverExpires":
+                    never_expires = values[0].lower() == "true"
+                    backups = [b for b in backups if b.never_expires == never_expires]
+
+        backups.sort(
+            key=lambda x: x.create_timestamp,
+            reverse=not sort_ascending if sort_ascending is not None else True,
+        )
+        if not max_results:
+            return [b.to_dict() for b in backups], None
+
+        paginator = Paginator(
+            max_results=max_results,
+            unique_attribute="BackupId",
+            starting_token=next_token,
+            fail_on_invalid_token=False,
+        )
+        results, token = paginator.paginate([b.to_dict() for b in backups])
+        return results, token
+
+    def put_resource_policy(self, resource_arn: str, policy: str) -> Dict[str, str]:
+        self.resource_policies[resource_arn] = policy
+        return {"ResourceArn": resource_arn, "Policy": policy}
+
+
+cloudhsmv2_backends = BackendDict(CloudHSMV2Backend, "cloudhsmv2")
diff --git a/moto/cloudhsmv2/responses.py b/moto/cloudhsmv2/responses.py
new file mode 100644
index 000000000..016fdc433
--- /dev/null
+++ b/moto/cloudhsmv2/responses.py
@@ -0,0 +1,160 @@
+"""Handles incoming cloudhsmv2 requests, invokes methods, returns responses."""
+
+import json
+from datetime import datetime
+from typing import Any
+
+from moto.core.responses import BaseResponse
+
+from .models import CloudHSMV2Backend, cloudhsmv2_backends
+
+
+class DateTimeEncoder(json.JSONEncoder):
+    def default(self, o: Any) -> Any:
+        if isinstance(o, datetime):
+            return o.isoformat()
+
+        return super().default(o)
+
+
+class CloudHSMV2Response(BaseResponse):
+    """Handler for CloudHSMV2 requests and responses."""
+
+    def __init__(self) -> None:
+        super().__init__(service_name="cloudhsmv2")
+
+    @property
+    def cloudhsmv2_backend(self) -> CloudHSMV2Backend:
+        """Return backend instance specific for this region."""
+        return cloudhsmv2_backends[self.current_account][self.region]
+
+    def list_tags(self) -> str:
+        params = json.loads(self.body)
+
+        resource_id = params.get("ResourceId")
+        next_token = params.get("NextToken")
+        max_results = params.get("MaxResults")
+
+        tag_list, next_token = self.cloudhsmv2_backend.list_tags(
+            resource_id=resource_id,
+            next_token=next_token,
+            max_results=max_results,
+        )
+
+        return json.dumps({"TagList": tag_list, "NextToken": next_token})
+
+    def tag_resource(self) -> str:
+        params = json.loads(self.body)
+
+        resource_id = params.get("ResourceId")
+        tag_list = params.get("TagList")
+
+        self.cloudhsmv2_backend.tag_resource(
+            resource_id=resource_id,
+            tag_list=tag_list,
+        )
+        return json.dumps(dict())
+
+    def untag_resource(self) -> str:
+        params = json.loads(self.body)
+
+        resource_id = params.get("ResourceId")
+        tag_key_list = params.get("TagKeyList")
+        self.cloudhsmv2_backend.untag_resource(
+            resource_id=resource_id,
+            tag_key_list=tag_key_list,
+        )
+        return json.dumps(dict())
+
+    def create_cluster(self) -> str:
+        backup_retention_policy = self._get_param("BackupRetentionPolicy", {})
+        hsm_type = self._get_param("HsmType")
+        source_backup_id = self._get_param("SourceBackupId")
+        subnet_ids = self._get_param("SubnetIds", [])
+        network_type = self._get_param("NetworkType")
+        tag_list = self._get_param("TagList")
+        mode = self._get_param("Mode")
+
+        cluster = self.cloudhsmv2_backend.create_cluster(
+            backup_retention_policy=backup_retention_policy,
+            hsm_type=hsm_type,
+            source_backup_id=source_backup_id,
+            subnet_ids=subnet_ids,
+            network_type=network_type,
+            tag_list=tag_list,
+            mode=mode,
+        )
+        return json.dumps({"Cluster": cluster}, cls=DateTimeEncoder)
+
+    def delete_cluster(self) -> str:
+        params = json.loads(self.body)
+
+        cluster_id = params.get("ClusterId")
+
+        cluster = self.cloudhsmv2_backend.delete_cluster(cluster_id=cluster_id)
+        return json.dumps({"Cluster": cluster}, cls=DateTimeEncoder)
+
+    def describe_clusters(self) -> str:
+        params = json.loads(self.body)
+
+        filters = params.get("Filters", {})
+        next_token = params.get("NextToken")
+        max_results = params.get("MaxResults")
+
+        clusters, next_token = self.cloudhsmv2_backend.describe_clusters(
+            filters=filters,
+            next_token=next_token,
+            max_results=max_results,
+        )
+
+        response = {"Clusters": clusters}
+        if next_token:
+            response["NextToken"] = next_token
+
+        return json.dumps(response, cls=DateTimeEncoder)
+
+    def get_resource_policy(self) -> str:
+        params = json.loads(self.body)
+        resource_arn = params.get("ResourceArn")
+        policy = self.cloudhsmv2_backend.get_resource_policy(
+            resource_arn=resource_arn,
+        )
+        return json.dumps({"Policy": policy})
+
+    def describe_backups(self) -> str:
+        params = json.loads(self.body)
+
+        next_token = params.get("NextToken")
+        max_results = params.get("MaxResults")
+        filters_raw = params.get("Filters", {})
+        filters = (
+            json.loads(filters_raw) if isinstance(filters_raw, str) else filters_raw
+        )
+        shared = params.get("Shared")
+        sort_ascending = params.get("SortAscending")
+
+        backups, next_token = self.cloudhsmv2_backend.describe_backups(
+            next_token=next_token,
+            max_results=max_results,
+            filters=filters,
+            shared=shared,
+            sort_ascending=sort_ascending,
+        )
+
+        response = {"Backups": backups}
+        if next_token:
+            response["NextToken"] = next_token
+
+        return json.dumps(response, cls=DateTimeEncoder)
+
+    def put_resource_policy(self) -> str:
+        params = json.loads(self.body)
+
+        resource_arn = params.get("ResourceArn")
+        policy = params.get("Policy")
+
+        result = self.cloudhsmv2_backend.put_resource_policy(
+            resource_arn=resource_arn,
+            policy=policy,
+        )
+        return json.dumps(result)
diff --git a/moto/cloudhsmv2/urls.py b/moto/cloudhsmv2/urls.py
new file mode 100644
index 000000000..d94cdd12a
--- /dev/null
+++ b/moto/cloudhsmv2/urls.py
@@ -0,0 +1,12 @@
+"""cloudhsmv2 base URL and path."""
+
+from .responses import CloudHSMV2Response
+
+url_bases = [
+    r"https?://cloudhsm\.(.+)\.amazonaws\.com",
+    r"https?://cloudhsmv2\.(.+)\.amazonaws\.com",
+]
+
+url_paths = {
+    "{0}/$": CloudHSMV2Response.dispatch,
+}
diff --git a/moto/logs/exceptions.py b/moto/logs/exceptions.py
index 4aea395c5..c34d74d04 100644
--- a/moto/logs/exceptions.py
+++ b/moto/logs/exceptions.py
@@ -44,3 +44,15 @@ class LimitExceededException(LogsClientError):
     def __init__(self) -> None:
         self.code = 400
         super().__init__("LimitExceededException", "Resource limit exceeded.")
+
+
+class ValidationException(LogsClientError):
+    def __init__(self, msg: str) -> None:
+        self.code = 400
+        super().__init__("ValidationException", msg)
+
+
+class ConflictException(LogsClientError):
+    def __init__(self, msg: str) -> None:
+        self.code = 400
+        super().__init__("ConflictException", msg)
diff --git a/moto/logs/models.py b/moto/logs/models.py
index 018b2fc55..3fc541c12 100644
--- a/moto/logs/models.py
+++ b/moto/logs/models.py
@@ -7,10 +7,12 @@ from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel, CloudFormationModel
 from moto.core.utils import unix_time_millis, utcnow
 from moto.logs.exceptions import (
+    ConflictException,
     InvalidParameterException,
     LimitExceededException,
     ResourceAlreadyExistsException,
     ResourceNotFoundException,
+    ValidationException,
 )
 from moto.logs.logs_query import execute_query
 from moto.logs.metric_filters import MetricFilters
@@ -760,6 +762,162 @@ class ExportTask(BaseModel):
         }
 
 
+class DeliveryDestination(BaseModel):
+    def __init__(
+        self,
+        account_id: str,
+        region: str,
+        name: str,
+        output_format: Optional[str],
+        delivery_destination_configuration: Dict[str, str],
+        tags: Optional[Dict[str, str]],
+        policy: Optional[str] = None,
+    ):
+        self.name = name
+        self.output_format = output_format
+        self.arn = f"arn:aws:logs:{region}:{account_id}:delivery-destination:{name}"
+        destination_type = delivery_destination_configuration[
+            "destinationResourceArn"
+        ].split(":")[2]
+        if destination_type == "s3":
+            self.delivery_destination_type = "S3"
+        elif destination_type == "logs":
+            self.delivery_destination_type = "CWL"
+        elif destination_type == "firehose":
+            self.delivery_destination_type = "FH"
+        self.delivery_destination_configuration = delivery_destination_configuration
+        self.tags = tags
+        self.policy = policy
+
+    def to_dict(self) -> Dict[str, Any]:
+        dct = {
+            "name": self.name,
+            "arn": self.arn,
+            "deliveryDestinationType": self.delivery_destination_type,
+            "outputFormat": self.output_format,
+            "deliveryDestinationConfiguration": self.delivery_destination_configuration,
+            "tags": self.tags,
+        }
+        dct_items = {k: v for k, v in dct.items() if v}
+        return dct_items
+
+
+class DeliverySource(BaseModel):
+    def __init__(
+        self,
+        account_id: str,
+        region: str,
+        name: str,
+        resource_arn: str,
+        log_type: str,
+        tags: Optional[Dict[str, str]],
+    ):
+        res_arns = []
+        res_arns.append(resource_arn)
+        self.name = name
+        self.arn = f"arn:aws:logs:{region}:{account_id}:delivery-source:{name}"
+        self.resource_arns = res_arns
+        self.service = resource_arn.split(":")[2]
+        self.log_type = log_type
+        self.tags = tags
+
+    def to_dict(self) -> Dict[str, Any]:
+        dct = {
+            "name": self.name,
+            "arn": self.arn,
+            "resourceArns": self.resource_arns,
+            "service": self.service,
+            "logType": self.log_type,
+            "tags": self.tags,
+        }
+        dct_items = {k: v for k, v in dct.items() if v}
+        return dct_items
+
+
+class Delivery(BaseModel):
+    def __init__(
+        self,
+        account_id: str,
+        region: str,
+        delivery_source_name: str,
+        delivery_destination_arn: str,
+        destination_type: str,
+        record_fields: Optional[List[str]],
+        field_delimiter: Optional[str],
+        s3_delivery_configuration: Optional[Dict[str, Any]],
+        tags: Optional[Dict[str, str]],
+    ):
+        self.id = mock_random.get_random_string(length=16)
+        self.arn = f"arn:aws:logs:{region}:{account_id}:delivery:{self.id}"
+        self.delivery_source_name = delivery_source_name
+        self.delivery_destination_arn = delivery_destination_arn
+        self.destination_type = destination_type
+        # Default record fields
+        default_record_fields = [
+            "date",
+            "time",
+            "x-edge-location",
+            "sc-bytes",
+            "c-ip",
+            "cs-method",
+            "cs(Host)",
+            "cs-uri-stem",
+            "sc-status",
+            "cs(Referer)",
+            "cs(User-Agent)",
+            "cs-uri-query",
+            "cs(Cookie)",
+            "x-edge-result-type",
+            "x-edge-request-id",
+            "x-host-header",
+            "cs-protocol",
+            "cs-bytes",
+            "time-taken",
+            "x-forwarded-for",
+            "ssl-protocol",
+            "ssl-cipher",
+            "x-edge-response-result-type",
+            "cs-protocol-version",
+            "fle-status",
+            "fle-encrypted-fields",
+            "c-port",
+            "time-to-first-byte",
+            "x-edge-detailed-result-type",
+            "sc-content-type",
+            "sc-content-len",
+            "sc-range-start",
+            "sc-range-end",
+        ]
+        self.record_fields = record_fields or default_record_fields
+        self.field_delimiter = field_delimiter
+        default_s3_configuration = {}
+        if destination_type == "S3":
+            # Default s3 configuration
+            default_s3_configuration = {
+                "suffixPath": "AWSLogs/{account-id}/CloudFront/",
+                "enableHiveCompatiblePath": False,
+            }
+        self.s3_delivery_configuration = (
+            s3_delivery_configuration or default_s3_configuration
+        )
+        self.tags = tags
+
+    def to_dict(self) -> Dict[str, Any]:
+        dct = {
+            "id": self.id,
+            "arn": self.arn,
+            "deliverySourceName": self.delivery_source_name,
+            "deliveryDestinationArn": self.delivery_destination_arn,
+            "deliveryDestinationType": self.destination_type,
+            "recordFields": self.record_fields,
+            "fieldDelimiter": self.field_delimiter,
+            "s3DeliveryConfiguration": self.s3_delivery_configuration,
+            "tags": self.tags,
+        }
+        dct_items = {k: v for k, v in dct.items() if v}
+        return dct_items
+
+
 class LogsBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
@@ -770,6 +928,9 @@ class LogsBackend(BaseBackend):
         self.destinations: Dict[str, Destination] = dict()
         self.tagger = TaggingService()
         self.export_tasks: Dict[str, ExportTask] = dict()
+        self.delivery_destinations: Dict[str, DeliveryDestination] = dict()
+        self.delivery_sources: Dict[str, DeliverySource] = dict()
+        self.deliveries: Dict[str, Delivery] = dict()
 
     def create_log_group(
         self, log_group_name: str, tags: Dict[str, str], **kwargs: Any
@@ -1346,5 +1507,253 @@ class LogsBackend(BaseBackend):
             raise ResourceNotFoundException()
         return log_group
 
+    def put_delivery_destination(
+        self,
+        name: str,
+        output_format: Optional[str],
+        delivery_destination_configuration: Dict[str, str],
+        tags: Optional[Dict[str, str]],
+    ) -> DeliveryDestination:
+        if output_format and output_format not in [
+            "w3c",
+            "raw",
+            "json",
+            "plain",
+            "parquet",
+        ]:
+            msg = f"1 validation error detected: Value '{output_format}' at 'outputFormat' failed to satisfy constraint: Member must satisfy enum value set: [w3c, raw, json, plain, parquet]"
+            raise ValidationException(msg)
+        if name in self.delivery_destinations:
+            if tags:
+                raise ConflictException(
+                    msg="Tags can only be provided when a resource is being created, not updated."
+                )
+            if (
+                output_format
+                and self.delivery_destinations[name].output_format != output_format
+            ):
+                msg = "Update to existing Delivery Destination with new Output Format is not allowed. Please create a new Delivery Destination instead."
+                raise ValidationException(msg)
+        delivery_destination = DeliveryDestination(
+            account_id=self.account_id,
+            region=self.region_name,
+            name=name,
+            output_format=output_format,
+            delivery_destination_configuration=delivery_destination_configuration,
+            tags=tags,
+            policy=None,
+        )
+        self.delivery_destinations[name] = delivery_destination
+        if tags:
+            self.tag_resource(delivery_destination.arn, tags)
+        return delivery_destination
+
+    def get_delivery_destination(self, name: str) -> DeliveryDestination:
+        if name not in self.delivery_destinations:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Destination does not exist in this account."
+            )
+        delivery_destination = self.delivery_destinations[name]
+        return delivery_destination
+
+    def describe_delivery_destinations(self) -> List[DeliveryDestination]:
+        # Pagination not yet implemented
+        delivery_destinations = list(self.delivery_destinations.values())
+        return delivery_destinations
+
+    def put_delivery_destination_policy(
+        self, delivery_destination_name: str, delivery_destination_policy: str
+    ) -> Dict[str, str]:
+        if delivery_destination_name not in self.delivery_destinations:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Destination does not exist in this account."
+            )
+        dd = self.delivery_destinations[delivery_destination_name]
+        dd.policy = delivery_destination_policy
+        return {"deliveryDestinationPolicy": delivery_destination_policy}
+
+    def get_delivery_destination_policy(
+        self, delivery_destination_name: str
+    ) -> Dict[str, Any]:
+        if delivery_destination_name not in self.delivery_destinations:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Destination does not exist in this account."
+            )
+        policy = self.delivery_destinations[delivery_destination_name].policy
+        return {"deliveryDestinationPolicy": policy}
+
+    def put_delivery_source(
+        self, name: str, resource_arn: str, log_type: str, tags: Dict[str, str]
+    ) -> DeliverySource:
+        log_types = {
+            "cloudfront": "ACCESS_LOGS",
+            "bedrock": "APPLICATION_LOGS",
+            "codewhisperer": "EVENT_LOGS",
+            "mediapackage": ["EGRESS_ACCESS_LOGS", "INGRESS_ACCESS_LOGS"],
+            "mediatailor": [
+                "AD_DECISION_SERVER_LOGS",
+                "MANIFEST_SERVICE_LOGS",
+                "TRANSCODE_LOGS",
+            ],
+            "sso": "ERROR_LOGS",
+            "qdeveloper": "EVENT_LOGS",
+            "ses": "APPLICATION_LOG",
+            "workmail": [
+                "ACCESS_CONTROL_LOGS",
+                "AUTHENTICATION_LOGS",
+                "WORKMAIL_AVAILABILITY_PROVIDER_LOGS",
+                "WORKMAIL_MAILBOX_ACCESS_LOGS",
+                "WORKMAIL_PERSONAL_ACCESS_TOKEN_LOGS",
+            ],
+        }
+        resource_type = resource_arn.split(":")[2]
+
+        if resource_type not in log_types:
+            raise ResourceNotFoundException(msg="Cannot access provided service.")
+
+        if log_type not in log_types[resource_type]:
+            raise ValidationException(
+                msg=" This service is not allowed for this LogSource."
+            )
+
+        if name in self.delivery_sources:
+            if tags:
+                raise ConflictException(
+                    msg="Tags can only be provided when a resource is being created, not updated."
+                )
+            if resource_arn not in self.delivery_sources[name].resource_arns:
+                raise ConflictException(
+                    msg="Update to existing Delivery Source with new ResourceId is not allowed. Please create a new Delivery Source instead."
+                )
+        delivery_source = DeliverySource(
+            account_id=self.account_id,
+            region=self.region_name,
+            name=name,
+            resource_arn=resource_arn,
+            log_type=log_type,
+            tags=tags,
+        )
+        self.delivery_sources[name] = delivery_source
+        if tags:
+            self.tag_resource(delivery_source.arn, tags)
+        return delivery_source
+
+    def describe_delivery_sources(self) -> List[DeliverySource]:
+        # Pagination not yet implemented
+        delivery_sources = list(self.delivery_sources.values())
+        return delivery_sources
+
+    def get_delivery_source(self, name: str) -> DeliverySource:
+        if name not in self.delivery_sources:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Source does not exist in this account.."
+            )
+        delivery_source = self.delivery_sources[name]
+        return delivery_source
+
+    def create_delivery(
+        self,
+        delivery_source_name: str,
+        delivery_destination_arn: str,
+        record_fields: Optional[List[str]],
+        field_delimiter: Optional[str],
+        s3_delivery_configuration: Optional[Dict[str, Any]],
+        tags: Optional[Dict[str, str]],
+    ) -> Delivery:
+        if delivery_source_name not in self.delivery_sources:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Source does not exist in this account."
+            )
+        if delivery_destination_arn not in [
+            dd.arn for dd in self.delivery_destinations.values()
+        ]:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Destination does not exist in this account."
+            )
+
+        for delivery in self.deliveries.values():
+            if (
+                delivery.delivery_source_name == delivery_source_name
+                and delivery.delivery_destination_arn == delivery_destination_arn
+            ):
+                raise ConflictException(msg="The specified Delivery already exists")
+            if delivery.delivery_source_name == delivery_source_name:
+                for dd in self.delivery_destinations.values():
+                    if (
+                        dd.arn == delivery_destination_arn
+                        and delivery.destination_type == dd.delivery_destination_type
+                    ):
+                        raise ConflictException(
+                            msg="Delivery already exists for this Delivery Source with the same Delivery Destination Type."
+                        )
+
+        for dd in list(self.delivery_destinations.values()):
+            if dd.arn == delivery_destination_arn:
+                destination_type = dd.delivery_destination_type
+
+        delivery = Delivery(
+            account_id=self.account_id,
+            region=self.region_name,
+            delivery_source_name=delivery_source_name,
+            delivery_destination_arn=delivery_destination_arn,
+            destination_type=destination_type,
+            record_fields=record_fields,
+            field_delimiter=field_delimiter,
+            s3_delivery_configuration=s3_delivery_configuration,
+            tags=tags,
+        )
+        self.deliveries[delivery.id] = delivery
+        if tags:
+            self.tag_resource(delivery.arn, tags)
+        return delivery
+
+    def describe_deliveries(self) -> List[Delivery]:
+        # Pagination not yet implemented
+        deliveries = list(self.deliveries.values())
+        return deliveries
+
+    def get_delivery(self, id: str) -> Delivery:
+        if id not in self.deliveries:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery does not exist in this account."
+            )
+        delivery = self.deliveries[id]
+        return delivery
+
+    def delete_delivery(self, id: str) -> None:
+        if id not in self.deliveries:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery does not exist in this account."
+            )
+        self.deliveries.pop(id)
+        return
+
+    def delete_delivery_destination(self, name: str) -> None:
+        if name not in self.delivery_destinations:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Destination does not exist in this account."
+            )
+        self.delivery_destinations.pop(name)
+        return
+
+    def delete_delivery_destination_policy(
+        self, delivery_destination_name: str
+    ) -> None:
+        delivery_destination = self.delivery_destinations.get(delivery_destination_name)
+        if not delivery_destination:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Destination does not exist in this account."
+            )
+        delivery_destination.policy = None
+        return
+
+    def delete_delivery_source(self, name: str) -> None:
+        if name not in self.delivery_sources:
+            raise ResourceNotFoundException(
+                msg="Requested Delivery Source does not exist in this account."
+            )
+        self.delivery_sources.pop(name)
+        return
+
 
 logs_backends = BackendDict(LogsBackend, "logs")
diff --git a/moto/logs/responses.py b/moto/logs/responses.py
index b6522f4d7..e374ebd62 100644
--- a/moto/logs/responses.py
+++ b/moto/logs/responses.py
@@ -474,3 +474,129 @@ class LogsResponse(BaseResponse):
         tag_keys = self._get_param("tagKeys")
         self.logs_backend.untag_resource(resource_arn, tag_keys)
         return "{}"
+
+    def put_delivery_destination(self) -> str:
+        name = self._get_param("name")
+        output_format = self._get_param("outputFormat")
+        delivery_destination_configuration = self._get_param(
+            "deliveryDestinationConfiguration"
+        )
+        tags = self._get_param("tags")
+        delivery_destination = self.logs_backend.put_delivery_destination(
+            name=name,
+            output_format=output_format,
+            delivery_destination_configuration=delivery_destination_configuration,
+            tags=tags,
+        )
+        return json.dumps(dict(deliveryDestination=delivery_destination.to_dict()))
+
+    def get_delivery_destination(self) -> str:
+        name = self._get_param("name")
+        delivery_destination = self.logs_backend.get_delivery_destination(
+            name=name,
+        )
+        return json.dumps(dict(deliveryDestination=delivery_destination.to_dict()))
+
+    def describe_delivery_destinations(self) -> str:
+        delivery_destinations = self.logs_backend.describe_delivery_destinations()
+        return json.dumps(
+            dict(deliveryDestinations=[dd.to_dict() for dd in delivery_destinations])
+        )
+
+    def put_delivery_destination_policy(self) -> str:
+        delivery_destination_name = self._get_param("deliveryDestinationName")
+        delivery_destination_policy = self._get_param("deliveryDestinationPolicy")
+        policy = self.logs_backend.put_delivery_destination_policy(
+            delivery_destination_name=delivery_destination_name,
+            delivery_destination_policy=delivery_destination_policy,
+        )
+        return json.dumps(dict(policy=policy))
+
+    def get_delivery_destination_policy(self) -> str:
+        delivery_destination_name = self._get_param("deliveryDestinationName")
+        policy = self.logs_backend.get_delivery_destination_policy(
+            delivery_destination_name=delivery_destination_name,
+        )
+        return json.dumps(dict(policy=policy))
+
+    def put_delivery_source(self) -> str:
+        name = self._get_param("name")
+        resource_arn = self._get_param("resourceArn")
+        log_type = self._get_param("logType")
+        tags = self._get_param("tags")
+        delivery_source = self.logs_backend.put_delivery_source(
+            name=name,
+            resource_arn=resource_arn,
+            log_type=log_type,
+            tags=tags,
+        )
+        return json.dumps(dict(deliverySource=delivery_source.to_dict()))
+
+    def describe_delivery_sources(self) -> str:
+        delivery_sources = self.logs_backend.describe_delivery_sources()
+        return json.dumps(
+            dict(deliverySources=[ds.to_dict() for ds in delivery_sources])
+        )
+
+    def get_delivery_source(self) -> str:
+        name = self._get_param("name")
+        delivery_source = self.logs_backend.get_delivery_source(
+            name=name,
+        )
+        return json.dumps(dict(deliverySource=delivery_source.to_dict()))
+
+    def create_delivery(self) -> str:
+        delivery_source_name = self._get_param("deliverySourceName")
+        delivery_destination_arn = self._get_param("deliveryDestinationArn")
+        record_fields = self._get_param("recordFields")
+        field_delimiter = self._get_param("fieldDelimiter")
+        s3_delivery_configuration = self._get_param("s3DeliveryConfiguration")
+        tags = self._get_param("tags")
+        delivery = self.logs_backend.create_delivery(
+            delivery_source_name=delivery_source_name,
+            delivery_destination_arn=delivery_destination_arn,
+            record_fields=record_fields,
+            field_delimiter=field_delimiter,
+            s3_delivery_configuration=s3_delivery_configuration,
+            tags=tags,
+        )
+        return json.dumps(dict(delivery=delivery.to_dict()))
+
+    def describe_deliveries(self) -> str:
+        deliveries = self.logs_backend.describe_deliveries()
+        return json.dumps(dict(deliveries=[d.to_dict() for d in deliveries]))
+
+    def get_delivery(self) -> str:
+        id = self._get_param("id")
+        delivery = self.logs_backend.get_delivery(
+            id=id,
+        )
+        return json.dumps(dict(delivery=delivery.to_dict()))
+
+    def delete_delivery(self) -> str:
+        id = self._get_param("id")
+        self.logs_backend.delete_delivery(
+            id=id,
+        )
+        return ""
+
+    def delete_delivery_destination(self) -> str:
+        name = self._get_param("name")
+        self.logs_backend.delete_delivery_destination(
+            name=name,
+        )
+        return ""
+
+    def delete_delivery_destination_policy(self) -> str:
+        delivery_destination_name = self._get_param("deliveryDestinationName")
+        self.logs_backend.delete_delivery_destination_policy(
+            delivery_destination_name=delivery_destination_name,
+        )
+        return ""
+
+    def delete_delivery_source(self) -> str:
+        name = self._get_param("name")
+        self.logs_backend.delete_delivery_source(
+            name=name,
+        )
+        return ""


Test patch:
diff --git a/tests/test_cloudhsmv2/__init__.py b/tests/test_cloudhsmv2/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_cloudhsmv2/test_cloudhsmv2.py b/tests/test_cloudhsmv2/test_cloudhsmv2.py
new file mode 100644
index 000000000..16cb4d8be
--- /dev/null
+++ b/tests/test_cloudhsmv2/test_cloudhsmv2.py
@@ -0,0 +1,317 @@
+"""Unit tests for cloudhsmv2-supported APIs."""
+
+import json
+from datetime import datetime
+
+import boto3
+
+from moto import mock_aws
+
+# See our Development Tips on writing tests for hints on how to write good tests:
+# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
+
+
+@mock_aws
+def test_list_tags():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    resource_id = "cluster-1234"
+    client.tag_resource(
+        ResourceId=resource_id,
+        TagList=[
+            {"Key": "Environment", "Value": "Production"},
+            {"Key": "Project", "Value": "Security"},
+        ],
+    )
+
+    response = client.list_tags(ResourceId=resource_id)
+    assert len(response["TagList"]) == 2
+    assert {"Key": "Environment", "Value": "Production"} in response["TagList"]
+    assert {"Key": "Project", "Value": "Security"} in response["TagList"]
+    assert "NextToken" not in response
+
+
+@mock_aws
+def test_tag_resource():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+    resource_id = "cluster-1234"
+
+    response = client.tag_resource(
+        ResourceId=resource_id,
+        TagList=[
+            {"Key": "Environment", "Value": "Production"},
+            {"Key": "Project", "Value": "Security"},
+        ],
+    )
+
+    tags = client.list_tags(ResourceId=resource_id)["TagList"]
+    assert len(tags) == 2
+    assert {"Key": "Environment", "Value": "Production"} in tags
+    assert {"Key": "Project", "Value": "Security"} in tags
+
+    response = client.tag_resource(
+        ResourceId=resource_id,
+        TagList=[{"Key": "Environment", "Value": "Development"}],
+    )
+
+    assert "ResponseMetadata" in response
+
+    tags = client.list_tags(ResourceId=resource_id)["TagList"]
+    assert len(tags) == 2
+    assert {"Key": "Environment", "Value": "Development"} in tags
+    assert {"Key": "Project", "Value": "Security"} in tags
+
+
+@mock_aws
+def test_list_tags_empty_resource():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    response = client.list_tags(ResourceId="non-existent-resource")
+    assert response["TagList"] == []
+    assert "NextToken" not in response
+
+
+@mock_aws
+def test_untag_resource():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+    resource_id = "cluster-1234"
+
+    client.tag_resource(
+        ResourceId=resource_id,
+        TagList=[
+            {"Key": "Environment", "Value": "Production"},
+            {"Key": "Project", "Value": "Security"},
+            {"Key": "Team", "Value": "DevOps"},
+        ],
+    )
+
+    initial_tags = client.list_tags(ResourceId=resource_id)["TagList"]
+    assert len(initial_tags) == 3
+
+    response = client.untag_resource(
+        ResourceId=resource_id, TagKeyList=["Environment", "Team"]
+    )
+    assert "ResponseMetadata" in response
+
+    remaining_tags = client.list_tags(ResourceId=resource_id)["TagList"]
+    assert len(remaining_tags) == 1
+    assert {"Key": "Project", "Value": "Security"} in remaining_tags
+
+
+@mock_aws
+def test_create_cluster():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    response = client.create_cluster(
+        BackupRetentionPolicy={"Type": "DAYS", "Value": "7"},
+        HsmType="hsm1.medium",
+        SubnetIds=["subnet-12345678"],
+        TagList=[{"Key": "Environment", "Value": "Production"}],
+    )
+
+    cluster = response["Cluster"]
+    assert cluster["BackupPolicy"] == "DEFAULT"
+    assert cluster["BackupRetentionPolicy"] == {"Type": "DAYS", "Value": "7"}
+    assert "ClusterId" in cluster
+    assert isinstance(cluster["CreateTimestamp"], datetime)
+    assert cluster["HsmType"] == "hsm1.medium"
+    assert cluster["State"] == "ACTIVE"
+    assert cluster["SubnetMapping"] == {"subnet-12345678": "us-east-1"}
+    assert cluster["TagList"] == [{"Key": "Environment", "Value": "Production"}]
+    assert "VpcId" in cluster
+
+    clusters = client.describe_clusters()["Clusters"]
+    assert len(clusters) == 1
+    assert clusters[0]["ClusterId"] == cluster["ClusterId"]
+
+
+@mock_aws
+def test_delete_cluster():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    response = client.create_cluster(
+        HsmType="hsm1.medium",
+        SubnetIds=["subnet-12345678"],
+        NetworkType="IPV4",
+        Mode="FIPS",
+    )
+    cluster_id = response["Cluster"]["ClusterId"]
+
+    delete_response = client.delete_cluster(ClusterId=cluster_id)
+
+    deleted_cluster = delete_response["Cluster"]
+    assert deleted_cluster["ClusterId"] == cluster_id
+    assert deleted_cluster["State"] == "DELETED"
+    assert deleted_cluster["StateMessage"] == "Cluster deleted"
+
+    clusters = client.describe_clusters()["Clusters"]
+    assert len(clusters) == 0
+
+
+@mock_aws
+def test_describe_clusters_no_clusters():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+    response = client.describe_clusters()
+
+    assert response["Clusters"] == []
+    assert "NextToken" not in response
+
+
+@mock_aws
+def test_describe_clusters_with_filters():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    cluster1 = client.create_cluster(
+        HsmType="hsm1.medium",
+        SubnetIds=["subnet-12345678"],
+        NetworkType="IPV4",
+        Mode="FIPS",
+    )
+
+    client.create_cluster(
+        HsmType="hsm1.medium",
+        SubnetIds=["subnet-87654321"],
+        NetworkType="IPV4",
+        Mode="FIPS",
+    )
+
+    response = client.describe_clusters(
+        Filters={"clusterIds": [cluster1["Cluster"]["ClusterId"]]}
+    )
+    assert len(response["Clusters"]) == 1
+    assert response["Clusters"][0]["ClusterId"] == cluster1["Cluster"]["ClusterId"]
+
+    # Test filtering by state
+    response = client.describe_clusters(Filters={"states": ["ACTIVE"]})
+    assert len(response["Clusters"]) == 2
+
+
+@mock_aws
+def test_describe_clusters_pagination():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    for i in range(3):
+        client.create_cluster(
+            HsmType="hsm1.medium",
+            SubnetIds=[f"subnet-{i}"],
+            NetworkType="IPV4",
+            Mode="FIPS",
+        )
+
+    response = client.describe_clusters(MaxResults=2)
+    assert len(response["Clusters"]) == 2
+    assert "NextToken" in response
+
+    # Get remaining clusters using NextToken
+    response2 = client.describe_clusters(MaxResults=2, NextToken=response["NextToken"])
+    assert len(response2["Clusters"]) == 1
+    assert "NextToken" not in response2
+
+
+@mock_aws
+def test_get_resource_policy():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    client.create_cluster(HsmType="hsm1.medium", SubnetIds=["subnet-12345678"])
+
+    backup_response = client.describe_backups()
+    backup_arn = backup_response["Backups"][0]["BackupArn"]
+
+    policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "EnableSharing",
+                "Effect": "Allow",
+                "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
+                "Action": ["cloudhsmv2:DescribeBackups"],
+                "Resource": backup_arn,
+            }
+        ],
+    }
+
+    client.put_resource_policy(ResourceArn=backup_arn, Policy=json.dumps(policy))
+    response = client.get_resource_policy(ResourceArn=backup_arn)
+
+    assert "Policy" in response
+    assert json.loads(response["Policy"]) == policy
+
+
+@mock_aws
+def test_describe_backups():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    cluster = client.create_cluster(
+        HsmType="hsm1.medium",
+        SubnetIds=["subnet-12345678"],
+    )
+    cluster_id = cluster["Cluster"]["ClusterId"]
+
+    response = client.describe_backups()
+    assert "Backups" in response
+    assert len(response["Backups"]) == 1
+
+    backup = response["Backups"][0]
+    assert backup["ClusterId"] == cluster_id
+    assert backup["HsmType"] == "hsm1.medium"
+    assert backup["BackupState"] == "READY"
+
+    filtered_response = client.describe_backups(Filters={"clusterIds": [cluster_id]})
+    assert len(filtered_response["Backups"]) == 1
+    assert filtered_response["Backups"][0]["ClusterId"] == cluster_id
+
+
+@mock_aws
+def test_put_resource_policy():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    client.create_cluster(HsmType="hsm1.medium", SubnetIds=["subnet-12345678"])
+
+    backup_response = client.describe_backups()
+    backup_arn = backup_response["Backups"][0]["BackupArn"]
+
+    policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "EnableSharing",
+                "Effect": "Allow",
+                "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
+                "Action": ["cloudhsmv2:DescribeBackups"],
+                "Resource": backup_arn,
+            }
+        ],
+    }
+
+    response = client.put_resource_policy(
+        ResourceArn=backup_arn, Policy=json.dumps(policy)
+    )
+
+    assert "ResourceArn" in response
+    assert "Policy" in response
+    assert response["ResourceArn"] == backup_arn
+    assert json.loads(response["Policy"]) == policy
+
+
+@mock_aws
+def test_describe_backups_pagination():
+    client = boto3.client("cloudhsmv2", region_name="us-east-1")
+
+    # Create a cluster which will automatically create a backup
+    for i in range(3):
+        client.create_cluster(
+            HsmType="hsm1.medium",
+            SubnetIds=[f"subnet-{i}"],
+            NetworkType="IPV4",
+            Mode="FIPS",
+        )
+
+    # Test pagination with MaxResults
+    response = client.describe_backups(MaxResults=2)
+    assert len(response["Backups"]) == 2
+    assert "NextToken" in response
+
+    # Get remaining backups using NextToken
+    response2 = client.describe_backups(MaxResults=2, NextToken=response["NextToken"])
+    assert len(response2["Backups"]) == 1
+    assert "NextToken" not in response2
diff --git a/tests/test_logs/test_logs.py b/tests/test_logs/test_logs.py
index cec52981c..d686fe6ea 100644
--- a/tests/test_logs/test_logs.py
+++ b/tests/test_logs/test_logs.py
@@ -48,6 +48,25 @@ access_policy_doc = json.dumps(
     }
 )
 
+delivery_destination_policy = json.dumps(
+    {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "AllowLogDeliveryActions",
+                "Effect": "Allow",
+                "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
+                "Action": "logs:CreateDelivery",
+                "Resource": [
+                    f"arn:aws:logs:{TEST_REGION}:123456789012:delivery-source:*",
+                    f"arn:aws:logs:{TEST_REGION}:123456789012:delivery:*",
+                    f"arn:aws:logs:{TEST_REGION}:123456789012:delivery-destination:*",
+                ],
+            }
+        ],
+    }
+)
+
 
 @pytest.fixture(name="log_group_name")
 def create_log_group():
@@ -1165,3 +1184,475 @@ def test_describe_log_streams_no_prefix():
     err = ex.value.response["Error"]
     assert err["Code"] == "InvalidParameterException"
     assert err["Message"] == "Cannot order by LastEventTime with a logStreamNamePrefix."
+
+
+@mock_aws
+def test_put_delivery_destination():
+    client = boto3.client("logs", "us-east-1")
+    resp = client.put_delivery_destination(
+        name="test-delivery-destination",
+        outputFormat="json",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+        tags={"key1": "value1"},
+    )
+    delivery_destination = resp["deliveryDestination"]
+    assert delivery_destination["name"] == "test-delivery-destination"
+    assert delivery_destination["outputFormat"] == "json"
+    assert delivery_destination["deliveryDestinationConfiguration"] == {
+        "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+    }
+    assert delivery_destination["tags"] == {"key1": "value1"}
+
+    # Invalid OutputFormat
+    with pytest.raises(ClientError) as ex:
+        client.put_delivery_destination(
+            name="test-dd",
+            outputFormat="foobar",
+            deliveryDestinationConfiguration={
+                "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+            },
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ValidationException"
+
+    # Cannot update OutoutFormat
+    with pytest.raises(ClientError) as ex:
+        client.put_delivery_destination(
+            name="test-delivery-destination",
+            outputFormat="plain",
+            deliveryDestinationConfiguration={
+                "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+            },
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ValidationException"
+
+
+@mock_aws
+def test_put_delivery_destination_update():
+    client = boto3.client("logs", "us-east-1")
+    client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    # Update destination resource
+    resp = client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket-2"
+        },
+    )
+    delivery_destination = resp["deliveryDestination"]
+    assert delivery_destination["deliveryDestinationConfiguration"] == {
+        "destinationResourceArn": "arn:aws:s3:::test-s3-bucket-2"
+    }
+
+
+@mock_aws
+def test_get_delivery_destination():
+    client = boto3.client("logs", "us-east-1")
+    for i in range(1, 3):
+        client.put_delivery_destination(
+            name=f"test-delivery-destination-{i}",
+            deliveryDestinationConfiguration={
+                "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+            },
+        )
+    resp = client.get_delivery_destination(name="test-delivery-destination-1")
+    assert "deliveryDestination" in resp
+    assert resp["deliveryDestination"]["name"] == "test-delivery-destination-1"
+
+    # Invalid name for delivery destination
+    with pytest.raises(ClientError) as ex:
+        client.get_delivery_destination(
+            name="foobar",
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_describe_delivery_destinations():
+    client = boto3.client("logs", "us-east-1")
+    for i in range(1, 3):
+        client.put_delivery_destination(
+            name=f"test-delivery-destination-{i}",
+            deliveryDestinationConfiguration={
+                "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+            },
+        )
+    resp = client.describe_delivery_destinations()
+    assert len(resp["deliveryDestinations"]) == 2
+
+
+@mock_aws
+def test_put_delivery_destination_policy():
+    client = boto3.client("logs", "us-east-1")
+    client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    resp = client.put_delivery_destination_policy(
+        deliveryDestinationName="test-delivery-destination",
+        deliveryDestinationPolicy=delivery_destination_policy,
+    )
+    assert "policy" in resp
+
+    # Invalid name for destination policy
+    with pytest.raises(ClientError) as ex:
+        client.put_delivery_destination_policy(
+            deliveryDestinationName="foobar",
+            deliveryDestinationPolicy=delivery_destination_policy,
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_get_delivery_destination_policy():
+    client = boto3.client("logs", "us-east-1")
+    client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    client.put_delivery_destination_policy(
+        deliveryDestinationName="test-delivery-destination",
+        deliveryDestinationPolicy=delivery_destination_policy,
+    )
+    resp = client.get_delivery_destination_policy(
+        deliveryDestinationName="test-delivery-destination"
+    )
+    assert "deliveryDestinationPolicy" in resp["policy"]
+
+    #  Invalide name for destination policy
+    with pytest.raises(ClientError) as ex:
+        client.get_delivery_destination_policy(
+            deliveryDestinationName="foobar",
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_put_delivery_source():
+    client = boto3.client("logs", "us-east-1")
+    resp = client.put_delivery_source(
+        name="test-delivery-source",
+        resourceArn="arn:aws:cloudfront::123456789012:distribution/E1Q5F5862X9VJ5",
+        logType="ACCESS_LOGS",
+        tags={"key1": "value1"},
+    )
+    assert "deliverySource" in resp
+    assert "name" in resp["deliverySource"]
+    assert "arn" in resp["deliverySource"]
+    assert "resourceArns" in resp["deliverySource"]
+    assert "service" in resp["deliverySource"]
+    assert "logType" in resp["deliverySource"]
+    assert "tags" in resp["deliverySource"]
+
+    # Invalid resource source.
+    with pytest.raises(ClientError) as ex:
+        client.put_delivery_source(
+            name="test-ds",
+            resourceArn="arn:aws:s3:::test-s3-bucket",  # S3 cannot be a source
+            logType="ACCESS_LOGS",
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+    # Invalid Log type
+    with pytest.raises(ClientError) as ex:
+        client.put_delivery_source(
+            name="test-ds",
+            resourceArn="arn:aws:cloudfront::123456789012:distribution/E1Q5F5862X9VJ5",
+            logType="EVENT_LOGS",
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ValidationException"
+
+    # Cannot update resource source with a differen resourceArn
+    with pytest.raises(ClientError) as ex:
+        client.put_delivery_source(
+            name="test-delivery-source",
+            resourceArn="arn:aws:cloudfront::123456789012:distribution/E19DL18TOXN9JU",
+            logType="ACCESS_LOGS",
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ConflictException"
+
+
+@mock_aws
+def test_describe_delivery_sources():
+    client = boto3.client("logs", "us-east-1")
+    for i in range(1, 3):
+        client.put_delivery_source(
+            name=f"test-delivery-source-{i}",
+            resourceArn="arn:aws:cloudfront::123456789012:distribution/E19DL18TOXN9JU",
+            logType="ACCESS_LOGS",
+        )
+    resp = client.describe_delivery_sources()
+    assert len(resp["deliverySources"]) == 2
+
+
+@mock_aws
+def test_get_delivery_source():
+    client = boto3.client("logs", "us-east-1")
+    for i in range(1, 3):
+        client.put_delivery_source(
+            name=f"test-delivery-source-{i}",
+            resourceArn="arn:aws:cloudfront::123456789012:distribution/E19DL18TOXN9JU",
+            logType="ACCESS_LOGS",
+        )
+    resp = client.get_delivery_source(name="test-delivery-source-1")
+    assert "deliverySource" in resp
+    assert resp["deliverySource"]["name"] == "test-delivery-source-1"
+
+    # Invalid name for delivery source
+    with pytest.raises(ClientError) as ex:
+        client.get_delivery_source(
+            name="foobar",
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_create_delivery():
+    client = boto3.client("logs", "us-east-1")
+    client.put_delivery_source(
+        name="test-delivery-source",
+        resourceArn="arn:aws:cloudfront::123456789012:distribution/E19DL18TOXN9JU",
+        logType="ACCESS_LOGS",
+    )
+    client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    resp = client.create_delivery(
+        deliverySourceName="test-delivery-source",
+        deliveryDestinationArn="arn:aws:logs:us-east-1:123456789012:delivery-destination:test-delivery-destination",
+        recordFields=[
+            "date",
+        ],
+        fieldDelimiter=",",
+        s3DeliveryConfiguration={
+            "suffixPath": "AWSLogs/123456789012/CloudFront/",
+            "enableHiveCompatiblePath": True,
+        },
+        tags={"key1": "value1"},
+    )
+    assert "delivery" in resp
+    assert "id" in resp["delivery"]
+    assert "arn" in resp["delivery"]
+    assert "deliverySourceName" in resp["delivery"]
+    assert "deliveryDestinationArn" in resp["delivery"]
+    assert "deliveryDestinationType" in resp["delivery"]
+    assert "recordFields" in resp["delivery"]
+    assert "fieldDelimiter" in resp["delivery"]
+    assert "s3DeliveryConfiguration" in resp["delivery"]
+    assert "tags" in resp["delivery"]
+
+    # Invalid delivery source
+    with pytest.raises(ClientError) as ex:
+        client.create_delivery(
+            deliverySourceName="foobar",
+            deliveryDestinationArn="arn:aws:logs:us-east-1:123456789012:delivery-destination:test-delivery-destination",
+        )
+    err = ex.value.response["Error"]
+
+    # Invalid Delivery destination
+    with pytest.raises(ClientError) as ex:
+        client.create_delivery(
+            deliverySourceName="test-delivery-source",
+            deliveryDestinationArn="arn:aws:logs:us-east-1:123456789012:delivery-destination:foobar",
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+    # Delivery already exists
+    with pytest.raises(ClientError) as ex:
+        client.create_delivery(
+            deliverySourceName="test-delivery-source",
+            deliveryDestinationArn="arn:aws:logs:us-east-1:123456789012:delivery-destination:test-delivery-destination",
+        )
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ConflictException"
+
+
+@mock_aws
+def test_describe_deliveries():
+    client = boto3.client("logs", "us-east-1")
+    client.put_delivery_source(
+        name="test-delivery-source",
+        resourceArn="arn:aws:cloudfront::123456789012:distribution/E19DL18TOXN9JU",
+        logType="ACCESS_LOGS",
+    )
+    client.put_delivery_destination(
+        name="test-delivery-destination-1",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    client.put_delivery_destination(
+        name="test-delivery-destination-2",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:firehose:us-east-1:123456789012:deliverystream/test-delivery-stream"
+        },
+    )
+    for i in range(1, 3):
+        client.create_delivery(
+            deliverySourceName="test-delivery-source",
+            deliveryDestinationArn=f"arn:aws:logs:us-east-1:123456789012:delivery-destination:test-delivery-destination-{i}",
+        )
+    resp = client.describe_deliveries()
+    assert len(resp["deliveries"]) == 2
+
+
+@mock_aws
+def test_get_delivery():
+    client = boto3.client("logs", "us-east-1")
+    client.put_delivery_source(
+        name="test-delivery-source",
+        resourceArn="arn:aws:cloudfront::123456789012:distribution/E19DL18TOXN9JU",
+        logType="ACCESS_LOGS",
+    )
+    client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    delivery = client.create_delivery(
+        deliverySourceName="test-delivery-source",
+        deliveryDestinationArn="arn:aws:logs:us-east-1:123456789012:delivery-destination:test-delivery-destination",
+    )
+    delivery_id = delivery["delivery"]["id"]
+    resp = client.get_delivery(id=delivery_id)
+    assert "delivery" in resp
+    assert resp["delivery"]["id"] == delivery_id
+
+    # Invalid delivery id
+    with pytest.raises(ClientError) as ex:
+        client.get_delivery(id="foobar")
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_delete_delivery():
+    client = boto3.client("logs", "us-east-1")
+    client.put_delivery_source(
+        name="test-delivery-source",
+        resourceArn="arn:aws:cloudfront::123456789012:distribution/E19DL18TOXN9JU",
+        logType="ACCESS_LOGS",
+    )
+    client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    delivery = client.create_delivery(
+        deliverySourceName="test-delivery-source",
+        deliveryDestinationArn="arn:aws:logs:us-east-1:123456789012:delivery-destination:test-delivery-destination",
+    )
+    delivery_id = delivery["delivery"]["id"]
+    resp = client.describe_deliveries()
+    assert len(resp["deliveries"]) == 1
+    client.delete_delivery(id=delivery_id)
+    resp = client.describe_deliveries()
+    assert len(resp["deliveries"]) == 0
+
+    # invalid delivery id
+    with pytest.raises(ClientError) as ex:
+        client.delete_delivery(id="foobar")
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_delete_delivery_destination():
+    client = boto3.client("logs", "us-east-1")
+    resp = client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    delivery_destination = resp["deliveryDestination"]
+    resp = client.describe_delivery_destinations()
+    assert len(resp["deliveryDestinations"]) == 1
+    resp = client.delete_delivery_destination(name=delivery_destination["name"])
+    resp = client.describe_delivery_destinations()
+    assert len(resp["deliveryDestinations"]) == 0
+
+    # Invalid name for delivery destination
+    with pytest.raises(ClientError) as ex:
+        client.delete_delivery_destination(name="foobar")
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_delete_delivery_destination_policy():
+    client = boto3.client("logs", "us-east-1")
+    client.put_delivery_destination(
+        name="test-delivery-destination",
+        deliveryDestinationConfiguration={
+            "destinationResourceArn": "arn:aws:s3:::test-s3-bucket"
+        },
+    )
+    client.put_delivery_destination_policy(
+        deliveryDestinationName="test-delivery-destination",
+        deliveryDestinationPolicy=delivery_destination_policy,
+    )
+    resp = client.get_delivery_destination_policy(
+        deliveryDestinationName="test-delivery-destination"
+    )
+    policy = resp["policy"]
+    assert "deliveryDestinationPolicy" in policy
+    client.delete_delivery_destination_policy(
+        deliveryDestinationName="test-delivery-destination"
+    )
+    resp = client.get_delivery_destination_policy(
+        deliveryDestinationName="test-delivery-destination"
+    )
+    assert resp["policy"] == {}
+
+    # Invalid name for delivery destination policy
+    with pytest.raises(ClientError) as ex:
+        client.delete_delivery_destination_policy(deliveryDestinationName="test")
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
+@mock_aws
+def test_delete_delivery_source():
+    client = boto3.client("logs", "us-east-1")
+    resp = client.put_delivery_source(
+        name="test-delivery-source",
+        resourceArn="arn:aws:cloudfront::123456789012:distribution/E1Q5F5862X9VJ5",
+        logType="ACCESS_LOGS",
+    )
+    delivery_source = resp["deliverySource"]
+    resp = client.describe_delivery_sources()
+    assert len(resp["deliverySources"]) == 1
+    client.delete_delivery_source(name=delivery_source["name"])
+    resp = client.describe_delivery_sources()
+    assert len(resp["deliverySources"]) == 0
+
+    # Invalid name for delivery source
+    with pytest.raises(ClientError) as ex:
+        client.delete_delivery_source(name="foobar")
+    err = ex.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"