<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import errno <EOL> import re <EOL> import ConfigParser <EOL> import argparse <EOL> import json <EOL> import psycopg2 <EOL> import dbutil <EOL> from EOtools . execute import execute <EOL> DEFAULT_DATACUBE_CONFIG_TEMPLATE = os . path . join ( dbutil . TEST_RESOURCES_ROOT , <EOL> '<STR_LIT>' ) <EOL> DEFAULT_DATACUBE_CONFIG_NAME = '<STR_LIT>' <EOL> class Systest ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , test_name , config ) : <EOL> self . test_name = test_name <EOL> self . result = '<STR_LIT>' <EOL> self . error_message = None <EOL> try : <EOL> self . test_dir = self . create_dir ( '<STR_LIT:.>' , test_name ) <EOL> self . scenes_dir = self . create_dir ( self . test_dir , '<STR_LIT>' ) <EOL> self . remove_old_links ( ) <EOL> if config . has_option ( test_name , '<STR_LIT>' ) : <EOL> scene_list = self . link_scenes ( test_name , config ) <EOL> else : <EOL> scene_list = None <EOL> for iscene in range ( len ( scene_list ) ) : <EOL> os . environ [ '<STR_LIT>' % iscene ] = scene_list [ iscene ] <EOL> os . environ [ '<STR_LIT>' ] = str ( len ( scene_list ) ) <EOL> os . environ [ '<STR_LIT>' ] = self . test_dir <EOL> self . temp_dir = self . create_dir ( self . test_dir , '<STR_LIT>' ) <EOL> self . tile_dir = self . create_dir ( self . test_dir , '<STR_LIT>' ) <EOL> self . dbname = dbutil . random_name ( test_name ) <EOL> self . load_initial_database ( test_name , config ) <EOL> self . make_datacube_config ( test_name , config ) <EOL> self . command = None <EOL> if config . has_option ( test_name , '<STR_LIT>' ) : <EOL> self . command = config . get ( test_name , '<STR_LIT>' ) <EOL> os . environ [ '<STR_LIT>' ] = config . get ( test_name , '<STR_LIT>' ) <EOL> self . logfile = self . open_logfile ( test_name ) <EOL> except AssertionError as e : <EOL> self . result = '<STR_LIT>' <EOL> self . error_message = e . message <EOL> @ staticmethod <EOL> def create_dir ( parent_dir , dir_name ) : <EOL> """<STR_LIT>""" <EOL> dir_path = os . path . abspath ( os . path . join ( parent_dir , dir_name ) ) <EOL> try : <EOL> os . mkdir ( dir_path ) <EOL> except OSError as e : <EOL> if e . errno != errno . EEXIST : <EOL> raise AssertionError ( '<STR_LIT>' % <EOL> ( dir_path , e . strerror ) ) <EOL> return dir_path <EOL> def remove_old_links ( self ) : <EOL> """<STR_LIT>""" <EOL> for file_name in os . listdir ( self . scenes_dir ) : <EOL> file_path = os . path . join ( self . scenes_dir , file_name ) <EOL> if os . path . islink ( file_path ) : <EOL> try : <EOL> os . remove ( file_path ) <EOL> except IOError as e : <EOL> raise AssertionError ( '<STR_LIT>' + <EOL> '<STR_LIT>' % ( file_path , e . strerror ) ) <EOL> def link_scenes ( self , test_name , config ) : <EOL> """<STR_LIT>""" <EOL> scene_list = json . loads ( config . get ( test_name , '<STR_LIT>' ) ) <EOL> for scene_path in scene_list : <EOL> if not os . path . exists ( scene_path ) : <EOL> raise AssertionError ( '<STR_LIT>' % scene_path ) <EOL> scene_path = os . path . abspath ( scene_path ) <EOL> scene_name = os . path . basename ( re . sub ( r'<STR_LIT>' , '<STR_LIT>' , scene_path ) ) <EOL> link_path = os . path . join ( self . scenes_dir , scene_name ) <EOL> try : <EOL> os . symlink ( scene_path , link_path ) <EOL> except IOError as e : <EOL> raise AssertionError ( '<STR_LIT>' % <EOL> ( scene_path , e . strerror ) ) <EOL> return scene_list <EOL> def make_datacube_config ( self , test_name , config ) : <EOL> """<STR_LIT>""" <EOL> template = config . get ( test_name , '<STR_LIT>' ) <EOL> ( template_dir , template_name ) = os . path . split ( template ) <EOL> output_name = config . get ( test_name , '<STR_LIT>' ) <EOL> try : <EOL> updates = { '<STR_LIT>' : self . dbname , <EOL> '<STR_LIT>' : self . temp_dir , <EOL> '<STR_LIT>' : self . tile_dir } <EOL> dbutil . update_config_file2 ( updates , template_dir , self . test_dir , <EOL> template_name , output_name ) <EOL> except IOError as e : <EOL> raise AssertionError ( '<STR_LIT>' % <EOL> e . strerror ) <EOL> def load_initial_database ( self , test_name , config ) : <EOL> """<STR_LIT>""" <EOL> initial_database = config . get ( test_name , '<STR_LIT>' ) <EOL> m = re . match ( r'<STR_LIT>' , initial_database ) <EOL> if m is None : <EOL> self . dbname = initial_database <EOL> return <EOL> ( save_dir , save_file ) = os . path . split ( initial_database ) <EOL> try : <EOL> dbutil . TESTSERVER . create ( self . dbname , save_dir , save_file ) <EOL> except psycopg2 . Error as e : <EOL> raise AssertionError ( '<STR_LIT>' % <EOL> e . pgerror ) <EOL> @ staticmethod <EOL> def open_logfile ( test_name ) : <EOL> """<STR_LIT>""" <EOL> logfile_path = os . path . join ( test_name , test_name + '<STR_LIT>' ) <EOL> try : <EOL> logfile = open ( logfile_path , '<STR_LIT:w>' ) <EOL> except IOError as e : <EOL> raise AssertionError ( '<STR_LIT>' % <EOL> ( logfile_path , e . strerror ) ) <EOL> return logfile <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . result : <EOL> return self . result <EOL> elif self . command : <EOL> print '<STR_LIT>' <EOL> os . chdir ( self . test_name ) <EOL> print '<STR_LIT>' , os . getcwd ( ) <EOL> print '<STR_LIT>' <EOL> print '<STR_LIT>' <EOL> print self . command <EOL> print '<STR_LIT>' <EOL> exe_result = execute ( self . command ) <EOL> self . logfile . write ( exe_result [ '<STR_LIT>' ] ) <EOL> self . logfile . write ( exe_result [ '<STR_LIT>' ] ) <EOL> if exe_result [ '<STR_LIT>' ] != <NUM_LIT:0> : <EOL> self . error_message = exe_result [ '<STR_LIT>' ] <EOL> return '<STR_LIT>' <EOL> os . chdir ( '<STR_LIT:..>' ) <EOL> return '<STR_LIT>' <EOL> else : <EOL> return '<STR_LIT>' <EOL> def parse_args ( ) : <EOL> """<STR_LIT>""" <EOL> argparser = argparse . ArgumentParser ( description = '<STR_LIT>' ) <EOL> argparser . add_argument ( '<STR_LIT>' , <EOL> help = '<STR_LIT>' ) <EOL> return argparser . parse_args ( ) <EOL> def read_config ( args ) : <EOL> """<STR_LIT>""" <EOL> script_path = os . path . abspath ( sys . argv [ <NUM_LIT:0> ] ) <EOL> script_dir = os . path . dirname ( script_path ) <EOL> datacube_root = os . environ . get ( '<STR_LIT>' , script_dir ) <EOL> defaults = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : datacube_root , <EOL> '<STR_LIT>' : DEFAULT_DATACUBE_CONFIG_TEMPLATE , <EOL> '<STR_LIT>' : DEFAULT_DATACUBE_CONFIG_NAME <EOL> } <EOL> config = ConfigParser . SafeConfigParser ( defaults ) <EOL> with open ( args . config_file ) as fp : <EOL> config . readfp ( fp ) <EOL> return config <EOL> def run_tests ( ) : <EOL> """<STR_LIT>""" <EOL> args = parse_args ( ) <EOL> config = read_config ( args ) <EOL> for test_name in config . sections ( ) : <EOL> print "<STR_LIT>" % test_name <EOL> systest = Systest ( test_name , config ) <EOL> result = systest . run ( ) <EOL> print "<STR_LIT>" % result <EOL> if result == '<STR_LIT>' : <EOL> print systest . error_message <EOL> print "<STR_LIT>" <EOL> def dump_environment ( ) : <EOL> """<STR_LIT>""" <EOL> print "<STR_LIT>" <EOL> print "<STR_LIT>" <EOL> print "<STR_LIT>" , os . environ [ '<STR_LIT>' ] <EOL> print "<STR_LIT>" <EOL> print "<STR_LIT>" , os . environ [ '<STR_LIT>' ] <EOL> print "<STR_LIT>" <EOL> print "<STR_LIT>" , os . environ [ '<STR_LIT>' ] <EOL> print "<STR_LIT>" <EOL> print "<STR_LIT>" <EOL> def dump_config ( config ) : <EOL> """<STR_LIT>""" <EOL> print "<STR_LIT>" <EOL> print "<STR_LIT>" <EOL> for section in config . sections ( ) : <EOL> print "<STR_LIT>" , section <EOL> print "<STR_LIT>" <EOL> for ( name , value ) in config . items ( section ) : <EOL> print "<STR_LIT>" % ( name , value ) <EOL> if config . has_option ( section , '<STR_LIT>' ) : <EOL> scenes = json . loads ( config . get ( section , '<STR_LIT>' ) ) <EOL> print "<STR_LIT>" , scenes <EOL> print "<STR_LIT>" <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> run_tests ( ) </s>
