Merge pull request #1451 from captainkerk/redshift-add-cross-region-snapshots

Redshift: Add Cross Region Snapshot Functionality

moto/redshift/exceptions.py

@@ -93,3 +93,24 @@ class ResourceNotFoundFaultError(RedshiftClientError):
             msg = message
         super(ResourceNotFoundFaultError, self).__init__(
             'ResourceNotFoundFault', msg)
+
+
+class SnapshotCopyDisabledFaultError(RedshiftClientError):
+    def __init__(self, cluster_identifier):
+        super(SnapshotCopyDisabledFaultError, self).__init__(
+            'SnapshotCopyDisabledFault',
+            "Cannot modify retention period because snapshot copy is disabled on Cluster {0}.".format(cluster_identifier))
+
+
+class SnapshotCopyAlreadyDisabledFaultError(RedshiftClientError):
+    def __init__(self, cluster_identifier):
+        super(SnapshotCopyAlreadyDisabledFaultError, self).__init__(
+            'SnapshotCopyAlreadyDisabledFault',
+            "Snapshot Copy is already disabled on Cluster {0}.".format(cluster_identifier))
+
+
+class SnapshotCopyAlreadyEnabledFaultError(RedshiftClientError):
+    def __init__(self, cluster_identifier):
+        super(SnapshotCopyAlreadyEnabledFaultError, self).__init__(
+            'SnapshotCopyAlreadyEnabledFault',
+            "Snapshot Copy is already enabled on Cluster {0}.".format(cluster_identifier))

moto/redshift/models.py

@@ -4,6 +4,7 @@ import copy
 import datetime
 
 import boto.redshift
+from botocore.exceptions import ClientError
 from moto.compat import OrderedDict
 from moto.core import BaseBackend, BaseModel
 from moto.core.utils import iso_8601_datetime_with_milliseconds
@@ -17,7 +18,10 @@ from .exceptions import (
     ClusterSubnetGroupNotFoundError,
     InvalidParameterValueError,
     InvalidSubnetError,
-    ResourceNotFoundFaultError
+    ResourceNotFoundFaultError,
+    SnapshotCopyDisabledFaultError,
+    SnapshotCopyAlreadyDisabledFaultError,
+    SnapshotCopyAlreadyEnabledFaultError,
 )
 
 
@@ -194,7 +198,7 @@ class Cluster(TaggableResourceMixin, BaseModel):
         return self.cluster_identifier
 
     def to_json(self):
-        return {
+        json_response = {
             "MasterUsername": self.master_username,
             "MasterUserPassword": "****",
             "ClusterVersion": self.cluster_version,
@@ -231,6 +235,12 @@ class Cluster(TaggableResourceMixin, BaseModel):
             "Tags": self.tags
         }
 
+        try:
+            json_response['ClusterSnapshotCopyStatus'] = self.cluster_snapshot_copy_status
+        except AttributeError:
+            pass
+        return json_response
+
 
 class SubnetGroup(TaggableResourceMixin, BaseModel):
 
@@ -417,6 +427,43 @@ class RedshiftBackend(BaseBackend):
         self.__dict__ = {}
         self.__init__(ec2_backend, region_name)
 
+    def enable_snapshot_copy(self, **kwargs):
+        cluster_identifier = kwargs['cluster_identifier']
+        cluster = self.clusters[cluster_identifier]
+        if not hasattr(cluster, 'cluster_snapshot_copy_status'):
+            if cluster.encrypted == 'true' and kwargs['snapshot_copy_grant_name'] is None:
+                raise ClientError(
+                    'InvalidParameterValue',
+                    'SnapshotCopyGrantName is required for Snapshot Copy '
+                    'on KMS encrypted clusters.'
+                )
+            status = {
+                'DestinationRegion': kwargs['destination_region'],
+                'RetentionPeriod': kwargs['retention_period'],
+                'SnapshotCopyGrantName': kwargs['snapshot_copy_grant_name'],
+            }
+            cluster.cluster_snapshot_copy_status = status
+            return cluster
+        else:
+            raise SnapshotCopyAlreadyEnabledFaultError(cluster_identifier)
+
+    def disable_snapshot_copy(self, **kwargs):
+        cluster_identifier = kwargs['cluster_identifier']
+        cluster = self.clusters[cluster_identifier]
+        if hasattr(cluster, 'cluster_snapshot_copy_status'):
+            del cluster.cluster_snapshot_copy_status
+            return cluster
+        else:
+            raise SnapshotCopyAlreadyDisabledFaultError(cluster_identifier)
+
+    def modify_snapshot_copy_retention_period(self, cluster_identifier, retention_period):
+        cluster = self.clusters[cluster_identifier]
+        if hasattr(cluster, 'cluster_snapshot_copy_status'):
+            cluster.cluster_snapshot_copy_status['RetentionPeriod'] = retention_period
+            return cluster
+        else:
+            raise SnapshotCopyDisabledFaultError(cluster_identifier)
+
     def create_cluster(self, **cluster_kwargs):
         cluster_identifier = cluster_kwargs['cluster_identifier']
         cluster = Cluster(self, **cluster_kwargs)

moto/redshift/responses.py

@@ -501,3 +501,58 @@ class RedshiftResponse(BaseResponse):
                 }
             }
         })
