rework to follow spec for encrypted/unencrypted clusters

This commit is contained in:
captainkerk 2018-01-28 03:53:32 +00:00
parent ed06658271
commit 7130dd5239
3 changed files with 32 additions and 3 deletions

View File

@ -4,6 +4,7 @@ import copy
import datetime import datetime
import boto.redshift import boto.redshift
from botocore.exceptions import ClientError
from moto.compat import OrderedDict from moto.compat import OrderedDict
from moto.core import BaseBackend, BaseModel from moto.core import BaseBackend, BaseModel
from moto.core.utils import iso_8601_datetime_with_milliseconds from moto.core.utils import iso_8601_datetime_with_milliseconds
@ -430,6 +431,12 @@ class RedshiftBackend(BaseBackend):
cluster_identifier = kwargs['cluster_identifier'] cluster_identifier = kwargs['cluster_identifier']
cluster = self.clusters[cluster_identifier] cluster = self.clusters[cluster_identifier]
if not hasattr(cluster, 'cluster_snapshot_copy_status'): 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 = { status = {
'DestinationRegion': kwargs['destination_region'], 'DestinationRegion': kwargs['destination_region'],
'RetentionPeriod': kwargs['retention_period'], 'RetentionPeriod': kwargs['retention_period'],

View File

@ -506,7 +506,7 @@ class RedshiftResponse(BaseResponse):
snapshot_copy_kwargs = { snapshot_copy_kwargs = {
'cluster_identifier': self._get_param('ClusterIdentifier'), 'cluster_identifier': self._get_param('ClusterIdentifier'),
'destination_region': self._get_param('DestinationRegion'), 'destination_region': self._get_param('DestinationRegion'),
'retention_period': self._get_param('RetentionPeriod'), 'retention_period': self._get_param('RetentionPeriod', 7),
'snapshot_copy_grant_name': self._get_param('SnapshotCopyGrantName'), 'snapshot_copy_grant_name': self._get_param('SnapshotCopyGrantName'),
} }
cluster = self.redshift_backend.enable_snapshot_copy(**snapshot_copy_kwargs) cluster = self.redshift_backend.enable_snapshot_copy(**snapshot_copy_kwargs)

View File

@ -1047,12 +1047,13 @@ def test_tagged_resource_not_found_error():
def test_enable_snapshot_copy(): def test_enable_snapshot_copy():
client = boto3.client('redshift', region_name='us-east-1') client = boto3.client('redshift', region_name='us-east-1')
client.create_cluster( client.create_cluster(
DBName='test',
ClusterIdentifier='test', ClusterIdentifier='test',
ClusterType='single-node', ClusterType='single-node',
NodeType='ds2.xlarge', DBName='test',
Encrypted=True,
MasterUsername='user', MasterUsername='user',
MasterUserPassword='password', MasterUserPassword='password',
NodeType='ds2.xlarge',
) )
client.enable_snapshot_copy( client.enable_snapshot_copy(
ClusterIdentifier='test', ClusterIdentifier='test',
@ -1067,6 +1068,27 @@ def test_enable_snapshot_copy():
cluster_snapshot_copy_status['SnapshotCopyGrantName'].should.equal('copy-us-east-1-to-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 @mock_redshift
def test_disable_snapshot_copy(): def test_disable_snapshot_copy():
client = boto3.client('redshift', region_name='us-east-1') client = boto3.client('redshift', region_name='us-east-1')