redshift: add copy grant functionality

moto/redshift/exceptions.py

@@ -58,6 +58,21 @@ class InvalidSubnetError(RedshiftClientError):
             "Subnet {0} not found.".format(subnet_identifier))
 
 
+class SnapshotCopyGrantAlreadyExistsFaultError(RedshiftClientError):
+    def __init__(self, snapshot_copy_grant_name):
+        super(SnapshotCopyGrantAlreadyExistsFaultError, self).__init__(
+            'SnapshotCopyGrantAlreadyExistsFault',
+            "Cannot create the snapshot copy grant because a grant "
+            "with the identifier '{0}' already exists".format(snapshot_copy_grant_name))
+
+
+class SnapshotCopyGrantNotFoundFaultError(RedshiftClientError):
+    def __init__(self, snapshot_copy_grant_name):
+        super(SnapshotCopyGrantNotFoundFaultError, self).__init__(
+            'SnapshotCopyGrantNotFoundFault',
+            "Snapshot copy grant not found: {0}".format(snapshot_copy_grant_name))
+
+
 class ClusterSnapshotNotFoundError(RedshiftClientError):
     def __init__(self, snapshot_identifier):
         super(ClusterSnapshotNotFoundError, self).__init__(

moto/redshift/models.py

@@ -17,7 +17,9 @@ from .exceptions import (
     ClusterSubnetGroupNotFoundError,
     InvalidParameterValueError,
     InvalidSubnetError,
-    ResourceNotFoundFaultError
+    ResourceNotFoundFaultError,
+    SnapshotCopyGrantAlreadyExistsFaultError,
+    SnapshotCopyGrantNotFoundFaultError
 )
 
 
@@ -231,6 +233,22 @@ class Cluster(TaggableResourceMixin, BaseModel):
             "Tags": self.tags
         }
 
+class SnapshotCopyGrant(TaggableResourceMixin, BaseModel):
+
+    resource_type = 'snapshotcopygrant'
+
+    def __init__(self, snapshot_copy_grant_name, kms_key_id, region_name):
+        self.snapshot_copy_grant_name = snapshot_copy_grant_name
+        self.kms_key_id = kms_key_id
+        self.region_name = region_name
+
+    def to_json(self):
+        return {
+            "SnapshotCopyGrantName": self.snapshot_copy_grant_name,
+            "KmsKeyId": self.kms_key_id
+        }
+
+
 
 class SubnetGroup(TaggableResourceMixin, BaseModel):
 
@@ -410,6 +428,7 @@ class RedshiftBackend(BaseBackend):
             'snapshot': self.snapshots,
             'subnetgroup': self.subnet_groups
         }
+        self.snapshot_copy_grants = {}
 
     def reset(self):
         ec2_backend = self.ec2_backend
@@ -568,6 +587,35 @@ class RedshiftBackend(BaseBackend):
         create_kwargs.update(kwargs)
         return self.create_cluster(**create_kwargs)
 
+    def create_snapshot_copy_grant(self, **kwargs):
+        snapshot_copy_grant_name = kwargs['snapshot_copy_grant_name']
+        kms_key_id = kwargs['kms_key_id']
+        region_name = kwargs['region_name']
+        if snapshot_copy_grant_name not in self.snapshot_copy_grants:
+            snapshot_copy_grant = SnapshotCopyGrant(snapshot_copy_grant_name,
+                                                    kms_key_id, region_name)
+            self.snapshot_copy_grants[snapshot_copy_grant_name] = snapshot_copy_grant
+            return snapshot_copy_grant
+
+        raise SnapshotCopyGrantAlreadyExistsFaultError(snapshot_copy_grant_name)
+
+    def delete_snapshot_copy_grant(self, **kwargs):
+        snapshot_copy_grant_name = kwargs['snapshot_copy_grant_name']
+        if snapshot_copy_grant_name in self.snapshot_copy_grants:
+            return self.snapshot_copy_grants.pop(snapshot_copy_grant_name)
+
+        raise SnapshotCopyGrantNotFoundFaultError(snapshot_copy_grant_name)
+
+    def describe_snapshot_copy_grants(self, **kwargs):
+        copy_grants = self.snapshot_copy_grants.values()
+        snapshot_copy_grant_name = kwargs['snapshot_copy_grant_name']
+        if snapshot_copy_grant_name:
+            if snapshot_copy_grant_name in self.snapshot_copy_grants:
+                return [self.snapshot_copy_grants[snapshot_copy_grant_name]]
+            else:
+                raise SnapshotCopyGrantNotFoundFaultError(snapshot_copy_grant_name)
+        return copy_grants
+
     def _get_resource_from_arn(self, arn):
         try:
             arn_breakdown = arn.split(':')

