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

@@ -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"] = {

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"

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")