<s> import time <EOL> from oslo_log import log as logging <EOL> from oslo_utils import timeutils <EOL> from saharaclient . api import base as sab <EOL> from saharaclient import client as sahara_client <EOL> from tempest import config <EOL> from tempest import exceptions <EOL> from tempest . scenario import manager <EOL> TEMPEST_CONF = config . CONF <EOL> CLUSTER_STATUS_ACTIVE = "<STR_LIT>" <EOL> CLUSTER_STATUS_ERROR = "<STR_LIT>" <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class BaseDataProcessingTest ( manager . ScenarioTest ) : <EOL> @ classmethod <EOL> def resource_setup ( cls ) : <EOL> cls . set_network_resources ( ) <EOL> super ( BaseDataProcessingTest , cls ) . resource_setup ( ) <EOL> endpoint_type = TEMPEST_CONF . data_processing . endpoint_type <EOL> catalog_type = TEMPEST_CONF . data_processing . catalog_type <EOL> auth_url = TEMPEST_CONF . identity . uri <EOL> credentials = cls . os_primary . credentials <EOL> cls . client = sahara_client . Client ( <EOL> TEMPEST_CONF . data_processing . saharaclient_version , <EOL> credentials . username , <EOL> credentials . password , <EOL> project_name = credentials . tenant_name , <EOL> endpoint_type = endpoint_type , <EOL> service_type = catalog_type , <EOL> auth_url = auth_url , <EOL> sahara_url = TEMPEST_CONF . data_processing . sahara_url ) <EOL> cls . object_client = cls . os_primary . object_client <EOL> cls . container_client = cls . os_primary . container_client <EOL> cls . networks_client = cls . os_primary . compute_networks_client <EOL> cls . floating_ip_pool = TEMPEST_CONF . network . floating_network_name <EOL> if TEMPEST_CONF . service_available . neutron : <EOL> cls . floating_ip_pool = cls . get_floating_ip_pool_id_for_neutron ( ) <EOL> cls . worker_template = { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] , <EOL> '<STR_LIT>' : TEMPEST_CONF . compute . flavor_ref , <EOL> '<STR_LIT>' : cls . floating_ip_pool <EOL> } <EOL> cls . master_template = { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] , <EOL> '<STR_LIT>' : TEMPEST_CONF . compute . flavor_ref , <EOL> '<STR_LIT>' : cls . floating_ip_pool , <EOL> '<STR_LIT>' : True <EOL> } <EOL> cls . cluster_template = { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> cls . swift_data_source_with_creds = { <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:user>' : '<STR_LIT:test>' , <EOL> '<STR_LIT:password>' : '<STR_LIT>' <EOL> } <EOL> } <EOL> cls . local_hdfs_data_source = { <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> } <EOL> cls . external_hdfs_data_source = { <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' <EOL> } <EOL> @ classmethod <EOL> def get_floating_ip_pool_id_for_neutron ( cls ) : <EOL> net_id = cls . _find_network_by_name ( <EOL> TEMPEST_CONF . network . floating_network_name ) <EOL> if not net_id : <EOL> raise exceptions . NotFound ( <EOL> '<STR_LIT>' <EOL> % TEMPEST_CONF . network . floating_network_name ) <EOL> return net_id <EOL> @ classmethod <EOL> def get_private_network_id ( cls ) : <EOL> net_id = cls . _find_network_by_name ( <EOL> TEMPEST_CONF . compute . fixed_network_name ) <EOL> if not net_id : <EOL> raise exceptions . NotFound ( <EOL> '<STR_LIT>' <EOL> % TEMPEST_CONF . compute . fixed_network_name ) <EOL> return net_id <EOL> @ classmethod <EOL> def _find_network_by_name ( cls , network_name ) : <EOL> for network in cls . networks_client . list_networks ( ) [ '<STR_LIT>' ] : <EOL> if network [ '<STR_LIT:label>' ] == network_name : <EOL> return network [ '<STR_LIT:id>' ] <EOL> return None <EOL> def create_node_group_template ( self , name , ** kwargs ) : <EOL> resp_body = self . client . node_group_templates . create ( <EOL> name , ** kwargs ) <EOL> self . addCleanup ( self . delete_resource , <EOL> self . client . node_group_templates , resp_body . id ) <EOL> return resp_body <EOL> def create_cluster_template ( self , name , ** kwargs ) : <EOL> resp_body = self . client . cluster_templates . create ( <EOL> name , ** kwargs ) <EOL> self . addCleanup ( self . delete_resource , <EOL> self . client . cluster_templates , resp_body . id ) <EOL> return resp_body <EOL> def create_data_source ( self , name , url , description , type , <EOL> credentials = None ) : <EOL> user = credentials [ '<STR_LIT:user>' ] if credentials else None <EOL> pas = credentials [ '<STR_LIT:password>' ] if credentials else None <EOL> resp_body = self . client . data_sources . create ( <EOL> name , description , type , url , credential_user = user , <EOL> credential_pass = pas ) <EOL> self . addCleanup ( self . delete_resource , <EOL> self . client . data_sources , resp_body . id ) <EOL> return resp_body <EOL> def create_job_binary ( self , name , url , description , extra = None ) : <EOL> resp_body = self . client . job_binaries . create ( <EOL> name , url , description , extra ) <EOL> self . addCleanup ( self . delete_resource , <EOL> self . client . job_binaries , resp_body . id ) <EOL> return resp_body <EOL> def create_job_binary_internal ( self , name , data ) : <EOL> resp_body = self . client . job_binary_internals . create ( name , data ) <EOL> self . addCleanup ( self . delete_resource , <EOL> self . client . job_binary_internals , resp_body . id ) <EOL> return resp_body <EOL> def create_job ( self , name , job_type , mains , libs = None , description = None ) : <EOL> libs = libs or ( ) <EOL> description = description or '<STR_LIT>' <EOL> resp_body = self . client . jobs . create ( <EOL> name , job_type , mains , libs , description ) <EOL> self . addCleanup ( self . delete_resource , self . client . jobs , resp_body . id ) <EOL> return resp_body <EOL> def create_cluster ( self , name , ** kwargs ) : <EOL> resp_body = self . client . clusters . create ( name , ** kwargs ) <EOL> self . addCleanup ( self . delete_resource , self . client . clusters , <EOL> resp_body . id ) <EOL> return resp_body <EOL> def check_cluster_active ( self , cluster_id ) : <EOL> timeout = TEMPEST_CONF . data_processing . cluster_timeout <EOL> s_time = timeutils . utcnow ( ) <EOL> while timeutils . delta_seconds ( s_time , timeutils . utcnow ( ) ) < timeout : <EOL> cluster = self . client . clusters . get ( cluster_id ) <EOL> if cluster . status == CLUSTER_STATUS_ACTIVE : <EOL> return <EOL> if cluster . status == CLUSTER_STATUS_ERROR : <EOL> raise exceptions . BuildErrorException ( <EOL> '<STR_LIT>' % <EOL> CLUSTER_STATUS_ERROR ) <EOL> time . sleep ( TEMPEST_CONF . data_processing . request_timeout ) <EOL> raise exceptions . TimeoutException ( <EOL> '<STR_LIT>' <EOL> % ( CLUSTER_STATUS_ACTIVE , timeout ) ) <EOL> def create_job_execution ( self , ** kwargs ) : <EOL> resp_body = self . client . job_executions . create ( ** kwargs ) <EOL> self . addCleanup ( self . delete_resource , self . client . job_executions , <EOL> resp_body . id ) <EOL> return resp_body <EOL> def create_container ( self , name ) : <EOL> self . container_client . create_container ( name ) <EOL> self . addCleanup ( self . delete_swift_container , name ) <EOL> def delete_resource ( self , resource_client , resource_id ) : <EOL> try : <EOL> resource_client . delete ( resource_id ) <EOL> except sab . APIException : <EOL> pass <EOL> else : <EOL> self . delete_timeout ( resource_client , resource_id ) <EOL> def delete_timeout ( <EOL> self , resource_client , resource_id , <EOL> timeout = TEMPEST_CONF . data_processing . cluster_timeout ) : <EOL> start = timeutils . utcnow ( ) <EOL> while timeutils . delta_seconds ( start , timeutils . utcnow ( ) ) < timeout : <EOL> try : <EOL> resource_client . get ( resource_id ) <EOL> except sab . APIException as sahara_api_exception : <EOL> if '<STR_LIT>' in sahara_api_exception . message : <EOL> return <EOL> raise sahara_api_exception <EOL> time . sleep ( TEMPEST_CONF . data_processing . request_timeout ) <EOL> raise exceptions . TimeoutException ( <EOL> '<STR_LIT>' <EOL> % ( resource_id , timeout ) ) <EOL> def delete_swift_container ( self , container ) : <EOL> objects = ( [ obj [ '<STR_LIT:name>' ] for obj in <EOL> self . container_client . list_all_container_objects ( <EOL> container ) ] ) <EOL> for obj in objects : <EOL> self . object_client . delete_object ( container , obj ) <EOL> self . container_client . delete_container ( container ) </s>
