<s> """<STR_LIT>""" <EOL> import re <EOL> from os import walk , environ <EOL> from os . path import exists , join , dirname , pardir <EOL> import yaml <EOL> from shutil import rmtree <EOL> from elasticsearch import TransportError <EOL> from elasticsearch . compat import string_types <EOL> from elasticsearch . helpers . test import _get_version <EOL> from . . test_cases import SkipTest <EOL> from . import ElasticsearchTestCase <EOL> PARAMS_RENAMES = { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> CATCH_CODES = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> IMPLEMENTED_FEATURES = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> SKIP_TESTS = { <EOL> ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> ) : set ( ( '<STR_LIT>' , ) ) , <EOL> '<STR_LIT:*>' : set ( ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> } <EOL> class InvalidActionType ( Exception ) : <EOL> pass <EOL> class YamlTestCase ( ElasticsearchTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( YamlTestCase , self ) . setUp ( ) <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . run_code ( self . _setup_code ) <EOL> self . last_response = None <EOL> self . _state = { } <EOL> def tearDown ( self ) : <EOL> super ( YamlTestCase , self ) . tearDown ( ) <EOL> for repo , definition in self . client . snapshot . get_repository ( repository = '<STR_LIT>' ) . items ( ) : <EOL> self . client . snapshot . delete_repository ( repository = repo ) <EOL> if definition [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> rmtree ( '<STR_LIT>' % definition [ '<STR_LIT>' ] [ '<STR_LIT:location>' ] ) <EOL> def _resolve ( self , value ) : <EOL> if isinstance ( value , string_types ) and value . startswith ( '<STR_LIT:$>' ) : <EOL> value = value [ <NUM_LIT:1> : ] <EOL> self . assertIn ( value , self . _state ) <EOL> value = self . _state [ value ] <EOL> if isinstance ( value , string_types ) : <EOL> value = value . strip ( ) <EOL> elif isinstance ( value , dict ) : <EOL> value = dict ( ( k , self . _resolve ( v ) ) for ( k , v ) in value . items ( ) ) <EOL> elif isinstance ( value , list ) : <EOL> value = list ( map ( self . _resolve , value ) ) <EOL> return value <EOL> def _lookup ( self , path ) : <EOL> value = self . last_response <EOL> if path == '<STR_LIT>' : <EOL> return value <EOL> path = path . replace ( r'<STR_LIT>' , '<STR_LIT>' ) <EOL> for step in path . split ( '<STR_LIT:.>' ) : <EOL> if not step : <EOL> continue <EOL> step = step . replace ( '<STR_LIT>' , '<STR_LIT:.>' ) <EOL> step = self . _resolve ( step ) <EOL> if step . isdigit ( ) and step not in value : <EOL> step = int ( step ) <EOL> self . assertIsInstance ( value , list ) <EOL> self . assertGreater ( len ( value ) , step ) <EOL> else : <EOL> self . assertIn ( step , value ) <EOL> value = value [ step ] <EOL> return value <EOL> def run_code ( self , test ) : <EOL> """<STR_LIT>""" <EOL> for action in test : <EOL> self . assertEquals ( <NUM_LIT:1> , len ( action ) ) <EOL> action_type , action = list ( action . items ( ) ) [ <NUM_LIT:0> ] <EOL> if hasattr ( self , '<STR_LIT>' + action_type ) : <EOL> getattr ( self , '<STR_LIT>' + action_type ) ( action ) <EOL> else : <EOL> raise InvalidActionType ( action_type ) <EOL> def run_do ( self , action ) : <EOL> """<STR_LIT>""" <EOL> catch = action . pop ( '<STR_LIT>' , None ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( action ) ) <EOL> method , args = list ( action . items ( ) ) [ <NUM_LIT:0> ] <EOL> api = self . client <EOL> for m in method . split ( '<STR_LIT:.>' ) : <EOL> self . assertTrue ( hasattr ( api , m ) ) <EOL> api = getattr ( api , m ) <EOL> for k in PARAMS_RENAMES : <EOL> if k in args : <EOL> args [ PARAMS_RENAMES [ k ] ] = args . pop ( k ) <EOL> for k in args : <EOL> args [ k ] = self . _resolve ( args [ k ] ) <EOL> try : <EOL> self . last_response = api ( ** args ) <EOL> except Exception as e : <EOL> if not catch : <EOL> raise <EOL> self . run_catch ( catch , e ) <EOL> else : <EOL> if catch : <EOL> raise AssertionError ( '<STR_LIT>' % ( catch , self . last_response ) ) <EOL> def _get_nodes ( self ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . _node_info = list ( self . client . nodes . info ( node_id = '<STR_LIT>' , metric = '<STR_LIT>' ) [ '<STR_LIT>' ] . values ( ) ) <EOL> return self . _node_info <EOL> def _get_data_nodes ( self ) : <EOL> return len ( [ info for info in self . _get_nodes ( ) if info . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT:data>' , '<STR_LIT:true>' ) == '<STR_LIT:true>' ] ) <EOL> def _get_benchmark_nodes ( self ) : <EOL> return len ( [ info for info in self . _get_nodes ( ) if info . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT>' , '<STR_LIT:false>' ) == '<STR_LIT:true>' ] ) <EOL> def run_skip ( self , skip ) : <EOL> if '<STR_LIT>' in skip : <EOL> if skip [ '<STR_LIT>' ] in IMPLEMENTED_FEATURES : <EOL> return <EOL> elif skip [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> if self . _get_data_nodes ( ) > <NUM_LIT:1> : <EOL> return <EOL> elif skip [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> if self . _get_benchmark_nodes ( ) : <EOL> return <EOL> raise SkipTest ( skip . get ( '<STR_LIT>' , '<STR_LIT>' % skip [ '<STR_LIT>' ] ) ) <EOL> if '<STR_LIT:version>' in skip : <EOL> version , reason = skip [ '<STR_LIT:version>' ] , skip [ '<STR_LIT>' ] <EOL> if version == '<STR_LIT:all>' : <EOL> raise SkipTest ( reason ) <EOL> min_version , max_version = version . split ( '<STR_LIT:->' ) <EOL> min_version = _get_version ( min_version ) or ( <NUM_LIT:0> , ) <EOL> max_version = _get_version ( max_version ) or ( <NUM_LIT> , ) <EOL> if min_version <= self . es_version <= max_version : <EOL> raise SkipTest ( reason ) <EOL> def run_catch ( self , catch , exception ) : <EOL> if catch == '<STR_LIT>' : <EOL> self . assertIsInstance ( exception , TypeError ) <EOL> return <EOL> self . assertIsInstance ( exception , TransportError ) <EOL> if catch in CATCH_CODES : <EOL> self . assertEquals ( CATCH_CODES [ catch ] , exception . status_code ) <EOL> elif catch [ <NUM_LIT:0> ] == '<STR_LIT:/>' and catch [ - <NUM_LIT:1> ] == '<STR_LIT:/>' : <EOL> self . assertTrue ( re . search ( catch [ <NUM_LIT:1> : - <NUM_LIT:1> ] , exception . error + '<STR_LIT:U+0020>' + repr ( exception . info ) ) , '<STR_LIT>' % ( catch , exception . info ) ) <EOL> self . last_response = exception . info <EOL> def run_gt ( self , action ) : <EOL> for key , value in action . items ( ) : <EOL> self . assertGreater ( self . _lookup ( key ) , value ) <EOL> def run_gte ( self , action ) : <EOL> for key , value in action . items ( ) : <EOL> self . assertGreaterEqual ( self . _lookup ( key ) , value ) <EOL> def run_lt ( self , action ) : <EOL> for key , value in action . items ( ) : <EOL> self . assertLess ( self . _lookup ( key ) , value ) <EOL> def run_lte ( self , action ) : <EOL> for key , value in action . items ( ) : <EOL> self . assertLessEqual ( self . _lookup ( key ) , value ) <EOL> def run_set ( self , action ) : <EOL> for key , value in action . items ( ) : <EOL> self . _state [ value ] = self . _lookup ( key ) <EOL> def run_is_false ( self , action ) : <EOL> try : <EOL> value = self . _lookup ( action ) <EOL> except AssertionError : <EOL> pass <EOL> else : <EOL> self . assertIn ( value , ( '<STR_LIT>' , None , False , <NUM_LIT:0> ) ) <EOL> def run_is_true ( self , action ) : <EOL> value = self . _lookup ( action ) <EOL> self . assertNotIn ( value , ( '<STR_LIT>' , None , False , <NUM_LIT:0> ) ) <EOL> def run_length ( self , action ) : <EOL> for path , expected in action . items ( ) : <EOL> value = self . _lookup ( path ) <EOL> expected = self . _resolve ( expected ) <EOL> self . assertEquals ( expected , len ( value ) ) <EOL> def run_match ( self , action ) : <EOL> for path , expected in action . items ( ) : <EOL> value = self . _lookup ( path ) <EOL> expected = self . _resolve ( expected ) <EOL> if isinstance ( expected , string_types ) and expected . startswith ( '<STR_LIT:/>' ) and expected . endswith ( '<STR_LIT:/>' ) : <EOL> expected = re . compile ( expected [ <NUM_LIT:1> : - <NUM_LIT:1> ] , re . VERBOSE ) <EOL> self . assertTrue ( expected . search ( value ) ) <EOL> else : <EOL> self . assertEquals ( expected , value ) <EOL> def construct_case ( filename , name ) : <EOL> """<STR_LIT>""" <EOL> def make_test ( test_name , definition , i ) : <EOL> def m ( self ) : <EOL> if name in SKIP_TESTS . get ( self . es_version , ( ) ) or name in SKIP_TESTS . get ( '<STR_LIT:*>' , ( ) ) : <EOL> raise SkipTest ( ) <EOL> self . run_code ( definition ) <EOL> m . __doc__ = '<STR_LIT>' % ( <EOL> __name__ , name , i , '<STR_LIT:/>' . join ( filename . split ( '<STR_LIT:/>' ) [ - <NUM_LIT:2> : ] ) , test_name ) <EOL> m . __name__ = '<STR_LIT>' % i <EOL> return m <EOL> with open ( filename ) as f : <EOL> tests = list ( yaml . load_all ( f ) ) <EOL> attrs = { <EOL> '<STR_LIT>' : filename <EOL> } <EOL> i = <NUM_LIT:0> <EOL> for test in tests : <EOL> for test_name , definition in test . items ( ) : <EOL> if test_name == '<STR_LIT>' : <EOL> attrs [ '<STR_LIT>' ] = definition <EOL> continue <EOL> attrs [ '<STR_LIT>' % i ] = make_test ( test_name , definition , i ) <EOL> i += <NUM_LIT:1> <EOL> return type ( name , ( YamlTestCase , ) , attrs ) <EOL> YAML_DIR = environ . get ( <EOL> '<STR_LIT>' , <EOL> join ( <EOL> dirname ( __file__ ) , pardir , pardir , pardir , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:src>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:test>' <EOL> ) <EOL> ) <EOL> if exists ( YAML_DIR ) : <EOL> for ( path , dirs , files ) in walk ( YAML_DIR ) : <EOL> for filename in files : <EOL> if not filename . endswith ( ( '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> continue <EOL> name = ( '<STR_LIT>' + '<STR_LIT>' . join ( s . title ( ) for s in path [ len ( YAML_DIR ) + <NUM_LIT:1> : ] . split ( '<STR_LIT:/>' ) ) + filename . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] . title ( ) ) . replace ( '<STR_LIT:_>' , '<STR_LIT>' ) . replace ( '<STR_LIT:.>' , '<STR_LIT>' ) <EOL> locals ( ) [ name ] = construct_case ( join ( path , filename ) , name ) </s>
