<s> """<STR_LIT>""" <EOL> import json <EOL> import os <EOL> import re <EOL> from time import sleep <EOL> from nose . tools import nottest <EOL> from prestoadmin . util import constants <EOL> from tests . base_test_case import BaseTestCase <EOL> from tests . configurable_cluster import ConfigurableCluster <EOL> from tests . docker_cluster import DockerCluster , DockerClusterException <EOL> from tests . product . mode_installers import StandaloneModeInstaller , YarnSliderModeInstaller <EOL> from tests . product . prestoadmin_installer import PrestoadminInstaller <EOL> from tests . product . standalone . presto_installer import StandalonePrestoInstaller <EOL> from tests . product . topology_installer import TopologyInstaller <EOL> from tests . product . yarn_slider . slider_installer import SliderInstaller <EOL> PRESTO_VERSION = r'<STR_LIT>' <EOL> RETRY_TIMEOUT = <NUM_LIT> <EOL> RETRY_INTERVAL = <NUM_LIT:5> <EOL> class BaseProductTestCase ( BaseTestCase ) : <EOL> STANDALONE_BARE_CLUSTER = '<STR_LIT>' <EOL> BARE_CLUSTER = '<STR_LIT>' <EOL> PA_ONLY_CLUSTER = '<STR_LIT>' <EOL> STANDALONE_PRESTO_CLUSTER = '<STR_LIT>' <EOL> PA_ONLY_YS_CLUSTER = '<STR_LIT>' <EOL> PA_SLIDER_CLUSTER = '<STR_LIT>' <EOL> _cluster_types = { <EOL> BARE_CLUSTER : [ ] , <EOL> PA_ONLY_CLUSTER : [ PrestoadminInstaller , <EOL> StandaloneModeInstaller ] , <EOL> STANDALONE_PRESTO_CLUSTER : [ PrestoadminInstaller , <EOL> StandaloneModeInstaller , <EOL> TopologyInstaller , <EOL> StandalonePrestoInstaller ] , <EOL> PA_ONLY_YS_CLUSTER : [ PrestoadminInstaller , <EOL> YarnSliderModeInstaller ] , <EOL> PA_SLIDER_CLUSTER : [ PrestoadminInstaller , <EOL> YarnSliderModeInstaller , <EOL> SliderInstaller ] <EOL> } <EOL> default_workers_config_ = """<STR_LIT>""" <EOL> default_workers_test_config_ = """<STR_LIT>""" <EOL> default_node_properties_ = """<STR_LIT>""" <EOL> default_jvm_config_ = """<STR_LIT>""" <EOL> default_coordinator_config_ = """<STR_LIT>""" <EOL> default_coordinator_test_config_ = """<STR_LIT>""" <EOL> down_node_connection_string = r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' <EOL> status_down_node_string = r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' r'<STR_LIT>' <EOL> len_down_node_error = <NUM_LIT:6> <EOL> def setUp ( self ) : <EOL> super ( BaseProductTestCase , self ) . setUp ( ) <EOL> self . maxDiff = None <EOL> self . cluster = None <EOL> self . default_keywords = { } <EOL> def _run_installers ( self , installers ) : <EOL> cluster = self . cluster <EOL> for installer in installers : <EOL> dependencies = installer . get_dependencies ( ) <EOL> for dependency in dependencies : <EOL> dependency . assert_installed ( self ) <EOL> installer_instance = installer ( self ) <EOL> installer_instance . install ( ) <EOL> self . default_keywords . update ( installer_instance . get_keywords ( ) ) <EOL> cluster . postinstall ( installer ) <EOL> def _apply_post_install_hooks ( self , installers ) : <EOL> for installer in installers : <EOL> self . cluster . postinstall ( installer ) <EOL> def _update_replacement_keywords ( self , installers ) : <EOL> for installer in installers : <EOL> installer_instance = installer ( self ) <EOL> self . default_keywords . update ( installer_instance . get_keywords ( ) ) <EOL> def setup_cluster ( self , bare_image_provider , cluster_type ) : <EOL> try : <EOL> installers = self . _cluster_types [ cluster_type ] <EOL> except KeyError : <EOL> self . fail ( <EOL> '<STR_LIT>' % <EOL> ( cluster_type , '<STR_LIT:U+002CU+0020>' . join ( self . _cluster_types . keys ( ) ) ) ) <EOL> config_filename = ConfigurableCluster . check_for_cluster_config ( ) <EOL> if config_filename : <EOL> self . cluster = ConfigurableCluster . start_bare_cluster ( <EOL> config_filename , self , <EOL> StandalonePrestoInstaller . assert_installed ) <EOL> else : <EOL> try : <EOL> self . cluster = DockerCluster . start_existing_images ( <EOL> bare_image_provider , cluster_type ) <EOL> if self . cluster : <EOL> self . _apply_post_install_hooks ( installers ) <EOL> self . _update_replacement_keywords ( installers ) <EOL> return <EOL> self . cluster = DockerCluster . start_bare_cluster ( <EOL> bare_image_provider ) <EOL> except DockerClusterException as e : <EOL> self . fail ( e . msg ) <EOL> self . _run_installers ( installers ) <EOL> if isinstance ( self . cluster , DockerCluster ) : <EOL> self . cluster . commit_images ( bare_image_provider , cluster_type ) <EOL> def tearDown ( self ) : <EOL> self . restore_stdout_stderr_keep_open ( ) <EOL> if self . cluster : <EOL> self . cluster . tear_down ( ) <EOL> super ( BaseProductTestCase , self ) . tearDown ( ) <EOL> def dump_and_cp_topology ( self , topology , cluster = None ) : <EOL> if not cluster : <EOL> cluster = self . cluster <EOL> cluster . write_content_to_host ( <EOL> json . dumps ( topology ) , <EOL> '<STR_LIT>' , <EOL> cluster . master <EOL> ) <EOL> def upload_topology ( self , topology = None , cluster = None ) : <EOL> if not cluster : <EOL> cluster = self . cluster <EOL> if not topology : <EOL> topology = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] } <EOL> self . dump_and_cp_topology ( topology , cluster ) <EOL> @ nottest <EOL> def write_test_configs ( self , cluster , extra_configs = None , <EOL> coordinator = None ) : <EOL> if not coordinator : <EOL> coordinator = self . cluster . internal_master <EOL> config = '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' % coordinator <EOL> if extra_configs : <EOL> config += '<STR_LIT:\n>' + extra_configs <EOL> coordinator_config = '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' % config <EOL> workers_config = '<STR_LIT>' % config <EOL> cluster . write_content_to_host ( <EOL> coordinator_config , <EOL> os . path . join ( constants . COORDINATOR_DIR , '<STR_LIT>' ) , <EOL> cluster . master <EOL> ) <EOL> cluster . write_content_to_host ( <EOL> workers_config , <EOL> os . path . join ( constants . WORKERS_DIR , '<STR_LIT>' ) , <EOL> cluster . master <EOL> ) <EOL> def fetch_log_tail ( self , lines = <NUM_LIT:50> ) : <EOL> return self . cluster . exec_cmd_on_host ( <EOL> self . cluster . get_master ( ) , <EOL> '<STR_LIT>' % ( lines , ) , <EOL> raise_error = False ) <EOL> def run_prestoadmin ( self , command , raise_error = True , cluster = None , <EOL> ** kwargs ) : <EOL> if not cluster : <EOL> cluster = self . cluster <EOL> command = self . replace_keywords ( command , cluster = cluster , ** kwargs ) <EOL> return cluster . exec_cmd_on_host ( <EOL> cluster . master , <EOL> "<STR_LIT>" % command , <EOL> raise_error = raise_error <EOL> ) <EOL> def run_script_from_prestoadmin_dir ( self , script_contents , host = '<STR_LIT>' , <EOL> raise_error = True , ** kwargs ) : <EOL> if not host : <EOL> host = self . cluster . master <EOL> script_contents = self . replace_keywords ( script_contents , <EOL> ** kwargs ) <EOL> temp_script = '<STR_LIT>' <EOL> self . cluster . write_content_to_host ( <EOL> '<STR_LIT>' % script_contents , <EOL> temp_script , host ) <EOL> self . cluster . exec_cmd_on_host ( <EOL> host , '<STR_LIT>' % temp_script ) <EOL> return self . cluster . exec_cmd_on_host ( <EOL> host , temp_script , raise_error = raise_error ) <EOL> def run_prestoadmin_expect ( self , command , expect_statements ) : <EOL> temp_script = '<STR_LIT>' <EOL> script_content = '<STR_LIT>' + '<STR_LIT>' % ( command , expect_statements ) <EOL> self . cluster . write_content_to_host ( script_content , temp_script , <EOL> self . cluster . master ) <EOL> self . cluster . exec_cmd_on_host ( <EOL> self . cluster . master , '<STR_LIT>' % temp_script ) <EOL> return self . cluster . exec_cmd_on_host ( <EOL> self . cluster . master , temp_script ) <EOL> def assert_path_exists ( self , host , file_path ) : <EOL> self . cluster . exec_cmd_on_host ( <EOL> host , '<STR_LIT>' % file_path ) <EOL> def get_file_content ( self , host , filepath ) : <EOL> return self . cluster . exec_cmd_on_host ( host , '<STR_LIT>' % ( filepath ) ) <EOL> def assert_config_perms ( self , host , filepath ) : <EOL> self . assert_file_perm_owner ( <EOL> host , filepath , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def assert_file_perm_owner ( <EOL> self , host , filepath , permissions , owner , group ) : <EOL> ls = self . cluster . exec_cmd_on_host ( host , "<STR_LIT>" % ( filepath , ) ) <EOL> fields = ls . split ( ) <EOL> self . assertEqual ( fields [ <NUM_LIT:0> ] , permissions ) <EOL> self . assertEqual ( fields [ <NUM_LIT:2> ] , owner ) <EOL> self . assertEqual ( fields [ <NUM_LIT:3> ] , group ) <EOL> def assert_file_content ( self , host , filepath , expected ) : <EOL> content = self . get_file_content ( host , filepath ) <EOL> split_path = os . path . split ( filepath ) <EOL> pa_file = None <EOL> if ( split_path [ <NUM_LIT:0> ] == '<STR_LIT>' and <EOL> split_path [ <NUM_LIT:1> ] in [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) : <EOL> if host in self . cluster . slaves : <EOL> config_dir = constants . WORKERS_DIR <EOL> else : <EOL> config_dir = constants . COORDINATOR_DIR <EOL> pa_file = os . path . join ( config_dir , split_path [ <NUM_LIT:1> ] ) <EOL> self . assertLazyMessage ( <EOL> lambda : self . file_content_message ( content , expected , pa_file ) , <EOL> self . assertEqual , <EOL> content , <EOL> expected ) <EOL> def file_content_message ( self , actual , expected , pa_file ) : <EOL> msg = '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' % ( actual , expected ) <EOL> if pa_file : <EOL> try : <EOL> msg += '<STR_LIT>' % ( pa_file , ) <EOL> msg += self . get_file_content ( self . cluster . get_master ( ) , <EOL> pa_file ) <EOL> msg += '<STR_LIT>' <EOL> except OSError as e : <EOL> msg += e . message <EOL> return msg <EOL> def assert_file_content_regex ( self , host , filepath , expected ) : <EOL> config = self . get_file_content ( host , filepath ) <EOL> self . assertRegexpMatches ( config , expected ) <EOL> def assert_has_default_connector ( self , container ) : <EOL> filepath = '<STR_LIT>' <EOL> self . assert_config_perms ( container , filepath ) <EOL> self . assert_file_content ( container , filepath , '<STR_LIT>' ) <EOL> def assert_has_jmx_connector ( self , container ) : <EOL> self . assert_file_content ( container , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def assert_path_removed ( self , container , directory ) : <EOL> self . cluster . exec_cmd_on_host ( <EOL> container , '<STR_LIT>' % directory ) <EOL> def assert_has_default_config ( self , host ) : <EOL> jvm_config_path = '<STR_LIT>' <EOL> self . assert_config_perms ( host , jvm_config_path ) <EOL> self . assert_file_content ( <EOL> host , jvm_config_path , self . default_jvm_config_ ) <EOL> self . assert_node_config ( host , self . default_node_properties_ ) <EOL> config_properties_path = os . path . join ( constants . REMOTE_CONF_DIR , <EOL> '<STR_LIT>' ) <EOL> self . assert_config_perms ( host , config_properties_path ) <EOL> if host in self . cluster . slaves : <EOL> self . assert_file_content ( host , config_properties_path , <EOL> self . default_workers_test_config_ ) <EOL> else : <EOL> self . assert_file_content ( host , config_properties_path , <EOL> self . default_coordinator_test_config_ ) <EOL> def assert_node_config ( self , host , expected ) : <EOL> node_properties_path = '<STR_LIT>' <EOL> self . assert_config_perms ( host , node_properties_path ) <EOL> node_properties = self . cluster . exec_cmd_on_host ( <EOL> host , '<STR_LIT>' % ( node_properties_path , ) ) <EOL> split_properties = node_properties . split ( '<STR_LIT:\n>' , <NUM_LIT:1> ) <EOL> self . assertRegexpMatches ( split_properties [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> actual = split_properties [ <NUM_LIT:1> ] <EOL> if host in self . cluster . slaves : <EOL> conf_dir = constants . COORDINATOR_DIR <EOL> else : <EOL> conf_dir = constants . WORKERS_DIR <EOL> self . assertLazyMessage ( <EOL> lambda : self . file_content_message ( actual , expected , <EOL> os . path . join ( conf_dir , <EOL> '<STR_LIT>' ) ) , <EOL> self . assertEqual , <EOL> actual , <EOL> expected ) <EOL> def expected_stop ( self , running = None , not_running = None ) : <EOL> if running is None : <EOL> running = self . cluster . all_internal_hosts ( ) <EOL> if not_running : <EOL> for host in not_running : <EOL> running . remove ( host ) <EOL> expected_output = [ ] <EOL> for host in running : <EOL> expected_output += [ r'<STR_LIT>' % host , <EOL> r'<STR_LIT>' % host , <EOL> r'<STR_LIT>' % host ] <EOL> if not_running : <EOL> for host in not_running : <EOL> expected_output += [ r'<STR_LIT>' % host , <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' % host , <EOL> r'<STR_LIT>' % host ] <EOL> return expected_output <EOL> def assert_stopped ( self , process_per_host ) : <EOL> for host , pid in process_per_host : <EOL> self . retry ( lambda : <EOL> self . assertRaisesRegexp ( OSError , <EOL> '<STR_LIT>' , <EOL> self . cluster . exec_cmd_on_host , <EOL> host , <EOL> '<STR_LIT>' % pid ) , <EOL> retry_timeout = <NUM_LIT:10> , <EOL> retry_interval = <NUM_LIT:2> ) <EOL> def get_process_per_host ( self , output_lines ) : <EOL> process_per_host = [ ] <EOL> if not hasattr ( output_lines , '<STR_LIT>' ) : <EOL> raise Exception ( '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> for line in output_lines : <EOL> match = re . search ( r'<STR_LIT>' , <EOL> line ) <EOL> if match : <EOL> process_per_host . append ( ( match . group ( '<STR_LIT:host>' ) , <EOL> match . group ( '<STR_LIT>' ) ) ) <EOL> return process_per_host <EOL> def assert_started ( self , process_per_host ) : <EOL> for host , pid in process_per_host : <EOL> self . cluster . exec_cmd_on_host ( host , '<STR_LIT>' % pid ) <EOL> return process_per_host <EOL> def replace_keywords ( self , text , cluster = None , ** kwargs ) : <EOL> if not cluster : <EOL> cluster = self . cluster <EOL> test_keywords = self . default_keywords . copy ( ) <EOL> test_keywords . update ( { <EOL> '<STR_LIT>' : cluster . internal_master <EOL> } ) <EOL> if cluster . internal_slaves : <EOL> test_keywords . update ( { <EOL> '<STR_LIT>' : cluster . internal_slaves [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : cluster . internal_slaves [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : cluster . internal_slaves [ <NUM_LIT:2> ] <EOL> } ) <EOL> test_keywords . update ( ** kwargs ) <EOL> return text % test_keywords <EOL> def escape_for_regex ( self , expected ) : <EOL> expected = expected . replace ( '<STR_LIT:[>' , '<STR_LIT>' ) <EOL> expected = expected . replace ( '<STR_LIT:]>' , '<STR_LIT>' ) <EOL> expected = expected . replace ( '<STR_LIT:)>' , '<STR_LIT>' ) <EOL> expected = expected . replace ( '<STR_LIT:(>' , '<STR_LIT>' ) <EOL> expected = expected . replace ( '<STR_LIT:+>' , '<STR_LIT>' ) <EOL> return expected <EOL> def retry ( self , method_to_check , retry_timeout = RETRY_TIMEOUT , <EOL> retry_interval = RETRY_INTERVAL ) : <EOL> time_spent_waiting = <NUM_LIT:0> <EOL> while time_spent_waiting <= retry_timeout : <EOL> try : <EOL> result = method_to_check ( ) <EOL> return result <EOL> except ( AssertionError , PrestoError , OSError ) : <EOL> pass <EOL> sleep ( retry_interval ) <EOL> time_spent_waiting += retry_interval <EOL> return method_to_check ( ) <EOL> def down_node_connection_error ( self , host ) : <EOL> hostname = self . cluster . get_down_hostname ( host ) <EOL> return self . down_node_connection_string % { '<STR_LIT:host>' : hostname } <EOL> def status_node_connection_error ( self , host ) : <EOL> hostname = self . cluster . get_down_hostname ( host ) <EOL> return self . status_down_node_string % { '<STR_LIT:host>' : hostname } <EOL> def docker_only ( original_function ) : <EOL> def test_inner ( self , * args , ** kwargs ) : <EOL> if type ( getattr ( self , '<STR_LIT>' ) ) is DockerCluster : <EOL> original_function ( self , * args , ** kwargs ) <EOL> else : <EOL> print '<STR_LIT>' <EOL> return test_inner <EOL> class PrestoError ( Exception ) : <EOL> pass </s>
