<s> from __future__ import unicode_literals <EOL> import boto . redshift <EOL> from moto . core import BaseBackend <EOL> from moto . ec2 import ec2_backends <EOL> from . exceptions import ( <EOL> ClusterNotFoundError , <EOL> ClusterParameterGroupNotFoundError , <EOL> ClusterSecurityGroupNotFoundError , <EOL> ClusterSubnetGroupNotFoundError , <EOL> InvalidSubnetError , <EOL> ) <EOL> class Cluster ( object ) : <EOL> def __init__ ( self , redshift_backend , cluster_identifier , node_type , master_username , <EOL> master_user_password , db_name , cluster_type , cluster_security_groups , <EOL> vpc_security_group_ids , cluster_subnet_group_name , availability_zone , <EOL> preferred_maintenance_window , cluster_parameter_group_name , <EOL> automated_snapshot_retention_period , port , cluster_version , <EOL> allow_version_upgrade , number_of_nodes , publicly_accessible , <EOL> encrypted , region ) : <EOL> self . redshift_backend = redshift_backend <EOL> self . cluster_identifier = cluster_identifier <EOL> self . node_type = node_type <EOL> self . master_username = master_username <EOL> self . master_user_password = master_user_password <EOL> self . db_name = db_name if db_name else "<STR_LIT>" <EOL> self . vpc_security_group_ids = vpc_security_group_ids <EOL> self . cluster_subnet_group_name = cluster_subnet_group_name <EOL> self . publicly_accessible = publicly_accessible <EOL> self . encrypted = encrypted <EOL> self . allow_version_upgrade = allow_version_upgrade if allow_version_upgrade is not None else True <EOL> self . cluster_version = cluster_version if cluster_version else "<STR_LIT:1.0>" <EOL> self . port = int ( port ) if port else <NUM_LIT> <EOL> self . automated_snapshot_retention_period = int ( automated_snapshot_retention_period ) if automated_snapshot_retention_period else <NUM_LIT:1> <EOL> self . preferred_maintenance_window = preferred_maintenance_window if preferred_maintenance_window else "<STR_LIT>" <EOL> if cluster_parameter_group_name : <EOL> self . cluster_parameter_group_name = [ cluster_parameter_group_name ] <EOL> else : <EOL> self . cluster_parameter_group_name = [ '<STR_LIT>' ] <EOL> if cluster_security_groups : <EOL> self . cluster_security_groups = cluster_security_groups <EOL> else : <EOL> self . cluster_security_groups = [ "<STR_LIT>" ] <EOL> self . region = region <EOL> if availability_zone : <EOL> self . availability_zone = availability_zone <EOL> else : <EOL> self . availability_zone = region + "<STR_LIT:a>" <EOL> if cluster_type == '<STR_LIT>' : <EOL> self . number_of_nodes = <NUM_LIT:1> <EOL> elif number_of_nodes : <EOL> self . number_of_nodes = int ( number_of_nodes ) <EOL> else : <EOL> self . number_of_nodes = <NUM_LIT:1> <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> redshift_backend = redshift_backends [ region_name ] <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in properties : <EOL> subnet_group_name = properties [ '<STR_LIT>' ] . cluster_subnet_group_name <EOL> else : <EOL> subnet_group_name = None <EOL> cluster = redshift_backend . create_cluster ( <EOL> cluster_identifier = resource_name , <EOL> node_type = properties . get ( '<STR_LIT>' ) , <EOL> master_username = properties . get ( '<STR_LIT>' ) , <EOL> master_user_password = properties . get ( '<STR_LIT>' ) , <EOL> db_name = properties . get ( '<STR_LIT>' ) , <EOL> cluster_type = properties . get ( '<STR_LIT>' ) , <EOL> cluster_security_groups = properties . get ( '<STR_LIT>' , [ ] ) , <EOL> vpc_security_group_ids = properties . get ( '<STR_LIT>' , [ ] ) , <EOL> cluster_subnet_group_name = subnet_group_name , <EOL> availability_zone = properties . get ( '<STR_LIT>' ) , <EOL> preferred_maintenance_window = properties . get ( '<STR_LIT>' ) , <EOL> cluster_parameter_group_name = properties . get ( '<STR_LIT>' ) , <EOL> automated_snapshot_retention_period = properties . get ( '<STR_LIT>' ) , <EOL> port = properties . get ( '<STR_LIT>' ) , <EOL> cluster_version = properties . get ( '<STR_LIT>' ) , <EOL> allow_version_upgrade = properties . get ( '<STR_LIT>' ) , <EOL> number_of_nodes = properties . get ( '<STR_LIT>' ) , <EOL> publicly_accessible = properties . get ( "<STR_LIT>" ) , <EOL> encrypted = properties . get ( "<STR_LIT>" ) , <EOL> region = region_name , <EOL> ) <EOL> return cluster <EOL> def get_cfn_attribute ( self , attribute_name ) : <EOL> from moto . cloudformation . exceptions import UnformattedGetAttTemplateException <EOL> if attribute_name == '<STR_LIT>' : <EOL> return self . endpoint <EOL> elif attribute_name == '<STR_LIT>' : <EOL> return self . port <EOL> raise UnformattedGetAttTemplateException ( ) <EOL> @ property <EOL> def endpoint ( self ) : <EOL> return "<STR_LIT>" . format ( <EOL> self . cluster_identifier , <EOL> self . region , <EOL> ) <EOL> @ property <EOL> def security_groups ( self ) : <EOL> return [ <EOL> security_group for security_group <EOL> in self . redshift_backend . describe_cluster_security_groups ( ) <EOL> if security_group . cluster_security_group_name in self . cluster_security_groups <EOL> ] <EOL> @ property <EOL> def vpc_security_groups ( self ) : <EOL> return [ <EOL> security_group for security_group <EOL> in self . redshift_backend . ec2_backend . describe_security_groups ( ) <EOL> if security_group . id in self . vpc_security_group_ids <EOL> ] <EOL> @ property <EOL> def parameter_groups ( self ) : <EOL> return [ <EOL> parameter_group for parameter_group <EOL> in self . redshift_backend . describe_cluster_parameter_groups ( ) <EOL> if parameter_group . cluster_parameter_group_name in self . cluster_parameter_group_name <EOL> ] <EOL> def to_json ( self ) : <EOL> return { <EOL> "<STR_LIT>" : self . master_username , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . cluster_version , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : group . id <EOL> } for group in self . vpc_security_groups ] , <EOL> "<STR_LIT>" : self . cluster_subnet_group_name , <EOL> "<STR_LIT>" : self . availability_zone , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . number_of_nodes , <EOL> "<STR_LIT>" : self . automated_snapshot_retention_period , <EOL> "<STR_LIT>" : self . publicly_accessible , <EOL> "<STR_LIT>" : self . encrypted , <EOL> "<STR_LIT>" : self . db_name , <EOL> "<STR_LIT>" : self . preferred_maintenance_window , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : group . cluster_parameter_group_name , <EOL> } for group in self . parameter_groups ] , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : group . cluster_security_group_name , <EOL> } for group in self . security_groups ] , <EOL> "<STR_LIT>" : self . port , <EOL> "<STR_LIT>" : self . node_type , <EOL> "<STR_LIT>" : self . cluster_identifier , <EOL> "<STR_LIT>" : self . allow_version_upgrade , <EOL> } <EOL> class SubnetGroup ( object ) : <EOL> def __init__ ( self , ec2_backend , cluster_subnet_group_name , description , subnet_ids ) : <EOL> self . ec2_backend = ec2_backend <EOL> self . cluster_subnet_group_name = cluster_subnet_group_name <EOL> self . description = description <EOL> self . subnet_ids = subnet_ids <EOL> if not self . subnets : <EOL> raise InvalidSubnetError ( subnet_ids ) <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> redshift_backend = redshift_backends [ region_name ] <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> subnet_group = redshift_backend . create_cluster_subnet_group ( <EOL> cluster_subnet_group_name = resource_name , <EOL> description = properties . get ( "<STR_LIT>" ) , <EOL> subnet_ids = properties . get ( "<STR_LIT>" , [ ] ) , <EOL> ) <EOL> return subnet_group <EOL> @ property <EOL> def subnets ( self ) : <EOL> return self . ec2_backend . get_all_subnets ( filters = { '<STR_LIT>' : self . subnet_ids } ) <EOL> @ property <EOL> def vpc_id ( self ) : <EOL> return self . subnets [ <NUM_LIT:0> ] . vpc_id <EOL> def to_json ( self ) : <EOL> return { <EOL> "<STR_LIT>" : self . vpc_id , <EOL> "<STR_LIT>" : self . description , <EOL> "<STR_LIT>" : self . cluster_subnet_group_name , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : subnet . id , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:Name>" : subnet . availability_zone <EOL> } , <EOL> } for subnet in self . subnets ] , <EOL> } <EOL> class SecurityGroup ( object ) : <EOL> def __init__ ( self , cluster_security_group_name , description ) : <EOL> self . cluster_security_group_name = cluster_security_group_name <EOL> self . description = description <EOL> def to_json ( self ) : <EOL> return { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : self . description , <EOL> "<STR_LIT>" : self . cluster_security_group_name , <EOL> } <EOL> class ParameterGroup ( object ) : <EOL> def __init__ ( self , cluster_parameter_group_name , group_family , description ) : <EOL> self . cluster_parameter_group_name = cluster_parameter_group_name <EOL> self . group_family = group_family <EOL> self . description = description <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> redshift_backend = redshift_backends [ region_name ] <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> parameter_group = redshift_backend . create_cluster_parameter_group ( <EOL> cluster_parameter_group_name = resource_name , <EOL> description = properties . get ( "<STR_LIT>" ) , <EOL> group_family = properties . get ( "<STR_LIT>" ) , <EOL> ) <EOL> return parameter_group <EOL> def to_json ( self ) : <EOL> return { <EOL> "<STR_LIT>" : self . group_family , <EOL> "<STR_LIT>" : self . description , <EOL> "<STR_LIT>" : self . cluster_parameter_group_name , <EOL> } <EOL> class RedshiftBackend ( BaseBackend ) : <EOL> def __init__ ( self , ec2_backend ) : <EOL> self . clusters = { } <EOL> self . subnet_groups = { } <EOL> self . security_groups = { <EOL> "<STR_LIT>" : SecurityGroup ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> } <EOL> self . parameter_groups = { <EOL> "<STR_LIT>" : ParameterGroup ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ) <EOL> } <EOL> self . ec2_backend = ec2_backend <EOL> def reset ( self ) : <EOL> ec2_backend = self . ec2_backend <EOL> self . __dict__ = { } <EOL> self . __init__ ( ec2_backend ) <EOL> def create_cluster ( self , ** cluster_kwargs ) : <EOL> cluster_identifier = cluster_kwargs [ '<STR_LIT>' ] <EOL> cluster = Cluster ( self , ** cluster_kwargs ) <EOL> self . clusters [ cluster_identifier ] = cluster <EOL> return cluster <EOL> def describe_clusters ( self , cluster_identifier = None ) : <EOL> clusters = self . clusters . values ( ) <EOL> if cluster_identifier : <EOL> if cluster_identifier in self . clusters : <EOL> return [ self . clusters [ cluster_identifier ] ] <EOL> else : <EOL> raise ClusterNotFoundError ( cluster_identifier ) <EOL> return clusters <EOL> def modify_cluster ( self , ** cluster_kwargs ) : <EOL> cluster_identifier = cluster_kwargs . pop ( '<STR_LIT>' ) <EOL> new_cluster_identifier = cluster_kwargs . pop ( '<STR_LIT>' , None ) <EOL> cluster = self . describe_clusters ( cluster_identifier ) [ <NUM_LIT:0> ] <EOL> for key , value in cluster_kwargs . items ( ) : <EOL> setattr ( cluster , key , value ) <EOL> if new_cluster_identifier : <EOL> self . delete_cluster ( cluster_identifier ) <EOL> cluster . cluster_identifier = new_cluster_identifier <EOL> self . clusters [ new_cluster_identifier ] = cluster <EOL> return cluster <EOL> def delete_cluster ( self , cluster_identifier ) : <EOL> if cluster_identifier in self . clusters : <EOL> return self . clusters . pop ( cluster_identifier ) <EOL> raise ClusterNotFoundError ( cluster_identifier ) <EOL> def create_cluster_subnet_group ( self , cluster_subnet_group_name , description , subnet_ids ) : <EOL> subnet_group = SubnetGroup ( self . ec2_backend , cluster_subnet_group_name , description , subnet_ids ) <EOL> self . subnet_groups [ cluster_subnet_group_name ] = subnet_group <EOL> return subnet_group <EOL> def describe_cluster_subnet_groups ( self , subnet_identifier = None ) : <EOL> subnet_groups = self . subnet_groups . values ( ) <EOL> if subnet_identifier : <EOL> if subnet_identifier in self . subnet_groups : <EOL> return [ self . subnet_groups [ subnet_identifier ] ] <EOL> else : <EOL> raise ClusterSubnetGroupNotFoundError ( subnet_identifier ) <EOL> return subnet_groups <EOL> def delete_cluster_subnet_group ( self , subnet_identifier ) : <EOL> if subnet_identifier in self . subnet_groups : <EOL> return self . subnet_groups . pop ( subnet_identifier ) <EOL> raise ClusterSubnetGroupNotFoundError ( subnet_identifier ) <EOL> def create_cluster_security_group ( self , cluster_security_group_name , description ) : <EOL> security_group = SecurityGroup ( cluster_security_group_name , description ) <EOL> self . security_groups [ cluster_security_group_name ] = security_group <EOL> return security_group <EOL> def describe_cluster_security_groups ( self , security_group_name = None ) : <EOL> security_groups = self . security_groups . values ( ) <EOL> if security_group_name : <EOL> if security_group_name in self . security_groups : <EOL> return [ self . security_groups [ security_group_name ] ] <EOL> else : <EOL> raise ClusterSecurityGroupNotFoundError ( security_group_name ) <EOL> return security_groups <EOL> def delete_cluster_security_group ( self , security_group_identifier ) : <EOL> if security_group_identifier in self . security_groups : <EOL> return self . security_groups . pop ( security_group_identifier ) <EOL> raise ClusterSecurityGroupNotFoundError ( security_group_identifier ) <EOL> def create_cluster_parameter_group ( self , cluster_parameter_group_name , <EOL> group_family , description ) : <EOL> parameter_group = ParameterGroup ( cluster_parameter_group_name , group_family , description ) <EOL> self . parameter_groups [ cluster_parameter_group_name ] = parameter_group <EOL> return parameter_group <EOL> def describe_cluster_parameter_groups ( self , parameter_group_name = None ) : <EOL> parameter_groups = self . parameter_groups . values ( ) <EOL> if parameter_group_name : <EOL> if parameter_group_name in self . parameter_groups : <EOL> return [ self . parameter_groups [ parameter_group_name ] ] <EOL> else : <EOL> raise ClusterParameterGroupNotFoundError ( parameter_group_name ) <EOL> return parameter_groups <EOL> def delete_cluster_parameter_group ( self , parameter_group_name ) : <EOL> if parameter_group_name in self . parameter_groups : <EOL> return self . parameter_groups . pop ( parameter_group_name ) <EOL> raise ClusterParameterGroupNotFoundError ( parameter_group_name ) <EOL> redshift_backends = { } <EOL> for region in boto . redshift . regions ( ) : <EOL> redshift_backends [ region . name ] = RedshiftBackend ( ec2_backends [ region . name ] ) </s>