moto/redshift/responses.py

@@ -457,6 +457,55 @@ class RedshiftResponse(BaseResponse):
             }
         })
 
+    def create_snapshot_copy_grant(self):
+        copy_grant_kwargs = {
+            'snapshot_copy_grant_name': self._get_param('SnapshotCopyGrantName'),
+            'kms_key_id': self._get_param('KmsKeyId'),
+            'region_name': self._get_param('Region'),
+        }
+
+        copy_grant = self.redshift_backend.create_snapshot_copy_grant(**copy_grant_kwargs)
+        return self.get_response({
+            "CreateSnapshotCopyGrantResponse": {
+                "CreateSnapshotCopyGrantResult": {
+                    "SnapshotCopyGrant": copy_grant.to_json()
+                },
+                "ResponseMetadata": {
+                    "RequestId": "384ac68d-3775-11df-8963-01868b7c937a",
+                }
+            }
+        })
+
+    def delete_snapshot_copy_grant(self):
+        copy_grant_kwargs = {
+            'snapshot_copy_grant_name': self._get_param('SnapshotCopyGrantName'),
+        }
+        self.redshift_backend.delete_snapshot_copy_grant(**copy_grant_kwargs)
+        return self.get_response({
+            "DeleteSnapshotCopyGrantResponse": {
+                "ResponseMetadata": {
+                    "RequestId": "384ac68d-3775-11df-8963-01868b7c937a",
+                }
+            }
+        })
+
+    def describe_snapshot_copy_grants(self):
+        copy_grant_kwargs = {
+            'snapshot_copy_grant_name': self._get_param('SnapshotCopyGrantName'),
+        }
+
+        copy_grants = self.redshift_backend.describe_snapshot_copy_grants(**copy_grant_kwargs)
+        return self.get_response({
+            "DescribeSnapshotCopyGrantsResponse": {
+                "DescribeSnapshotCopyGrantsResult": {
+                    "SnapshotCopyGrants": [copy_grant.to_json() for copy_grant in copy_grants]
+                },
+                "ResponseMetadata": {
+                    "RequestId": "384ac68d-3775-11df-8963-01868b7c937a",
+                }
+            }
+        })
+
     def create_tags(self):
         resource_name = self._get_param('ResourceName')
         tags = self.unpack_complex_list_params('Tags.Tag', ('Key', 'Value'))

tests/test_redshift/test_redshift.py

@@ -34,6 +34,45 @@ def test_create_cluster_boto3():
     response['Cluster']['NodeType'].should.equal('ds2.xlarge')
 
 
+@mock_redshift
+def test_create_snapshot_copy_grant():
+    client = boto3.client('redshift', region_name='us-east-1')
+    grants = client.create_snapshot_copy_grant(
+        SnapshotCopyGrantName='test-us-east-1',
+        KmsKeyId='fake',
+    )
+    grants['SnapshotCopyGrant']['SnapshotCopyGrantName'].should.equal('test-us-east-1')
+    grants['SnapshotCopyGrant']['KmsKeyId'].should.equal('fake')
+
+    client.delete_snapshot_copy_grant(
+        SnapshotCopyGrantName='test-us-east-1',
+    )
+
+    client.describe_snapshot_copy_grants.when.called_with(
+        SnapshotCopyGrantName='test-us-east-1',
+    ).should.throw(Exception)
+
+
+@mock_redshift
+def test_create_many_snapshot_copy_grants():
+    client = boto3.client('redshift', region_name='us-east-1')
+
+    for i in range(10):
+        client.create_snapshot_copy_grant(
+            SnapshotCopyGrantName='test-us-east-1-{0}'.format(i),
+            KmsKeyId='fake',
+        )
+    response = client.describe_snapshot_copy_grants()
+    len(response['SnapshotCopyGrants']).should.equal(10)
+
+
+@mock_redshift
+def test_no_snapshot_copy_grants():
+    client = boto3.client('redshift', region_name='us-east-1')
+    response = client.describe_snapshot_copy_grants()
+    len(response['SnapshotCopyGrants']).should.equal(0)
+
+
 @mock_redshift_deprecated
 def test_create_cluster():
     conn = boto.redshift.connect_to_region("us-east-1")