From 55f207050ef133888ac8dd231e3d124953096391 Mon Sep 17 00:00:00 2001
From: Brian Pandola <bpandola@hsdp.io>
Date: Wed, 6 May 2020 14:28:40 -0700
Subject: [PATCH] Add `Redshift.ClusterAlreadyExists` Error

Closes #2967
---
 moto/redshift/exceptions.py          |  7 +++++++
 moto/redshift/models.py              |  3 +++
 tests/test_redshift/test_redshift.py | 22 ++++++++++++++++++++++
 3 files changed, 32 insertions(+)

diff --git a/moto/redshift/exceptions.py b/moto/redshift/exceptions.py
index 0a17e8aab..b5f83d3bc 100644
--- a/moto/redshift/exceptions.py
+++ b/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"
+        )
diff --git a/moto/redshift/models.py b/moto/redshift/models.py
index 17840fb86..07baf18c0 100644
--- a/moto/redshift/models.py
+++ b/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
diff --git a/tests/test_redshift/test_redshift.py b/tests/test_redshift/test_redshift.py
index 6bb3b1396..cf96ee15f 100644
--- a/tests/test_redshift/test_redshift.py
+++ b/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"
+    )