<s> """<STR_LIT>""" <EOL> import time <EOL> from cloudcafe . common . tools . datagen import rand_name <EOL> from cloudcafe . networking . networks . common . behaviors import NetworkingBaseBehaviors , NetworkingResponse <EOL> from cloudcafe . networking . networks . common . exceptions import ResourceBuildException , ResourceDeleteException , ResourceGetException , ResourceListException , ResourceUpdateException <EOL> from cloudcafe . networking . networks . extensions . security_groups_api . constants import SecurityGroupsResponseCodes <EOL> class SecurityGroupsBehaviors ( NetworkingBaseBehaviors ) : <EOL> def __init__ ( self , security_groups_client , security_groups_config ) : <EOL> super ( SecurityGroupsBehaviors , self ) . __init__ ( ) <EOL> self . config = security_groups_config <EOL> self . client = security_groups_client <EOL> def create_security_group ( self , name = None , description = None , <EOL> tenant_id = None , resource_build_attempts = None , <EOL> raise_exception = True , use_exact_name = False , <EOL> poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> if name is None and not use_exact_name : <EOL> name = rand_name ( self . config . starts_with_name ) <EOL> elif not use_exact_name : <EOL> name = rand_name ( name ) <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_build_attempts = ( resource_build_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> err_msg = '<STR_LIT>' <EOL> for attempt in range ( resource_build_attempts ) : <EOL> self . _log . debug ( <EOL> '<STR_LIT>' . format ( <EOL> attempt + <NUM_LIT:1> , resource_build_attempts , name ) ) <EOL> resp = self . client . create_security_group ( <EOL> name = name , description = description , tenant_id = tenant_id ) <EOL> resp_check = self . check_response ( <EOL> resp = resp , <EOL> status_code = SecurityGroupsResponseCodes . CREATE_SECURITY_GROUP , <EOL> label = name , <EOL> message = err_msg ) <EOL> result . response = resp <EOL> if not resp_check : <EOL> return result <EOL> result . failures . append ( resp_check ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( name , resource_build_attempts , result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceBuildException ( err_msg ) <EOL> return result <EOL> def update_security_group ( self , security_group_id , name = None , <EOL> description = None , tenant_id = None , <EOL> resource_update_attempts = None , <EOL> raise_exception = False , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_update_attempts = ( resource_update_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> err_msg = '<STR_LIT>' <EOL> for attempt in range ( resource_update_attempts ) : <EOL> self . _log . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( attempt + <NUM_LIT:1> , resource_update_attempts , <EOL> security_group_id ) ) <EOL> resp = self . client . update_security_group ( <EOL> security_group_id = security_group_id , name = name , <EOL> description = description , tenant_id = tenant_id ) <EOL> resp_check = self . check_response ( <EOL> resp = resp , <EOL> status_code = SecurityGroupsResponseCodes . UPDATE_SECURITY_GROUP , <EOL> label = security_group_id , message = err_msg ) <EOL> result . response = resp <EOL> if not resp_check : <EOL> return result <EOL> result . failures . append ( resp_check ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( security_group_id , resource_update_attempts , <EOL> result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceUpdateException ( err_msg ) <EOL> return result <EOL> def get_security_group ( self , security_group_id , resource_get_attempts = None , <EOL> raise_exception = False , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_get_attempts = ( resource_get_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> err_msg = '<STR_LIT>' <EOL> for attempt in range ( resource_get_attempts ) : <EOL> self . _log . debug ( <EOL> '<STR_LIT>' . format ( <EOL> attempt + <NUM_LIT:1> , <EOL> resource_get_attempts , <EOL> security_group_id ) ) <EOL> resp = self . client . get_security_group ( <EOL> security_group_id = security_group_id ) <EOL> resp_check = self . check_response ( <EOL> resp = resp , <EOL> status_code = SecurityGroupsResponseCodes . GET_SECURITY_GROUP , <EOL> label = security_group_id , <EOL> message = err_msg ) <EOL> result . response = resp <EOL> if not resp_check : <EOL> return result <EOL> result . failures . append ( resp_check ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( security_group_id , resource_get_attempts , <EOL> result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceGetException ( err_msg ) <EOL> return result <EOL> def list_security_groups ( self , security_group_id = None , name = None , <EOL> description = None , tenant_id = None , <EOL> limit = None , marker = None , page_reverse = None , <EOL> resource_list_attempts = None , <EOL> raise_exception = False , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_list_attempts = ( resource_list_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> err_msg = '<STR_LIT>' <EOL> for attempt in range ( resource_list_attempts ) : <EOL> self . _log . debug ( <EOL> '<STR_LIT>' . format ( <EOL> attempt + <NUM_LIT:1> , <EOL> resource_list_attempts ) ) <EOL> resp = self . client . list_security_groups ( <EOL> security_group_id = security_group_id , name = name , <EOL> description = description , tenant_id = tenant_id , limit = limit , <EOL> marker = marker , page_reverse = page_reverse ) <EOL> resp_check = self . check_response ( <EOL> resp = resp , <EOL> status_code = SecurityGroupsResponseCodes . LIST_SECURITY_GROUPS , <EOL> label = '<STR_LIT>' , <EOL> message = err_msg ) <EOL> result . response = resp <EOL> if not resp_check : <EOL> return result <EOL> result . failures . append ( resp_check ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( resource_list_attempts , result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceListException ( err_msg ) <EOL> return result <EOL> def delete_security_group ( self , security_group_id , <EOL> resource_delete_attempts = None , <EOL> raise_exception = False , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_delete_attempts = ( resource_delete_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> for attempt in range ( resource_delete_attempts ) : <EOL> self . _log . debug ( <EOL> '<STR_LIT>' . format ( <EOL> attempt + <NUM_LIT:1> , resource_delete_attempts , security_group_id ) ) <EOL> resp = self . client . delete_security_group ( <EOL> security_group_id = security_group_id ) <EOL> result . response = resp <EOL> if ( resp . ok and resp . status_code == <EOL> SecurityGroupsResponseCodes . DELETE_SECURITY_GROUP ) : <EOL> return result <EOL> del_status_code = SecurityGroupsResponseCodes . DELETE_SECURITY_GROUP <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) . format ( security_group = security_group_id , <EOL> expected_status = del_status_code , <EOL> status = resp . status_code , <EOL> reason = resp . reason , <EOL> content = resp . content ) <EOL> self . _log . error ( err_msg ) <EOL> result . failures . append ( err_msg ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( security_group_id , resource_delete_attempts , <EOL> result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceDeleteException ( err_msg ) <EOL> return result <EOL> def delete_security_groups ( self , security_group_list = None , name = None , <EOL> tenant_id = None ) : <EOL> """<STR_LIT>""" <EOL> if security_group_list is None : <EOL> resp = self . list_security_groups ( tenant_id = tenant_id ) <EOL> if ( resp . response . status_code != <EOL> SecurityGroupsResponseCodes . LIST_SECURITY_GROUPS ) : <EOL> get_msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . _log . info ( get_msg ) <EOL> return None <EOL> security_groups = resp . response . entity <EOL> if tenant_id : <EOL> security_groups = self . filter_entity_list_by_attr ( <EOL> entity_list = security_groups , attr = '<STR_LIT>' , <EOL> value = tenant_id ) <EOL> security_group_list = self . get_id_list_from_entity_list ( <EOL> entity_list = security_groups , name = name ) <EOL> log_msg = '<STR_LIT>' . format ( security_group_list ) <EOL> self . _log . info ( log_msg ) <EOL> failed_deletes = [ ] <EOL> for security_group_id in security_group_list : <EOL> result = self . delete_security_group ( <EOL> security_group_id = security_group_id ) <EOL> if result . failures : <EOL> failed_deletes . append ( result . failures ) <EOL> return failed_deletes <EOL> def clean_security_group ( self , security_group_id , timeout = None , <EOL> poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> timeout = timeout or self . config . resource_delete_timeout <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> endtime = time . time ( ) + int ( timeout ) <EOL> log_msg = '<STR_LIT>' . format ( <EOL> security_group_id , timeout ) <EOL> self . _log . info ( log_msg ) <EOL> resp = None <EOL> while time . time ( ) < endtime : <EOL> try : <EOL> self . client . delete_security_group ( <EOL> security_group_id = security_group_id ) <EOL> resp = self . client . get_security_group ( <EOL> security_group_id = security_group_id ) <EOL> except Exception as err : <EOL> err_msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( err ) <EOL> self . _log . error ( err_msg ) <EOL> if ( resp is not None and <EOL> resp . status_code == SecurityGroupsResponseCodes . NOT_FOUND ) : <EOL> return None <EOL> time . sleep ( poll_interval ) <EOL> err_msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( security_group_id , timeout ) <EOL> self . _log . error ( err_msg ) <EOL> return security_group_id <EOL> def clean_security_groups ( self , security_group_list , timeout = None , <EOL> poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> log_msg = '<STR_LIT>' . format ( security_group_list ) <EOL> self . _log . info ( log_msg ) <EOL> undeleted_security_groups = [ ] <EOL> for security_group in security_group_list : <EOL> result = self . clean_security_group ( <EOL> security_group_id = security_group , timeout = timeout , <EOL> poll_interval = poll_interval ) <EOL> if result : <EOL> undeleted_security_groups . append ( result ) <EOL> if undeleted_security_groups : <EOL> err_msg = '<STR_LIT>' . format ( <EOL> undeleted_security_groups ) <EOL> self . _log . error ( err_msg ) <EOL> return undeleted_security_groups <EOL> def create_security_group_rule ( self , security_group_id , <EOL> direction = '<STR_LIT>' , ethertype = '<STR_LIT>' , <EOL> protocol = None , port_range_min = None , <EOL> port_range_max = None , remote_group_id = None , <EOL> remote_ip_prefix = None , <EOL> resource_build_attempts = None , <EOL> raise_exception = True , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_build_attempts = ( resource_build_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> err_msg = '<STR_LIT>' <EOL> for attempt in range ( resource_build_attempts ) : <EOL> self . _log . debug ( ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> . format ( attempt + <NUM_LIT:1> , <EOL> resource_build_attempts , <EOL> security_group_id ) ) <EOL> resp = self . client . create_security_group_rule ( <EOL> security_group_id = security_group_id , direction = direction , <EOL> ethertype = ethertype , protocol = protocol , <EOL> port_range_min = port_range_min , port_range_max = port_range_max , <EOL> remote_group_id = remote_group_id , <EOL> remote_ip_prefix = remote_ip_prefix ) <EOL> label = '<STR_LIT>' . format ( security_group_id ) <EOL> exp_code = SecurityGroupsResponseCodes . CREATE_SECURITY_GROUP_RULE <EOL> resp_check = self . check_response ( <EOL> resp = resp , status_code = exp_code , label = label , message = err_msg ) <EOL> result . response = resp <EOL> if not resp_check : <EOL> return result <EOL> result . failures . append ( resp_check ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) . format ( security_group_id , <EOL> resource_build_attempts , <EOL> result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceBuildException ( err_msg ) <EOL> return result <EOL> def get_security_group_rule ( self , security_group_rule_id , <EOL> resource_get_attempts = None , <EOL> raise_exception = False , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_get_attempts = ( resource_get_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> err_msg = '<STR_LIT>' <EOL> for attempt in range ( resource_get_attempts ) : <EOL> self . _log . debug ( <EOL> '<STR_LIT>' . format ( <EOL> attempt + <NUM_LIT:1> , resource_get_attempts , <EOL> security_group_rule_id ) ) <EOL> resp = self . client . get_security_group_rule ( <EOL> security_group_rule_id = security_group_rule_id ) <EOL> resp_check = self . check_response ( <EOL> resp = resp , <EOL> status_code = SecurityGroupsResponseCodes . GET_SECURITY_GROUP_RULE , <EOL> label = security_group_rule_id , message = err_msg ) <EOL> result . response = resp <EOL> if not resp_check : <EOL> return result <EOL> result . failures . append ( resp_check ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( security_group_rule_id , resource_get_attempts , <EOL> result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceGetException ( err_msg ) <EOL> return result <EOL> def list_security_group_rules ( self , security_group_rule_id = None , <EOL> security_group_id = None , direction = None , <EOL> ethertype = None , protocol = None , <EOL> port_range_min = None , port_range_max = None , <EOL> remote_group_id = None , remote_ip_prefix = None , <EOL> tenant_id = None , limit = None , marker = None , <EOL> page_reverse = None , <EOL> resource_list_attempts = None , <EOL> raise_exception = False , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_list_attempts = ( resource_list_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> err_msg = '<STR_LIT>' <EOL> for attempt in range ( resource_list_attempts ) : <EOL> self . _log . debug ( <EOL> '<STR_LIT>' . format ( <EOL> attempt + <NUM_LIT:1> , resource_list_attempts ) ) <EOL> resp = self . client . list_security_group_rules ( <EOL> security_group_rule_id = security_group_rule_id , <EOL> security_group_id = security_group_id , direction = direction , <EOL> ethertype = ethertype , protocol = protocol , <EOL> port_range_min = port_range_min , port_range_max = port_range_max , <EOL> remote_group_id = remote_group_id , <EOL> remote_ip_prefix = remote_ip_prefix , tenant_id = tenant_id , <EOL> limit = limit , marker = marker , page_reverse = page_reverse ) <EOL> resp_check = self . check_response ( <EOL> resp = resp , <EOL> status_code = SecurityGroupsResponseCodes . LIST_SECURITY_GROUP_RULES , <EOL> label = '<STR_LIT>' , message = err_msg ) <EOL> result . response = resp <EOL> if not resp_check : <EOL> return result <EOL> result . failures . append ( resp_check ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( resource_list_attempts , result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceListException ( err_msg ) <EOL> return result <EOL> def delete_security_group_rule ( self , security_group_rule_id , <EOL> resource_delete_attempts = None , <EOL> raise_exception = False , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> resource_delete_attempts = ( resource_delete_attempts or <EOL> self . config . api_retries ) <EOL> result = NetworkingResponse ( ) <EOL> for attempt in range ( resource_delete_attempts ) : <EOL> self . _log . debug ( <EOL> '<STR_LIT>' . format ( <EOL> attempt + <NUM_LIT:1> , resource_delete_attempts , <EOL> security_group_rule_id ) ) <EOL> resp = self . client . delete_security_group_rule ( <EOL> security_group_rule_id = security_group_rule_id ) <EOL> result . response = resp <EOL> if ( resp . ok and resp . status_code == <EOL> SecurityGroupsResponseCodes . DELETE_SECURITY_GROUP_RULE ) : <EOL> return result <EOL> del_status_code = SecurityGroupsResponseCodes . DELETE_SECURITY_GROUP_RULE <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( <EOL> security_group_rule = security_group_rule_id , <EOL> expected_status = del_status_code , <EOL> status = resp . status_code , reason = resp . reason , <EOL> content = resp . content ) <EOL> self . _log . error ( err_msg ) <EOL> result . failures . append ( err_msg ) <EOL> time . sleep ( poll_interval ) <EOL> else : <EOL> err_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( security_group_rule_id , resource_delete_attempts , <EOL> result . failures ) <EOL> self . _log . error ( err_msg ) <EOL> if raise_exception : <EOL> raise ResourceDeleteException ( err_msg ) <EOL> return result <EOL> def delete_security_group_rules ( self , security_group_rule_list = None , <EOL> security_group_id = None , tenant_id = None ) : <EOL> """<STR_LIT>""" <EOL> if security_group_rule_list is None : <EOL> resp = self . list_security_group_rules ( <EOL> security_group_id = security_group_id , tenant_id = tenant_id ) <EOL> if ( resp . response . status_code != <EOL> SecurityGroupsResponseCodes . LIST_SECURITY_GROUP_RULES ) : <EOL> get_msg = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . _log . info ( get_msg ) <EOL> return None <EOL> security_group_rules = resp . response . entity <EOL> if security_group_id : <EOL> security_group_rules = self . filter_entity_list_by_attr ( <EOL> entity_list = security_group_rules , attr = '<STR_LIT>' , <EOL> value = security_group_id ) <EOL> if tenant_id : <EOL> security_group_rules = self . filter_entity_list_by_attr ( <EOL> entity_list = security_group_rules , attr = '<STR_LIT>' , <EOL> value = tenant_id ) <EOL> security_group_rule_list = self . get_id_list_from_entity_list ( <EOL> entity_list = security_group_rules ) <EOL> log_msg = '<STR_LIT>' . format ( <EOL> security_group_rule_list ) <EOL> self . _log . info ( log_msg ) <EOL> failed_deletes = [ ] <EOL> for security_group_rule_id in security_group_rule_list : <EOL> result = self . delete_security_group_rule ( <EOL> security_group_rule_id = security_group_rule_id ) <EOL> if result . failures : <EOL> failed_deletes . append ( result . failures ) <EOL> return failed_deletes <EOL> def clean_security_group_rule ( self , security_group_rule_id , timeout = None , <EOL> poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> timeout = timeout or self . config . resource_delete_timeout <EOL> poll_interval = poll_interval or self . config . api_poll_interval <EOL> endtime = time . time ( ) + int ( timeout ) <EOL> log_msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( security_group_rule_id , timeout ) <EOL> self . _log . info ( log_msg ) <EOL> resp = None <EOL> while time . time ( ) < endtime : <EOL> try : <EOL> self . client . delete_security_group_rule ( <EOL> security_group_rule_id = security_group_rule_id ) <EOL> resp = self . client . get_security_group_rule ( <EOL> security_group_rule_id = security_group_rule_id ) <EOL> except Exception as err : <EOL> err_msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( err ) <EOL> self . _log . error ( err_msg ) <EOL> if ( resp is not None and <EOL> resp . status_code == SecurityGroupsResponseCodes . NOT_FOUND ) : <EOL> return None <EOL> time . sleep ( poll_interval ) <EOL> err_msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( security_group_rule_id , timeout ) <EOL> self . _log . error ( err_msg ) <EOL> return security_group_rule_id <EOL> def clean_security_group_rules ( self , security_group_rule_list , <EOL> timeout = None , poll_interval = None ) : <EOL> """<STR_LIT>""" <EOL> log_msg = '<STR_LIT>' . format ( <EOL> security_group_rule_list ) <EOL> self . _log . info ( log_msg ) <EOL> undeleted_security_group_rules = [ ] <EOL> for security_group_rule in security_group_rule_list : <EOL> result = self . clean_security_group_rule ( <EOL> security_group_rule_id = security_group_rule , timeout = timeout , <EOL> poll_interval = poll_interval ) <EOL> if result : <EOL> undeleted_security_group_rules . append ( result ) <EOL> if undeleted_security_group_rules : <EOL> err_msg = '<STR_LIT>' . format ( <EOL> undeleted_security_group_rules ) <EOL> self . _log . error ( err_msg ) <EOL> return undeleted_security_group_rules </s>
