From 791bc77f3ae39172c9ca3ae3a0f17d1fa3869271 Mon Sep 17 00:00:00 2001
From: John Flores <56614456+jnsf-cg@users.noreply.github.com>
Date: Mon, 8 Feb 2021 10:52:47 -0800
Subject: [PATCH] Add KmsKeyId to Redshift Cluster (#3666)

* Add KmsKeyId to Redshift Cluster

Add the KmsKeyId property when creating a cluster so that it is also
returned when querying the describe_clusters endpoint.

* Run black on updated files

* Add unit test for Redshift KmsKeyId

* Re-run black with correct version
---
 moto/redshift/models.py              |  4 ++++
 moto/redshift/responses.py           |  1 +
 tests/test_redshift/test_redshift.py | 26 ++++++++++++++++++++++++++
 3 files changed, 31 insertions(+)

diff --git a/moto/redshift/models.py b/moto/redshift/models.py
index f31de9c96..0574112f1 100644
--- a/moto/redshift/models.py
+++ b/moto/redshift/models.py
@@ -97,6 +97,7 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
         iam_roles_arn=None,
         enhanced_vpc_routing=None,
         restored_from_snapshot=False,
+        kms_key_id=None,
     ):
         super(Cluster, self).__init__(region_name, tags)
         self.redshift_backend = redshift_backend
@@ -159,6 +160,7 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
 
         self.iam_roles_arn = iam_roles_arn or []
         self.restored_from_snapshot = restored_from_snapshot
+        self.kms_key_id = kms_key_id
 
     @staticmethod
     def cloudformation_name_type():
@@ -207,6 +209,7 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
             publicly_accessible=properties.get("PubliclyAccessible"),
             encrypted=properties.get("Encrypted"),
             region_name=region_name,
+            kms_key_id=properties.get("KmsKeyId"),
         )
         return cluster
 
@@ -300,6 +303,7 @@ class Cluster(TaggableResourceMixin, CloudFormationModel):
                 {"ApplyStatus": "in-sync", "IamRoleArn": iam_role_arn}
                 for iam_role_arn in self.iam_roles_arn
             ],
+            "KmsKeyId": self.kms_key_id,
         }
         if self.restored_from_snapshot:
             json_response["RestoreStatus"] = {
diff --git a/moto/redshift/responses.py b/moto/redshift/responses.py
index 1159eb93f..9397d5b4f 100644
--- a/moto/redshift/responses.py
+++ b/moto/redshift/responses.py
@@ -147,6 +147,7 @@ class RedshiftResponse(BaseResponse):
             "tags": self.unpack_complex_list_params("Tags.Tag", ("Key", "Value")),
             "iam_roles_arn": self._get_iam_roles(),
             "enhanced_vpc_routing": self._get_param("EnhancedVpcRouting"),
+            "kms_key_id": self._get_param("KmsKeyId"),
         }
         cluster = self.redshift_backend.create_cluster(**cluster_kwargs).to_json()
         cluster["ClusterStatus"] = "creating"
diff --git a/tests/test_redshift/test_redshift.py b/tests/test_redshift/test_redshift.py
index 1c42c00bd..62613458f 100644
--- a/tests/test_redshift/test_redshift.py
+++ b/tests/test_redshift/test_redshift.py
@@ -41,6 +41,7 @@ def test_create_cluster_boto3():
         datetime.datetime.now(create_time.tzinfo) - datetime.timedelta(minutes=1)
     )
     response["Cluster"]["EnhancedVpcRouting"].should.equal(False)
+    response["Cluster"]["KmsKeyId"].should.equal("")
 
 
 @mock_redshift
@@ -64,6 +65,31 @@ def test_create_cluster_with_enhanced_vpc_routing_enabled():
     response["Cluster"]["EnhancedVpcRouting"].should.equal(True)
 
 
+@mock_redshift
+def test_create_and_describe_cluster_with_kms_key_id():
+    kms_key_id = (
+        "arn:aws:kms:us-east-1:123456789012:key/00000000-0000-0000-0000-000000000000"
+    )
+    client = boto3.client("redshift", region_name="us-east-1")
+    response = client.create_cluster(
+        DBName="test",
+        ClusterIdentifier="test",
+        ClusterType="single-node",
+        NodeType="ds2.xlarge",
+        MasterUsername="user",
+        MasterUserPassword="password",
+        KmsKeyId=kms_key_id,
+    )
+    response["Cluster"]["KmsKeyId"].should.equal(kms_key_id)
+
+    response = client.describe_clusters()
+    clusters = response.get("Clusters", [])
+    len(clusters).should.equal(1)
+
+    cluster = clusters[0]
+    cluster["KmsKeyId"].should.equal(kms_key_id)
+
+
 @mock_redshift
 def test_create_snapshot_copy_grant():
     client = boto3.client("redshift", region_name="us-east-1")