<s> __author__ = '<STR_LIT>' <EOL> import logging <EOL> import os <EOL> import random <EOL> import sys <EOL> if ( sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> , <NUM_LIT:0> ) ) : <EOL> import unittest2 as unittest <EOL> else : <EOL> import unittest <EOL> try : <EOL> import ConfigParser as cp <EOL> except ImportError : <EOL> import configparser as cp <EOL> import shutil <EOL> import getopt <EOL> import tempfile <EOL> import time <EOL> try : <EOL> import zmq <EOL> except ImportError : <EOL> zmq = None <EOL> import pypet . pypetconstants as pypetconstants <EOL> import pypet . compat as compat <EOL> from pypet import HasLogger <EOL> from pypet . pypetlogging import LoggingManager , rename_log_file <EOL> from pypet . utils . decorators import copydoc <EOL> from pypet . utils . helpful_functions import port_to_tcp <EOL> testParams = dict ( <EOL> tempdir = '<STR_LIT>' , <EOL> remove = True , <EOL> actual_tempdir = '<STR_LIT>' , <EOL> user_tempdir = '<STR_LIT>' , <EOL> log_config = '<STR_LIT:test>' <EOL> ) <EOL> TEST_IMPORT_ERROR = '<STR_LIT>' <EOL> generic_log_folder = None <EOL> def errwrite ( text ) : <EOL> """<STR_LIT>""" <EOL> sys . __stderr__ . write ( text + '<STR_LIT:\n>' ) <EOL> def get_root_logger ( ) : <EOL> """<STR_LIT>""" <EOL> return logging . getLogger ( ) <EOL> def get_log_config ( ) : <EOL> """<STR_LIT>""" <EOL> return testParams [ '<STR_LIT>' ] <EOL> def get_log_path ( traj , process_name = None ) : <EOL> """<STR_LIT>""" <EOL> return rename_log_file ( generic_log_folder , trajectory = traj , process_name = process_name ) <EOL> def get_random_port_url ( ) : <EOL> """<STR_LIT>""" <EOL> url = port_to_tcp ( ) <EOL> errwrite ( '<STR_LIT>' % url ) <EOL> return url <EOL> def prepare_log_config ( ) : <EOL> """<STR_LIT>""" <EOL> conf = testParams [ '<STR_LIT>' ] <EOL> pypet_path = os . path . abspath ( os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' ) ) <EOL> init_path = os . path . join ( pypet_path , '<STR_LIT>' ) <EOL> if conf == '<STR_LIT:test>' : <EOL> conf_file = os . path . join ( init_path , '<STR_LIT>' ) <EOL> conf_parser = handle_config_file ( conf_file ) <EOL> conf = conf_parser <EOL> elif conf == '<STR_LIT>' : <EOL> conf_file = os . path . join ( init_path , '<STR_LIT>' ) <EOL> conf_parser = handle_config_file ( conf_file ) <EOL> conf = conf_parser <EOL> testParams [ '<STR_LIT>' ] = conf <EOL> def _rename_filename ( filename ) : <EOL> """<STR_LIT>""" <EOL> global generic_log_folder <EOL> if not filename . startswith ( '<STR_LIT>' ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> temp_folder = make_temp_dir ( '<STR_LIT>' ) <EOL> filename = filename . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> filename = os . path . join ( temp_folder , filename ) <EOL> filename = os . path . normpath ( filename ) <EOL> if generic_log_folder is None : <EOL> generic_log_folder = os . path . dirname ( filename ) <EOL> return filename <EOL> def handle_config_file ( config_file ) : <EOL> """<STR_LIT>""" <EOL> parser = cp . ConfigParser ( ) <EOL> parser . read ( config_file ) <EOL> sections = parser . sections ( ) <EOL> for section in sections : <EOL> options = parser . options ( section ) <EOL> for option in options : <EOL> arg = parser . get ( section , option , raw = True ) <EOL> if '<STR_LIT>' in arg : <EOL> LoggingManager . _check_and_replace_parser_args ( parser , section , option , <EOL> rename_func = _rename_filename , <EOL> make_dirs = False ) <EOL> return parser <EOL> def make_temp_dir ( filename , signal = False ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if ( ( testParams [ '<STR_LIT>' ] is not None and testParams [ '<STR_LIT>' ] != '<STR_LIT>' ) and <EOL> testParams [ '<STR_LIT>' ] == '<STR_LIT>' ) : <EOL> testParams [ '<STR_LIT>' ] = testParams [ '<STR_LIT>' ] <EOL> if not os . path . isdir ( testParams [ '<STR_LIT>' ] ) : <EOL> os . makedirs ( testParams [ '<STR_LIT>' ] ) <EOL> return os . path . join ( testParams [ '<STR_LIT>' ] , filename ) <EOL> except OSError as exc : <EOL> actual_tempdir = os . path . join ( tempfile . gettempdir ( ) , testParams [ '<STR_LIT>' ] ) <EOL> if signal : <EOL> errwrite ( '<STR_LIT>' <EOL> '<STR_LIT>' % actual_tempdir ) <EOL> testParams [ '<STR_LIT>' ] = actual_tempdir <EOL> if not os . path . isdir ( testParams [ '<STR_LIT>' ] ) : <EOL> try : <EOL> os . makedirs ( testParams [ '<STR_LIT>' ] ) <EOL> except Exception : <EOL> pass <EOL> return os . path . join ( actual_tempdir , filename ) <EOL> except : <EOL> get_root_logger ( ) . error ( '<STR_LIT>' ) <EOL> raise <EOL> def run_suite ( remove = None , folder = None , suite = None ) : <EOL> """<STR_LIT>""" <EOL> if remove is not None : <EOL> testParams [ '<STR_LIT>' ] = remove <EOL> testParams [ '<STR_LIT>' ] = folder <EOL> prepare_log_config ( ) <EOL> make_temp_dir ( '<STR_LIT>' , signal = True ) <EOL> success = False <EOL> try : <EOL> if suite is None : <EOL> unittest . main ( verbosity = <NUM_LIT:2> ) <EOL> else : <EOL> runner = unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) <EOL> result = runner . run ( suite ) <EOL> success = result . wasSuccessful ( ) <EOL> finally : <EOL> remove_data ( ) <EOL> if not success : <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> def combined_suites ( * test_suites ) : <EOL> """<STR_LIT>""" <EOL> combined_suite = unittest . TestSuite ( test_suites ) <EOL> return combined_suite <EOL> def combine_tests ( * tests ) : <EOL> """<STR_LIT>""" <EOL> loader = unittest . TestLoader ( ) <EOL> suites_list = [ ] <EOL> for test in tests : <EOL> if test is not None : <EOL> suite = loader . loadTestsFromTestCase ( test ) <EOL> suites_list . append ( suite ) <EOL> return combined_suites ( * suites_list ) <EOL> def remove_data ( ) : <EOL> """<STR_LIT>""" <EOL> global testParams <EOL> if testParams [ '<STR_LIT>' ] : <EOL> get_root_logger ( ) . log ( <NUM_LIT> , '<STR_LIT>' ) <EOL> shutil . rmtree ( testParams [ '<STR_LIT>' ] , True ) <EOL> def make_trajectory_name ( testcase ) : <EOL> """<STR_LIT>""" <EOL> testid = testcase . id ( ) <EOL> split_names = testid . split ( '<STR_LIT:.>' ) <EOL> seed = len ( testid ) + int ( <NUM_LIT:10> * time . time ( ) ) <EOL> random . seed ( seed ) <EOL> randintstr = str ( random . randint ( <NUM_LIT:0> , <NUM_LIT:10> ** <NUM_LIT:5> ) ) <EOL> name = '<STR_LIT>' + split_names [ - <NUM_LIT:1> ] + '<STR_LIT>' + randintstr <EOL> maxlen = pypetconstants . HDF5_STRCOL_MAX_NAME_LENGTH - <NUM_LIT> <EOL> if len ( name ) > maxlen : <EOL> name = name [ len ( name ) - maxlen : ] <EOL> while name . startswith ( '<STR_LIT:_>' ) : <EOL> name = name [ <NUM_LIT:1> : ] <EOL> return name <EOL> class LambdaTestDiscoverer ( unittest . TestLoader , HasLogger ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , predicate = None ) : <EOL> super ( LambdaTestDiscoverer , self ) . __init__ ( ) <EOL> if predicate is not None : <EOL> self . predicate = predicate <EOL> else : <EOL> self . predicate = lambda x , y , z : True <EOL> self . _set_logger ( ) <EOL> @ staticmethod <EOL> def _input2set ( element ) : <EOL> if isinstance ( element , compat . base_type ) : <EOL> return set ( [ element ] ) <EOL> elif element is None : <EOL> return set ( ) <EOL> else : <EOL> return set ( element ) <EOL> @ staticmethod <EOL> def _flatten_suite ( suite ) : <EOL> res = [ ] <EOL> for case in suite : <EOL> if isinstance ( case , unittest . TestSuite ) : <EOL> res . extend ( LambdaTestDiscoverer . _flatten_suite ( case ) ) <EOL> else : <EOL> res . append ( case ) <EOL> return res <EOL> @ copydoc ( unittest . TestLoader . discover ) <EOL> def discover ( self , start_dir , pattern = '<STR_LIT>' , top_level_dir = None ) : <EOL> tmp_suite = super ( LambdaTestDiscoverer , self ) . discover ( start_dir = start_dir , pattern = pattern , <EOL> top_level_dir = top_level_dir ) <EOL> res_suite = unittest . TestSuite ( ) <EOL> flattened_suite = self . _flatten_suite ( tmp_suite ) <EOL> found_set = set ( ) <EOL> test_list = [ ] <EOL> for case in flattened_suite : <EOL> test_name = str ( case ) . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:0> ] <EOL> class_name = case . __class__ . __name__ <EOL> if not hasattr ( case , '<STR_LIT>' ) : <EOL> tags = set ( ) <EOL> else : <EOL> tags = self . _input2set ( case . tags ) <EOL> combined = ( class_name , test_name , tuple ( tags ) ) <EOL> if combined in found_set : <EOL> continue <EOL> else : <EOL> found_set . add ( combined ) <EOL> if class_name == '<STR_LIT>' : <EOL> self . _logger . error ( '<STR_LIT>' % test_name ) <EOL> if class_name == '<STR_LIT>' : <EOL> self . _logger . error ( '<STR_LIT>' <EOL> '<STR_LIT>' % test_name ) <EOL> add = self . predicate ( class_name , test_name , tags ) <EOL> if add : <EOL> test_list . append ( case ) <EOL> combined_name_key = lambda key : key . __class__ . __name__ + str ( key ) . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:0> ] <EOL> test_list = sorted ( test_list , key = combined_name_key ) <EOL> res_suite . addTests ( test_list ) <EOL> return res_suite <EOL> def discover_tests ( predicate = None ) : <EOL> """<STR_LIT>""" <EOL> loader = LambdaTestDiscoverer ( predicate ) <EOL> start_dir = os . path . dirname ( os . path . abspath ( __file__ ) ) <EOL> start_dir = os . path . abspath ( os . path . join ( start_dir , '<STR_LIT:..>' ) ) <EOL> suite = loader . discover ( start_dir = start_dir , pattern = '<STR_LIT>' ) <EOL> return suite <EOL> def parse_args ( ) : <EOL> """<STR_LIT>""" <EOL> opt_list , _ = getopt . getopt ( sys . argv [ <NUM_LIT:1> : ] , '<STR_LIT:k>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> opt_dict = { } <EOL> for opt , arg in opt_list : <EOL> if opt == '<STR_LIT>' : <EOL> opt_dict [ '<STR_LIT>' ] = False <EOL> errwrite ( '<STR_LIT>' ) <EOL> if opt == '<STR_LIT>' : <EOL> opt_dict [ '<STR_LIT>' ] = arg <EOL> errwrite ( '<STR_LIT>' % arg ) <EOL> if opt == '<STR_LIT>' : <EOL> opt_dict [ '<STR_LIT>' ] = arg <EOL> errwrite ( '<STR_LIT>' % arg ) <EOL> sys . argv = [ sys . argv [ <NUM_LIT:0> ] ] <EOL> return opt_dict <EOL> prepare_log_config ( ) </s>
