<s> import os <EOL> import sys <EOL> import time <EOL> import glob <EOL> import fnmatch <EOL> import shutil <EOL> import subprocess <EOL> try : <EOL> from robot . version import get_version <EOL> from robot . api import logger <EOL> from robot . utils import ( ConnectionCache , seq2str , timestr_to_secs , <EOL> secs_to_timestr , plural_or_not , get_time , abspath , <EOL> secs_to_timestamp , parse_time , unic , decode_output , <EOL> get_env_var , set_env_var , del_env_var , get_env_vars , <EOL> decode_from_system ) <EOL> __version__ = get_version ( ) <EOL> PROCESSES = ConnectionCache ( '<STR_LIT>' ) <EOL> del ConnectionCache , get_version <EOL> except ImportError : <EOL> from os . path import abspath <EOL> from os import ( getenv as get_env_var , putenv as set_env_var , <EOL> unsetenv as del_env_var , environ ) <EOL> __version__ = '<STR_LIT>' <EOL> get_env_vars = environ . copy <EOL> logger = None <EOL> seq2str = lambda items : '<STR_LIT:U+002CU+0020>' . join ( "<STR_LIT>" % item for item in items ) <EOL> timestr_to_secs = int <EOL> plural_or_not = lambda count : '<STR_LIT>' if count == <NUM_LIT:1> else '<STR_LIT:s>' <EOL> secs_to_timestr = lambda secs : '<STR_LIT>' % ( secs , plural_or_not ( secs ) ) <EOL> unic = unicode <EOL> decode_output = decode_from_system = lambda string : string <EOL> class _NotImplemented : <EOL> def __getattr__ ( self , name ) : <EOL> raise NotImplementedError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> get_time = secs_to_timestamp = parse_time = PROCESSES = _NotImplemented ( ) <EOL> class OperatingSystem : <EOL> """<STR_LIT>""" <EOL> ROBOT_LIBRARY_SCOPE = '<STR_LIT>' <EOL> ROBOT_LIBRARY_VERSION = __version__ <EOL> def run ( self , command ) : <EOL> """<STR_LIT>""" <EOL> return self . _run ( command ) [ <NUM_LIT:1> ] <EOL> def run_and_return_rc ( self , command ) : <EOL> """<STR_LIT>""" <EOL> return self . _run ( command ) [ <NUM_LIT:0> ] <EOL> def run_and_return_rc_and_output ( self , command ) : <EOL> """<STR_LIT>""" <EOL> return self . _run ( command ) <EOL> def _run ( self , command ) : <EOL> process = _Process ( command ) <EOL> self . _info ( "<STR_LIT>" % process ) <EOL> stdout = process . read ( ) <EOL> rc = process . close ( ) <EOL> return rc , stdout <EOL> def start_process ( self , command , stdin = None , alias = None ) : <EOL> """<STR_LIT>""" <EOL> process = _Process2 ( command , stdin ) <EOL> self . _info ( "<STR_LIT>" % process ) <EOL> return PROCESSES . register ( process , alias ) <EOL> def switch_process ( self , index_or_alias ) : <EOL> """<STR_LIT>""" <EOL> PROCESSES . switch ( index_or_alias ) <EOL> def read_process_output ( self ) : <EOL> """<STR_LIT>""" <EOL> output = PROCESSES . current . read ( ) <EOL> PROCESSES . current . close ( ) <EOL> return output <EOL> def stop_process ( self ) : <EOL> """<STR_LIT>""" <EOL> PROCESSES . current . close ( ) <EOL> def stop_all_processes ( self ) : <EOL> """<STR_LIT>""" <EOL> PROCESSES . close_all ( ) <EOL> def get_file ( self , path , encoding = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> content = self . get_binary_file ( path ) <EOL> return unicode ( content , encoding ) . replace ( '<STR_LIT:\r\n>' , '<STR_LIT:\n>' ) <EOL> def get_binary_file ( self , path ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> f = open ( path , '<STR_LIT:rb>' ) <EOL> try : <EOL> return f . read ( ) <EOL> finally : <EOL> f . close ( ) <EOL> def grep_file ( self , path , pattern , encoding = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> pattern = '<STR_LIT>' % pattern <EOL> orig = self . get_file ( path , encoding ) . splitlines ( ) <EOL> lines = [ line for line in orig if fnmatch . fnmatchcase ( line , pattern ) ] <EOL> self . _info ( '<STR_LIT>' % ( len ( lines ) , len ( orig ) ) ) <EOL> return '<STR_LIT:\n>' . join ( lines ) <EOL> def log_file ( self , path , encoding = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> content = self . get_file ( path , encoding ) <EOL> self . _info ( content ) <EOL> return content <EOL> def should_exist ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> if not glob . glob ( path ) : <EOL> self . _fail ( msg , "<STR_LIT>" % path ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def should_not_exist ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> matches = glob . glob ( path ) <EOL> if not matches : <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> return <EOL> if not msg : <EOL> if self . _is_pattern_path ( path ) : <EOL> matches . sort ( ) <EOL> msg = "<STR_LIT>" % ( path , seq2str ( matches ) ) <EOL> else : <EOL> msg = "<STR_LIT>" % path <EOL> raise AssertionError ( msg ) <EOL> def file_should_exist ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> matches = [ p for p in glob . glob ( path ) if os . path . isfile ( p ) ] <EOL> if not matches : <EOL> self . _fail ( msg , "<STR_LIT>" % path ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def file_should_not_exist ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> matches = [ p for p in glob . glob ( path ) if os . path . isfile ( p ) ] <EOL> if not matches : <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> return <EOL> if not msg : <EOL> if self . _is_pattern_path ( path ) : <EOL> matches . sort ( ) <EOL> name = len ( matches ) == <NUM_LIT:1> and '<STR_LIT:file>' or '<STR_LIT>' <EOL> msg = "<STR_LIT>" % ( path , name , seq2str ( matches ) ) <EOL> else : <EOL> msg = "<STR_LIT>" % path <EOL> raise AssertionError ( msg ) <EOL> def directory_should_exist ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> matches = [ p for p in glob . glob ( path ) if os . path . isdir ( p ) ] <EOL> if not matches : <EOL> self . _fail ( msg , "<STR_LIT>" % path ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def directory_should_not_exist ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> matches = [ p for p in glob . glob ( path ) if os . path . isdir ( p ) ] <EOL> if not matches : <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> return <EOL> if not msg : <EOL> if self . _is_pattern_path ( path ) : <EOL> matches . sort ( ) <EOL> name = len ( matches ) == <NUM_LIT:1> and '<STR_LIT>' or '<STR_LIT>' <EOL> msg = "<STR_LIT>" % ( path , name , seq2str ( matches ) ) <EOL> else : <EOL> msg = "<STR_LIT>" % path <EOL> raise AssertionError ( msg ) <EOL> def _is_pattern_path ( self , path ) : <EOL> return '<STR_LIT:*>' in path or '<STR_LIT:?>' in path or ( '<STR_LIT:[>' in path and '<STR_LIT:]>' in path ) <EOL> def wait_until_removed ( self , path , timeout = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> timeout = timestr_to_secs ( timeout ) <EOL> maxtime = time . time ( ) + timeout <EOL> while glob . glob ( path ) : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> if timeout >= <NUM_LIT:0> and time . time ( ) > maxtime : <EOL> raise AssertionError ( "<STR_LIT>" <EOL> % ( path , secs_to_timestr ( timeout ) ) ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def wait_until_created ( self , path , timeout = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> timeout = timestr_to_secs ( timeout ) <EOL> maxtime = time . time ( ) + timeout <EOL> while not glob . glob ( path ) : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> if timeout >= <NUM_LIT:0> and time . time ( ) > maxtime : <EOL> raise AssertionError ( "<STR_LIT>" <EOL> % ( path , secs_to_timestr ( timeout ) ) ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def directory_should_be_empty ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> items = self . _list_dir ( path ) <EOL> if items : <EOL> if not msg : <EOL> msg = "<STR_LIT>" % ( path , seq2str ( items , lastsep = '<STR_LIT:U+002CU+0020>' ) ) <EOL> raise AssertionError ( msg ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def directory_should_not_be_empty ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> count = len ( self . _list_dir ( path ) ) <EOL> if count == <NUM_LIT:0> : <EOL> self . _fail ( msg , "<STR_LIT>" % path ) <EOL> plural = plural_or_not ( count ) <EOL> self . _link ( "<STR_LIT>" % ( count , plural ) , <EOL> path ) <EOL> def file_should_be_empty ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> if not os . path . isfile ( path ) : <EOL> raise AssertionError ( "<STR_LIT>" % path ) <EOL> size = os . stat ( path ) . st_size <EOL> if size > <NUM_LIT:0> : <EOL> self . _fail ( msg , "<STR_LIT>" % ( path , size ) ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def file_should_not_be_empty ( self , path , msg = None ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> if not os . path . isfile ( path ) : <EOL> raise AssertionError ( "<STR_LIT>" % path ) <EOL> size = os . stat ( path ) . st_size <EOL> if size == <NUM_LIT:0> : <EOL> self . _fail ( msg , "<STR_LIT>" % path ) <EOL> self . _link ( "<STR_LIT>" % size , path ) <EOL> def create_file ( self , path , content = '<STR_LIT>' , encoding = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> path = self . _write_to_file ( path , content , encoding , '<STR_LIT:w>' ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def append_to_file ( self , path , content , encoding = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> path = self . _write_to_file ( path , content , encoding , '<STR_LIT:a>' ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def _write_to_file ( self , path , content , encoding , mode ) : <EOL> path = self . _absnorm ( path ) <EOL> parent = os . path . dirname ( path ) <EOL> if not os . path . exists ( parent ) : <EOL> os . makedirs ( parent ) <EOL> f = open ( path , mode + '<STR_LIT:b>' ) <EOL> try : <EOL> f . write ( content . encode ( encoding ) ) <EOL> finally : <EOL> f . close ( ) <EOL> return path <EOL> def remove_file ( self , path ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> matches = glob . glob ( path ) <EOL> if not matches : <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> for match in matches : <EOL> if not os . path . isfile ( match ) : <EOL> raise RuntimeError ( "<STR_LIT>" % match ) <EOL> os . remove ( match ) <EOL> self . _link ( "<STR_LIT>" , match ) <EOL> def remove_files ( self , * paths ) : <EOL> """<STR_LIT>""" <EOL> for path in paths : <EOL> self . remove_file ( path ) <EOL> def empty_directory ( self , path ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> items = [ os . path . join ( path , item ) for item in self . _list_dir ( path ) ] <EOL> for item in items : <EOL> if os . path . isdir ( item ) : <EOL> shutil . rmtree ( item ) <EOL> else : <EOL> os . remove ( item ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def create_directory ( self , path ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> if os . path . isdir ( path ) : <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> return <EOL> if os . path . exists ( path ) : <EOL> raise RuntimeError ( "<STR_LIT>" % path ) <EOL> os . makedirs ( path ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def remove_directory ( self , path , recursive = False ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> if not os . path . exists ( path ) : <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> return <EOL> if os . path . isfile ( path ) : <EOL> raise RuntimeError ( "<STR_LIT>" % path ) <EOL> if recursive : <EOL> shutil . rmtree ( path ) <EOL> else : <EOL> msg = "<STR_LIT>" % path <EOL> self . directory_should_be_empty ( path , msg ) <EOL> os . rmdir ( path ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def copy_file ( self , source , destination ) : <EOL> """<STR_LIT>""" <EOL> source , destination = self . _copy_file ( source , destination ) <EOL> self . _link ( "<STR_LIT>" , source , destination ) <EOL> def move_file ( self , source , destination ) : <EOL> """<STR_LIT>""" <EOL> source , destination = self . _copy_file ( source , destination ) <EOL> os . remove ( source ) <EOL> self . _link ( "<STR_LIT>" , source , destination ) <EOL> def _copy_file ( self , source , dest ) : <EOL> source = self . _absnorm ( source ) <EOL> dest = dest . replace ( '<STR_LIT:/>' , os . sep ) <EOL> dest_is_dir = dest . endswith ( os . sep ) <EOL> dest = self . _absnorm ( dest ) <EOL> if not os . path . exists ( source ) : <EOL> raise RuntimeError ( "<STR_LIT>" % source ) <EOL> if not os . path . isfile ( source ) : <EOL> raise RuntimeError ( "<STR_LIT>" % source ) <EOL> if not os . path . exists ( dest ) : <EOL> if dest_is_dir : <EOL> parent = dest <EOL> else : <EOL> parent = os . path . dirname ( dest ) <EOL> if not os . path . exists ( parent ) : <EOL> os . makedirs ( parent ) <EOL> shutil . copy ( source , dest ) <EOL> return source , dest <EOL> def copy_directory ( self , source , destination ) : <EOL> """<STR_LIT>""" <EOL> source , destination = self . _copy_dir ( source , destination ) <EOL> self . _link ( "<STR_LIT>" , source , destination ) <EOL> def move_directory ( self , source , destination ) : <EOL> """<STR_LIT>""" <EOL> source , destination = self . _copy_dir ( source , destination ) <EOL> shutil . rmtree ( source ) <EOL> self . _link ( "<STR_LIT>" , source , destination ) <EOL> def _copy_dir ( self , source , dest ) : <EOL> source = self . _absnorm ( source ) <EOL> dest = self . _absnorm ( dest ) <EOL> if not os . path . exists ( source ) : <EOL> raise RuntimeError ( "<STR_LIT>" % source ) <EOL> if not os . path . isdir ( source ) : <EOL> raise RuntimeError ( "<STR_LIT>" % source ) <EOL> if os . path . exists ( dest ) and not os . path . isdir ( dest ) : <EOL> raise RuntimeError ( "<STR_LIT>" % dest ) <EOL> if os . path . exists ( dest ) : <EOL> base = os . path . basename ( source ) <EOL> dest = os . path . join ( dest , base ) <EOL> else : <EOL> parent = os . path . dirname ( dest ) <EOL> if not os . path . exists ( parent ) : <EOL> os . makedirs ( parent ) <EOL> shutil . copytree ( source , dest ) <EOL> return source , dest <EOL> def get_environment_variable ( self , name , default = None ) : <EOL> """<STR_LIT>""" <EOL> value = get_env_var ( name , default ) <EOL> if value is None : <EOL> raise RuntimeError ( "<STR_LIT>" % name ) <EOL> return value <EOL> def set_environment_variable ( self , name , value ) : <EOL> """<STR_LIT>""" <EOL> set_env_var ( name , value ) <EOL> self . _info ( "<STR_LIT>" % ( name , value ) ) <EOL> def remove_environment_variable ( self , * names ) : <EOL> """<STR_LIT>""" <EOL> for name in names : <EOL> value = del_env_var ( name ) <EOL> if value : <EOL> self . _info ( "<STR_LIT>" % name ) <EOL> else : <EOL> self . _info ( "<STR_LIT>" % name ) <EOL> def environment_variable_should_be_set ( self , name , msg = None ) : <EOL> """<STR_LIT>""" <EOL> value = get_env_var ( name ) <EOL> if not value : <EOL> self . _fail ( msg , "<STR_LIT>" % name ) <EOL> self . _info ( "<STR_LIT>" % ( name , value ) ) <EOL> def environment_variable_should_not_be_set ( self , name , msg = None ) : <EOL> """<STR_LIT>""" <EOL> value = get_env_var ( name ) <EOL> if value : <EOL> self . _fail ( msg , "<STR_LIT>" % ( name , value ) ) <EOL> self . _info ( "<STR_LIT>" % name ) <EOL> def get_environment_variables ( self ) : <EOL> """<STR_LIT>""" <EOL> return get_env_vars ( ) <EOL> def log_environment_variables ( self , level = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> vars = get_env_vars ( ) <EOL> for name , value in sorted ( vars . items ( ) , key = lambda item : item [ <NUM_LIT:0> ] . lower ( ) ) : <EOL> self . _log ( '<STR_LIT>' % ( name , value ) , level ) <EOL> return vars <EOL> def join_path ( self , base , * parts ) : <EOL> """<STR_LIT>""" <EOL> base = base . replace ( '<STR_LIT:/>' , os . sep ) <EOL> parts = [ p . replace ( '<STR_LIT:/>' , os . sep ) for p in parts ] <EOL> return self . normalize_path ( os . path . join ( base , * parts ) ) <EOL> def join_paths ( self , base , * paths ) : <EOL> """<STR_LIT>""" <EOL> return [ self . join_path ( base , path ) for path in paths ] <EOL> def normalize_path ( self , path ) : <EOL> """<STR_LIT>""" <EOL> return os . path . normpath ( path . replace ( '<STR_LIT:/>' , os . sep ) ) or '<STR_LIT:.>' <EOL> def split_path ( self , path ) : <EOL> """<STR_LIT>""" <EOL> return os . path . split ( self . normalize_path ( path ) ) <EOL> def split_extension ( self , path ) : <EOL> """<STR_LIT>""" <EOL> path = self . normalize_path ( path ) <EOL> basename = os . path . basename ( path ) <EOL> if basename . startswith ( '<STR_LIT:.>' * basename . count ( '<STR_LIT:.>' ) ) : <EOL> return path , '<STR_LIT>' <EOL> if path . endswith ( '<STR_LIT:.>' ) : <EOL> path2 = path . rstrip ( '<STR_LIT:.>' ) <EOL> trailing_dots = '<STR_LIT:.>' * ( len ( path ) - len ( path2 ) ) <EOL> path = path2 <EOL> else : <EOL> trailing_dots = '<STR_LIT>' <EOL> basepath , ext = os . path . splitext ( path ) <EOL> if ext . startswith ( '<STR_LIT:.>' ) : <EOL> ext = ext [ <NUM_LIT:1> : ] <EOL> if ext : <EOL> ext += trailing_dots <EOL> else : <EOL> basepath += trailing_dots <EOL> return basepath , ext <EOL> def get_modified_time ( self , path , format = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> if not os . path . exists ( path ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" % path ) <EOL> mtime = get_time ( format , os . stat ( path ) . st_mtime ) <EOL> self . _link ( "<STR_LIT>" % mtime , path ) <EOL> return mtime <EOL> def set_modified_time ( self , path , mtime ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> try : <EOL> if not os . path . exists ( path ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if not os . path . isfile ( path ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> mtime = parse_time ( mtime ) <EOL> except ValueError , err : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> % ( path , unicode ( err ) ) ) <EOL> os . utime ( path , ( mtime , mtime ) ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> tstamp = secs_to_timestamp ( mtime , ( '<STR_LIT:->' , '<STR_LIT:U+0020>' , '<STR_LIT::>' ) ) <EOL> self . _link ( "<STR_LIT>" % tstamp , path ) <EOL> def get_file_size ( self , path ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> if not os . path . isfile ( path ) : <EOL> raise RuntimeError ( "<STR_LIT>" % path ) <EOL> size = os . stat ( path ) . st_size <EOL> plural = plural_or_not ( size ) <EOL> self . _link ( "<STR_LIT>" % ( size , plural ) , path ) <EOL> return size <EOL> def list_directory ( self , path , pattern = None , absolute = False ) : <EOL> """<STR_LIT>""" <EOL> items = self . _list_dir ( path , pattern , absolute ) <EOL> self . _info ( '<STR_LIT>' % ( len ( items ) , plural_or_not ( items ) , '<STR_LIT:\n>' . join ( items ) ) ) <EOL> return items <EOL> def list_files_in_directory ( self , path , pattern = None , absolute = False ) : <EOL> """<STR_LIT>""" <EOL> files = self . _list_files_in_dir ( path , pattern , absolute ) <EOL> self . _info ( '<STR_LIT>' % ( len ( files ) , plural_or_not ( files ) , '<STR_LIT:\n>' . join ( files ) ) ) <EOL> return files <EOL> def list_directories_in_directory ( self , path , pattern = None , absolute = False ) : <EOL> """<STR_LIT>""" <EOL> dirs = self . _list_dirs_in_dir ( path , pattern , absolute ) <EOL> self . _info ( '<STR_LIT>' % ( len ( dirs ) , '<STR_LIT:y>' if len ( dirs ) == <NUM_LIT:1> else '<STR_LIT>' , '<STR_LIT:\n>' . join ( dirs ) ) ) <EOL> return dirs <EOL> def count_items_in_directory ( self , path , pattern = None ) : <EOL> """<STR_LIT>""" <EOL> count = len ( self . _list_dir ( path , pattern ) ) <EOL> self . _info ( "<STR_LIT>" % ( count , plural_or_not ( count ) ) ) <EOL> return count <EOL> def count_files_in_directory ( self , path , pattern = None ) : <EOL> """<STR_LIT>""" <EOL> count = len ( self . _list_files_in_dir ( path , pattern ) ) <EOL> self . _info ( "<STR_LIT>" % ( count , plural_or_not ( count ) ) ) <EOL> return count <EOL> def count_directories_in_directory ( self , path , pattern = None ) : <EOL> """<STR_LIT>""" <EOL> count = len ( self . _list_dirs_in_dir ( path , pattern ) ) <EOL> self . _info ( "<STR_LIT>" % ( count , '<STR_LIT:y>' if count == <NUM_LIT:1> else '<STR_LIT>' ) ) <EOL> return count <EOL> def _list_dir ( self , path , pattern = None , absolute = False ) : <EOL> path = self . _absnorm ( path ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> if not os . path . isdir ( path ) : <EOL> raise RuntimeError ( "<STR_LIT>" % path ) <EOL> items = sorted ( unic ( item ) for item in os . listdir ( path ) ) <EOL> if pattern : <EOL> items = [ i for i in items if fnmatch . fnmatchcase ( i , pattern ) ] <EOL> if absolute : <EOL> path = os . path . normpath ( path ) <EOL> items = [ os . path . join ( path , item ) for item in items ] <EOL> return items <EOL> def _list_files_in_dir ( self , path , pattern = None , absolute = False ) : <EOL> return [ item for item in self . _list_dir ( path , pattern , absolute ) <EOL> if os . path . isfile ( os . path . join ( path , item ) ) ] <EOL> def _list_dirs_in_dir ( self , path , pattern = None , absolute = False ) : <EOL> return [ item for item in self . _list_dir ( path , pattern , absolute ) <EOL> if os . path . isdir ( os . path . join ( path , item ) ) ] <EOL> def touch ( self , path ) : <EOL> """<STR_LIT>""" <EOL> path = self . _absnorm ( path ) <EOL> if os . path . isdir ( path ) : <EOL> raise RuntimeError ( "<STR_LIT>" % path ) <EOL> if not os . path . exists ( os . path . dirname ( path ) ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" % path ) <EOL> if os . path . exists ( path ) : <EOL> mtime = round ( time . time ( ) ) <EOL> os . utime ( path , ( mtime , mtime ) ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> else : <EOL> open ( path , '<STR_LIT:w>' ) . close ( ) <EOL> self . _link ( "<STR_LIT>" , path ) <EOL> def _absnorm ( self , path ) : <EOL> try : <EOL> return abspath ( path . replace ( '<STR_LIT:/>' , os . sep ) ) <EOL> except ValueError : <EOL> return os . path . normpath ( path . replace ( '<STR_LIT:/>' , os . sep ) ) <EOL> def _fail ( self , error , default ) : <EOL> raise AssertionError ( error or default ) <EOL> def _info ( self , msg ) : <EOL> self . _log ( msg , '<STR_LIT>' ) <EOL> def _link ( self , msg , * paths ) : <EOL> paths = tuple ( '<STR_LIT>' % ( p , p ) for p in paths ) <EOL> self . _log ( msg % paths , '<STR_LIT>' ) <EOL> def _warn ( self , msg ) : <EOL> self . _log ( msg , '<STR_LIT>' ) <EOL> def _log ( self , msg , level ) : <EOL> if logger : <EOL> logger . write ( msg , level ) <EOL> else : <EOL> print '<STR_LIT>' % ( level , msg ) <EOL> class _Process : <EOL> def __init__ ( self , command ) : <EOL> self . _command = self . _process_command ( command ) <EOL> self . _process = os . popen ( self . _command ) <EOL> def __str__ ( self ) : <EOL> return self . _command <EOL> def read ( self ) : <EOL> return self . _process_output ( self . _process . read ( ) ) <EOL> def close ( self ) : <EOL> try : <EOL> rc = self . _process . close ( ) <EOL> except IOError : <EOL> return <NUM_LIT:255> <EOL> if rc is None : <EOL> return <NUM_LIT:0> <EOL> if os . sep == '<STR_LIT:\\>' or sys . platform . startswith ( '<STR_LIT>' ) : <EOL> return rc % <NUM_LIT> <EOL> return rc >> <NUM_LIT:8> <EOL> def _process_command ( self , command ) : <EOL> if '<STR_LIT:>>' not in command : <EOL> if command . endswith ( '<STR_LIT:&>' ) : <EOL> command = command [ : - <NUM_LIT:1> ] + '<STR_LIT>' <EOL> else : <EOL> command += '<STR_LIT>' <EOL> return self . _encode_to_file_system ( command ) <EOL> def _encode_to_file_system ( self , string ) : <EOL> enc = sys . getfilesystemencoding ( ) <EOL> return string . encode ( enc ) if enc else string <EOL> def _process_output ( self , stdout ) : <EOL> stdout = stdout . replace ( '<STR_LIT:\r\n>' , '<STR_LIT:\n>' ) <EOL> if stdout . endswith ( '<STR_LIT:\n>' ) : <EOL> stdout = stdout [ : - <NUM_LIT:1> ] <EOL> return decode_output ( stdout ) <EOL> class _Process2 ( _Process ) : <EOL> def __init__ ( self , command , input_ ) : <EOL> self . _command = self . _process_command ( command ) <EOL> p = subprocess . Popen ( self . _command , shell = True , stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , stderr = subprocess . STDOUT , <EOL> close_fds = os . sep == '<STR_LIT:/>' ) <EOL> stdin , self . stdout = p . stdin , p . stdout <EOL> if input_ : <EOL> stdin . write ( input_ ) <EOL> stdin . close ( ) <EOL> self . closed = False <EOL> def read ( self ) : <EOL> if self . closed : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> return self . _process_output ( self . stdout . read ( ) ) <EOL> def close ( self ) : <EOL> if not self . closed : <EOL> self . stdout . close ( ) <EOL> self . closed = True </s>