+
+    def enable_snapshot_copy(self):
+        snapshot_copy_kwargs = {
+            'cluster_identifier': self._get_param('ClusterIdentifier'),
+            'destination_region': self._get_param('DestinationRegion'),
+            'retention_period': self._get_param('RetentionPeriod', 7),
+            'snapshot_copy_grant_name': self._get_param('SnapshotCopyGrantName'),
+        }
+        cluster = self.redshift_backend.enable_snapshot_copy(**snapshot_copy_kwargs)
+
+        return self.get_response({
+            "EnableSnapshotCopyResponse": {
+                "EnableSnapshotCopyResult": {
+                    "Cluster": cluster.to_json()
+                },
+                "ResponseMetadata": {
+                    "RequestId": "384ac68d-3775-11df-8963-01868b7c937a",
+                }
+            }
+        })
+
+    def disable_snapshot_copy(self):
+        snapshot_copy_kwargs = {
+            'cluster_identifier': self._get_param('ClusterIdentifier'),
+        }
+        cluster = self.redshift_backend.disable_snapshot_copy(**snapshot_copy_kwargs)
+
+        return self.get_response({
+            "DisableSnapshotCopyResponse": {
+                "DisableSnapshotCopyResult": {
+                    "Cluster": cluster.to_json()
+                },
+                "ResponseMetadata": {
+                    "RequestId": "384ac68d-3775-11df-8963-01868b7c937a",
+                }
+            }
+        })
+
+    def modify_snapshot_copy_retention_period(self):
+        snapshot_copy_kwargs = {
+            'cluster_identifier': self._get_param('ClusterIdentifier'),
+            'retention_period': self._get_param('RetentionPeriod'),
+        }
+        cluster = self.redshift_backend.modify_snapshot_copy_retention_period(**snapshot_copy_kwargs)
+
+        return self.get_response({
+            "ModifySnapshotCopyRetentionPeriodResponse": {
+                "ModifySnapshotCopyRetentionPeriodResult": {
+                    "Clusters": [cluster.to_json()]
+                },
+                "ResponseMetadata": {
+                    "RequestId": "384ac68d-3775-11df-8963-01868b7c937a",
+                }
+            }
+        })

tests/test_redshift/test_redshift.py

@@ -1042,3 +1042,98 @@ def test_tagged_resource_not_found_error():
         ResourceName='bad:arn'
     ).should.throw(ClientError, "Tagging is not supported for this type of resource")
 
