Add Redshift.ClusterAlreadyExists Error

Closes #2967

moto/redshift/exceptions.py

@@ -136,3 +136,10 @@ class SnapshotCopyAlreadyEnabledFaultError(RedshiftClientError):
                 cluster_identifier
             ),
         )
+
+
+class ClusterAlreadyExistsFaultError(RedshiftClientError):
+    def __init__(self):
+        super(ClusterAlreadyExistsFaultError, self).__init__(
+            "ClusterAlreadyExists", "Cluster already exists"
+        )

moto/redshift/models.py

@@ -10,6 +10,7 @@ from moto.core import BaseBackend, BaseModel
 from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.ec2 import ec2_backends
 from .exceptions import (
+    ClusterAlreadyExistsFaultError,
     ClusterNotFoundError,
     ClusterParameterGroupNotFoundError,
     ClusterSecurityGroupNotFoundError,
@@ -580,6 +581,8 @@ class RedshiftBackend(BaseBackend):
 
     def create_cluster(self, **cluster_kwargs):
         cluster_identifier = cluster_kwargs["cluster_identifier"]
+        if cluster_identifier in self.clusters:
+            raise ClusterAlreadyExistsFaultError()
         cluster = Cluster(self, **cluster_kwargs)
         self.clusters[cluster_identifier] = cluster
         return cluster

tests/test_redshift/test_redshift.py

@@ -915,6 +915,11 @@ def test_create_cluster_from_snapshot():
         ClusterIdentifier=original_cluster_identifier,
     )
 
+    client.restore_from_cluster_snapshot.when.called_with(
+        ClusterIdentifier=original_cluster_identifier,
+        SnapshotIdentifier=original_snapshot_identifier,
+    ).should.throw(ClientError, "ClusterAlreadyExists")
+
     response = client.restore_from_cluster_snapshot(
         ClusterIdentifier=new_cluster_identifier,
         SnapshotIdentifier=original_snapshot_identifier,
@@ -1333,3 +1338,20 @@ def test_modify_snapshot_copy_retention_period():
     response = client.describe_clusters(ClusterIdentifier="test")
     cluster_snapshot_copy_status = response["Clusters"][0]["ClusterSnapshotCopyStatus"]
     cluster_snapshot_copy_status["RetentionPeriod"].should.equal(5)
+
+
+@mock_redshift
+def test_create_duplicate_cluster_fails():
+    kwargs = {
+        "ClusterIdentifier": "test",
+        "ClusterType": "single-node",
+        "DBName": "test",
+        "MasterUsername": "user",
+        "MasterUserPassword": "password",
+        "NodeType": "ds2.xlarge",
+    }
+    client = boto3.client("redshift", region_name="us-east-1")
+    client.create_cluster(**kwargs)
+    client.create_cluster.when.called_with(**kwargs).should.throw(
+        ClientError, "ClusterAlreadyExists"
+    )