+
+@mock_redshift
+def test_enable_snapshot_copy():
+    client = boto3.client('redshift', region_name='us-east-1')
+    client.create_cluster(
+        ClusterIdentifier='test',
+        ClusterType='single-node',
+        DBName='test',
+        Encrypted=True,
+        MasterUsername='user',
+        MasterUserPassword='password',
+        NodeType='ds2.xlarge',
+    )
+    client.enable_snapshot_copy(
+        ClusterIdentifier='test',
+        DestinationRegion='us-west-2',
+        RetentionPeriod=3,
+        SnapshotCopyGrantName='copy-us-east-1-to-us-west-2'
+    )
+    response = client.describe_clusters(ClusterIdentifier='test')
+    cluster_snapshot_copy_status = response['Clusters'][0]['ClusterSnapshotCopyStatus']
+    cluster_snapshot_copy_status['RetentionPeriod'].should.equal(3)
+    cluster_snapshot_copy_status['DestinationRegion'].should.equal('us-west-2')
+    cluster_snapshot_copy_status['SnapshotCopyGrantName'].should.equal('copy-us-east-1-to-us-west-2')
+
+
+@mock_redshift
+def test_enable_snapshot_copy_unencrypted():
+    client = boto3.client('redshift', region_name='us-east-1')
+    client.create_cluster(
+        ClusterIdentifier='test',
+        ClusterType='single-node',
+        DBName='test',
+        MasterUsername='user',
+        MasterUserPassword='password',
+        NodeType='ds2.xlarge',
+    )
+    client.enable_snapshot_copy(
+        ClusterIdentifier='test',
+        DestinationRegion='us-west-2',
+    )
+    response = client.describe_clusters(ClusterIdentifier='test')
+    cluster_snapshot_copy_status = response['Clusters'][0]['ClusterSnapshotCopyStatus']
+    cluster_snapshot_copy_status['RetentionPeriod'].should.equal(7)
+    cluster_snapshot_copy_status['DestinationRegion'].should.equal('us-west-2')
+
+
+@mock_redshift
+def test_disable_snapshot_copy():
+    client = boto3.client('redshift', region_name='us-east-1')
+    client.create_cluster(
+        DBName='test',
+        ClusterIdentifier='test',
+        ClusterType='single-node',
+        NodeType='ds2.xlarge',
+        MasterUsername='user',
+        MasterUserPassword='password',
+    )
+    client.enable_snapshot_copy(
+        ClusterIdentifier='test',
+        DestinationRegion='us-west-2',
+        RetentionPeriod=3,
+        SnapshotCopyGrantName='copy-us-east-1-to-us-west-2',
+    )
+    client.disable_snapshot_copy(
+        ClusterIdentifier='test',
+    )
+    response = client.describe_clusters(ClusterIdentifier='test')
+    response['Clusters'][0].shouldnt.contain('ClusterSnapshotCopyStatus')
+
+
+@mock_redshift
+def test_modify_snapshot_copy_retention_period():
+    client = boto3.client('redshift', region_name='us-east-1')
+    client.create_cluster(
+        DBName='test',
+        ClusterIdentifier='test',
+        ClusterType='single-node',
+        NodeType='ds2.xlarge',
+        MasterUsername='user',
+        MasterUserPassword='password',
+    )
+    client.enable_snapshot_copy(
+        ClusterIdentifier='test',
+        DestinationRegion='us-west-2',
+        RetentionPeriod=3,
+        SnapshotCopyGrantName='copy-us-east-1-to-us-west-2',
+    )
+    client.modify_snapshot_copy_retention_period(
+        ClusterIdentifier='test',
+        RetentionPeriod=5,
+    )
+    response = client.describe_clusters(ClusterIdentifier='test')
+    cluster_snapshot_copy_status = response['Clusters'][0]['ClusterSnapshotCopyStatus']
+    cluster_snapshot_copy_status['RetentionPeriod'].should.equal(5)