<s> '''<STR_LIT>''' <EOL> try : <EOL> import pty <EOL> import termios <EOL> import tty <EOL> import resource <EOL> import fcntl <EOL> except ImportError : <EOL> pty = None <EOL> import threading <EOL> import subprocess <EOL> try : <EOL> import Queue <EOL> except ImportError : <EOL> import queue as Queue <EOL> import ctypes <EOL> import os <EOL> import sys <EOL> import time <EOL> import re <EOL> import struct <EOL> import types <EOL> import errno <EOL> import traceback <EOL> import signal <EOL> import codecs <EOL> import stat <EOL> import select <EOL> __version__ = '<STR_LIT>' <EOL> __revision__ = '<STR_LIT>' <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> PY3 = ( sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> ) <EOL> class ExceptionPexpect ( Exception ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , value ) : <EOL> super ( ExceptionPexpect , self ) . __init__ ( value ) <EOL> self . value = value <EOL> def __str__ ( self ) : <EOL> return str ( self . value ) <EOL> def get_trace ( self ) : <EOL> '''<STR_LIT>''' <EOL> tblist = traceback . extract_tb ( sys . exc_info ( ) [ <NUM_LIT:2> ] ) <EOL> tblist = [ item for item in tblist if ( '<STR_LIT>' not in item [ <NUM_LIT:0> ] ) <EOL> and ( '<STR_LIT>' not in item [ <NUM_LIT:0> ] ) ] <EOL> tblist = traceback . format_list ( tblist ) <EOL> return '<STR_LIT>' . join ( tblist ) <EOL> class EOF ( ExceptionPexpect ) : <EOL> '''<STR_LIT>''' <EOL> class TIMEOUT ( ExceptionPexpect ) : <EOL> '''<STR_LIT>''' <EOL> class Expecter ( object ) : <EOL> def __init__ ( self , spawn , searcher , searchwindowsize = - <NUM_LIT:1> ) : <EOL> self . spawn = spawn <EOL> self . searcher = searcher <EOL> if searchwindowsize == - <NUM_LIT:1> : <EOL> searchwindowsize = spawn . searchwindowsize <EOL> self . searchwindowsize = searchwindowsize <EOL> def new_data ( self , data ) : <EOL> spawn = self . spawn <EOL> searcher = self . searcher <EOL> incoming = spawn . buffer + data <EOL> freshlen = len ( data ) <EOL> index = searcher . search ( incoming , freshlen , self . searchwindowsize ) <EOL> if index >= <NUM_LIT:0> : <EOL> spawn . buffer = incoming [ searcher . end : ] <EOL> spawn . before = incoming [ : searcher . start ] <EOL> spawn . after = incoming [ searcher . start : searcher . end ] <EOL> spawn . match = searcher . match <EOL> spawn . match_index = index <EOL> return index <EOL> spawn . buffer = incoming <EOL> def eof ( self , err = None ) : <EOL> spawn = self . spawn <EOL> spawn . before = spawn . buffer <EOL> spawn . buffer = spawn . string_type ( ) <EOL> spawn . after = EOF <EOL> index = self . searcher . eof_index <EOL> if index >= <NUM_LIT:0> : <EOL> spawn . match = EOF <EOL> spawn . match_index = index <EOL> return index <EOL> else : <EOL> spawn . match = None <EOL> spawn . match_index = None <EOL> msg = str ( spawn ) <EOL> if err is not None : <EOL> msg = str ( err ) + '<STR_LIT:\n>' + msg <EOL> raise EOF ( msg ) <EOL> def timeout ( self , err = None ) : <EOL> spawn = self . spawn <EOL> spawn . before = spawn . buffer <EOL> spawn . after = TIMEOUT <EOL> index = self . searcher . timeout_index <EOL> if index >= <NUM_LIT:0> : <EOL> spawn . match = TIMEOUT <EOL> spawn . match_index = index <EOL> return index <EOL> else : <EOL> spawn . match = None <EOL> spawn . match_index = None <EOL> msg = str ( spawn ) <EOL> if err is not None : <EOL> msg = str ( err ) + '<STR_LIT:\n>' + msg <EOL> raise TIMEOUT ( msg ) <EOL> def errored ( self ) : <EOL> spawn = self . spawn <EOL> spawn . before = spawn . buffer <EOL> spawn . after = None <EOL> spawn . match = None <EOL> spawn . match_index = None <EOL> def expect_loop ( self , timeout = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> spawn = self . spawn <EOL> if timeout is not None : <EOL> end_time = time . time ( ) + timeout <EOL> try : <EOL> incoming = spawn . buffer <EOL> spawn . buffer = spawn . string_type ( ) <EOL> while True : <EOL> idx = self . new_data ( incoming ) <EOL> if idx is not None : <EOL> return idx <EOL> if ( timeout is not None ) and ( timeout < <NUM_LIT:0> ) : <EOL> return self . timeout ( ) <EOL> incoming = spawn . read_nonblocking ( spawn . maxread , timeout ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> if timeout is not None : <EOL> timeout = end_time - time . time ( ) <EOL> except EOF as e : <EOL> return self . eof ( e ) <EOL> except TIMEOUT as e : <EOL> return self . timeout ( e ) <EOL> except : <EOL> self . errored ( ) <EOL> raise <EOL> def run ( command , timeout = - <NUM_LIT:1> , withexitstatus = False , events = None , <EOL> extra_args = None , logfile = None , cwd = None , env = None ) : <EOL> '''<STR_LIT>''' <EOL> return _run ( command , timeout = timeout , withexitstatus = withexitstatus , <EOL> events = events , extra_args = extra_args , logfile = logfile , cwd = cwd , <EOL> env = env , _spawn = spawn ) <EOL> def runu ( command , timeout = - <NUM_LIT:1> , withexitstatus = False , events = None , <EOL> extra_args = None , logfile = None , cwd = None , env = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return _run ( command , timeout = timeout , withexitstatus = withexitstatus , <EOL> events = events , extra_args = extra_args , logfile = logfile , cwd = cwd , <EOL> env = env , _spawn = spawnu , ** kwargs ) <EOL> def _run ( command , timeout , withexitstatus , events , extra_args , logfile , cwd , <EOL> env , _spawn , ** kwargs ) : <EOL> if timeout == - <NUM_LIT:1> : <EOL> child = _spawn ( command , maxread = <NUM_LIT> , logfile = logfile , cwd = cwd , env = env , <EOL> ** kwargs ) <EOL> else : <EOL> child = _spawn ( command , timeout = timeout , maxread = <NUM_LIT> , logfile = logfile , <EOL> cwd = cwd , env = env , ** kwargs ) <EOL> if isinstance ( events , list ) : <EOL> patterns = [ x for x , y in events ] <EOL> responses = [ y for x , y in events ] <EOL> elif isinstance ( events , dict ) : <EOL> patterns = list ( events . keys ( ) ) <EOL> responses = list ( events . values ( ) ) <EOL> else : <EOL> patterns = None <EOL> responses = None <EOL> child_result_list = [ ] <EOL> event_count = <NUM_LIT:0> <EOL> while True : <EOL> try : <EOL> index = child . expect ( patterns ) <EOL> if isinstance ( child . after , child . allowed_string_types ) : <EOL> child_result_list . append ( child . before + child . after ) <EOL> else : <EOL> child_result_list . append ( child . before ) <EOL> if isinstance ( responses [ index ] , child . allowed_string_types ) : <EOL> child . send ( responses [ index ] ) <EOL> elif isinstance ( responses [ index ] , types . FunctionType ) : <EOL> callback_result = responses [ index ] ( locals ( ) ) <EOL> sys . stdout . flush ( ) <EOL> if isinstance ( callback_result , child . allowed_string_types ) : <EOL> child . send ( callback_result ) <EOL> elif callback_result : <EOL> break <EOL> else : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> event_count = event_count + <NUM_LIT:1> <EOL> except TIMEOUT : <EOL> child_result_list . append ( child . before ) <EOL> break <EOL> except EOF : <EOL> child_result_list . append ( child . before ) <EOL> break <EOL> child_result = child . string_type ( ) . join ( child_result_list ) <EOL> if withexitstatus : <EOL> child . close ( ) <EOL> return ( child_result , child . exitstatus ) <EOL> else : <EOL> return child_result <EOL> class spawn ( object ) : <EOL> '''<STR_LIT>''' <EOL> string_type = bytes <EOL> if PY3 : <EOL> allowed_string_types = ( bytes , str ) <EOL> @ staticmethod <EOL> def _chr ( c ) : <EOL> return bytes ( [ c ] ) <EOL> linesep = os . linesep . encode ( '<STR_LIT:ascii>' ) <EOL> crlf = '<STR_LIT:\r\n>' . encode ( '<STR_LIT:ascii>' ) <EOL> lf = '<STR_LIT:\n>' . encode ( '<STR_LIT:ascii>' ) <EOL> @ staticmethod <EOL> def write_to_stdout ( b ) : <EOL> try : <EOL> return sys . stdout . buffer . write ( b ) <EOL> except AttributeError : <EOL> return sys . stdout . write ( b . decode ( '<STR_LIT:ascii>' , '<STR_LIT:replace>' ) ) <EOL> else : <EOL> allowed_string_types = ( basestring , ) <EOL> _chr = staticmethod ( chr ) <EOL> linesep = os . linesep <EOL> crlf = '<STR_LIT:\r\n>' <EOL> lf = '<STR_LIT:\n>' <EOL> write_to_stdout = sys . stdout . write <EOL> encoding = None <EOL> def __init__ ( self , command , args = [ ] , timeout = <NUM_LIT:30> , maxread = <NUM_LIT> , <EOL> searchwindowsize = None , logfile = None , cwd = None , env = None , <EOL> ignore_sighup = True , echo = True ) : <EOL> '''<STR_LIT>''' <EOL> if pty : <EOL> self . STDIN_FILENO = pty . STDIN_FILENO <EOL> self . STDOUT_FILENO = pty . STDOUT_FILENO <EOL> self . STDERR_FILENO = pty . STDERR_FILENO <EOL> self . stdin = sys . stdin <EOL> self . stdout = sys . stdout <EOL> self . stderr = sys . stderr <EOL> self . searcher = None <EOL> self . ignorecase = False <EOL> self . before = None <EOL> self . after = None <EOL> self . match = None <EOL> self . match_index = None <EOL> self . terminated = True <EOL> self . exitstatus = None <EOL> self . signalstatus = None <EOL> self . status = None <EOL> self . flag_eof = False <EOL> self . pid = None <EOL> self . child_fd = - <NUM_LIT:1> <EOL> self . timeout = timeout <EOL> self . delimiter = EOF <EOL> self . logfile = logfile <EOL> self . logfile_read = None <EOL> self . logfile_send = None <EOL> self . maxread = maxread <EOL> self . buffer = self . string_type ( ) <EOL> self . searchwindowsize = searchwindowsize <EOL> self . delaybeforesend = <NUM_LIT> <EOL> self . delayafterclose = <NUM_LIT:0.1> <EOL> self . delayafterterminate = <NUM_LIT:0.1> <EOL> self . softspace = False <EOL> self . name = '<STR_LIT:<>' + repr ( self ) + '<STR_LIT:>>' <EOL> self . closed = True <EOL> self . cwd = cwd <EOL> self . env = env <EOL> self . echo = echo <EOL> self . ignore_sighup = ignore_sighup <EOL> _platform = sys . platform . lower ( ) <EOL> self . __irix_hack = _platform . startswith ( '<STR_LIT>' ) <EOL> self . use_native_pty_fork = not ( <EOL> _platform . startswith ( '<STR_LIT>' ) or <EOL> _platform . startswith ( '<STR_LIT>' ) ) <EOL> if pty : <EOL> try : <EOL> from termios import VEOF , VINTR <EOL> try : <EOL> fd = sys . __stdin__ . fileno ( ) <EOL> except ValueError : <EOL> fd = sys . __stdout__ . fileno ( ) <EOL> self . _INTR = ord ( termios . tcgetattr ( fd ) [ <NUM_LIT:6> ] [ VINTR ] ) <EOL> self . _EOF = ord ( termios . tcgetattr ( fd ) [ <NUM_LIT:6> ] [ VEOF ] ) <EOL> except ( ImportError , OSError , IOError , ValueError , termios . error ) : <EOL> try : <EOL> from termios import CEOF , CINTR <EOL> ( self . _INTR , self . _EOF ) = ( CINTR , CEOF ) <EOL> except ImportError : <EOL> ( self . _INTR , self . _EOF ) = ( <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> if command is None : <EOL> self . command = None <EOL> self . args = None <EOL> self . name = '<STR_LIT>' <EOL> else : <EOL> sig = signal . signal ( signal . SIGINT , signal . SIG_DFL ) <EOL> try : <EOL> self . _spawn ( command , args ) <EOL> finally : <EOL> signal . signal ( signal . SIGINT , sig ) <EOL> @ staticmethod <EOL> def _coerce_expect_string ( s ) : <EOL> if not isinstance ( s , bytes ) : <EOL> return s . encode ( '<STR_LIT:ascii>' ) <EOL> return s <EOL> @ staticmethod <EOL> def _coerce_send_string ( s ) : <EOL> if not isinstance ( s , bytes ) : <EOL> return s . encode ( '<STR_LIT:utf-8>' ) <EOL> return s <EOL> @ staticmethod <EOL> def _coerce_read_string ( s ) : <EOL> return s <EOL> def __del__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> if not self . closed : <EOL> try : <EOL> self . close ( ) <EOL> except : <EOL> pass <EOL> def __str__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> s = [ ] <EOL> s . append ( repr ( self ) ) <EOL> s . append ( '<STR_LIT>' + __version__ ) <EOL> s . append ( '<STR_LIT>' + str ( self . command ) ) <EOL> s . append ( '<STR_LIT>' % ( self . args , ) ) <EOL> s . append ( '<STR_LIT>' % ( self . searcher , ) ) <EOL> s . append ( '<STR_LIT>' % ( self . buffer ) [ - <NUM_LIT:100> : ] , ) <EOL> s . append ( '<STR_LIT>' % ( self . before ) [ - <NUM_LIT:100> : ] , ) <EOL> s . append ( '<STR_LIT>' % ( self . after , ) ) <EOL> s . append ( '<STR_LIT>' % ( self . match , ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . match_index ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . exitstatus ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . flag_eof ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . pid ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . child_fd ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . closed ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . timeout ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . delimiter ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . logfile ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . logfile_read ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . logfile_send ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . maxread ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . ignorecase ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . searchwindowsize ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . delaybeforesend ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . delayafterclose ) ) <EOL> s . append ( '<STR_LIT>' + str ( self . delayafterterminate ) ) <EOL> return '<STR_LIT:\n>' . join ( s ) <EOL> def _spawn ( self , command , args = [ ] ) : <EOL> '''<STR_LIT>''' <EOL> if isinstance ( command , type ( <NUM_LIT:0> ) ) : <EOL> raise ExceptionPexpect ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> if not isinstance ( args , type ( [ ] ) ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> if args == [ ] : <EOL> self . args = split_command_line ( command ) <EOL> self . command = self . args [ <NUM_LIT:0> ] <EOL> else : <EOL> self . args = args [ : ] <EOL> self . args . insert ( <NUM_LIT:0> , command ) <EOL> self . command = command <EOL> command_with_path = which ( self . command ) <EOL> if command_with_path is None : <EOL> raise ExceptionPexpect ( '<STR_LIT>' + <EOL> '<STR_LIT>' % self . command ) <EOL> self . command = command_with_path <EOL> self . args [ <NUM_LIT:0> ] = self . command <EOL> self . name = '<STR_LIT:<>' + '<STR_LIT:U+0020>' . join ( self . args ) + '<STR_LIT:>>' <EOL> assert self . pid is None , '<STR_LIT>' <EOL> assert self . command is not None , '<STR_LIT>' <EOL> self . terminated = False <EOL> self . closed = False <EOL> if not pty : <EOL> self . _winspawn ( ) <EOL> return <EOL> if self . use_native_pty_fork : <EOL> try : <EOL> self . pid , self . child_fd = pty . fork ( ) <EOL> except OSError : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> raise ExceptionPexpect ( '<STR_LIT>' + str ( err ) ) <EOL> else : <EOL> self . pid , self . child_fd = self . __fork_pty ( ) <EOL> if self . pid == pty . CHILD : <EOL> self . child_fd = self . STDIN_FILENO <EOL> try : <EOL> self . setwinsize ( <NUM_LIT> , <NUM_LIT> ) <EOL> except IOError as err : <EOL> if err . args [ <NUM_LIT:0> ] not in ( errno . EINVAL , errno . ENOTTY ) : <EOL> raise <EOL> if not self . echo : <EOL> try : <EOL> self . setecho ( self . echo ) <EOL> except ( IOError , termios . error ) as err : <EOL> if err . args [ <NUM_LIT:0> ] not in ( errno . EINVAL , errno . ENOTTY ) : <EOL> raise <EOL> max_fd = resource . getrlimit ( resource . RLIMIT_NOFILE ) [ <NUM_LIT:0> ] <EOL> os . closerange ( <NUM_LIT:3> , max_fd ) <EOL> if self . ignore_sighup : <EOL> signal . signal ( signal . SIGHUP , signal . SIG_IGN ) <EOL> if self . cwd is not None : <EOL> os . chdir ( self . cwd ) <EOL> if self . env is None : <EOL> os . execv ( self . command , self . args ) <EOL> else : <EOL> os . execvpe ( self . command , self . args , self . env ) <EOL> try : <EOL> self . setwinsize ( <NUM_LIT> , <NUM_LIT> ) <EOL> except IOError as err : <EOL> if err . args [ <NUM_LIT:0> ] not in ( errno . EINVAL , errno . ENOTTY ) : <EOL> raise <EOL> def _winspawn ( self ) : <EOL> self . echo = False <EOL> kwargs = ( dict ( bufsize = <NUM_LIT:0> , stdin = subprocess . PIPE , <EOL> stderr = subprocess . STDOUT , <EOL> stdout = subprocess . PIPE ) ) <EOL> if self . env : <EOL> kwargs [ '<STR_LIT>' ] = self . env <EOL> startupinfo = subprocess . STARTUPINFO ( ) <EOL> startupinfo . dwFlags |= subprocess . STARTF_USESHOWWINDOW <EOL> kwargs [ '<STR_LIT>' ] = startupinfo <EOL> kwargs [ '<STR_LIT>' ] = subprocess . CREATE_NEW_PROCESS_GROUP <EOL> self . proc = subprocess . Popen ( self . args , ** kwargs ) <EOL> self . child_fd = self . proc . stdout . fileno ( ) <EOL> self . write_fd = self . proc . stdin . fileno ( ) <EOL> self . pid = self . proc . pid <EOL> self . _read_queue = Queue . Queue ( ) <EOL> self . _read_thread = threading . Thread ( target = self . _read_incoming ) <EOL> self . _read_thread . setDaemon ( True ) <EOL> self . _read_thread . start ( ) <EOL> def fileno ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . child_fd <EOL> def close ( self , force = True ) : <EOL> '''<STR_LIT>''' <EOL> if not self . closed : <EOL> self . flush ( ) <EOL> os . close ( self . child_fd ) <EOL> time . sleep ( self . delayafterclose ) <EOL> if self . isalive ( ) : <EOL> if not self . terminate ( force ) : <EOL> raise ExceptionPexpect ( '<STR_LIT>' ) <EOL> self . child_fd = - <NUM_LIT:1> <EOL> self . closed = True <EOL> def flush ( self ) : <EOL> '''<STR_LIT>''' <EOL> pass <EOL> def isatty ( self ) : <EOL> '''<STR_LIT>''' <EOL> return os . isatty ( self . child_fd ) <EOL> def waitnoecho ( self , timeout = - <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> if not self . isatty ( ) : <EOL> return <EOL> if timeout == - <NUM_LIT:1> : <EOL> timeout = self . timeout <EOL> if timeout is not None : <EOL> end_time = time . time ( ) + timeout <EOL> while True : <EOL> if not self . getecho ( ) : <EOL> return True <EOL> if timeout < <NUM_LIT:0> and timeout is not None : <EOL> return False <EOL> if timeout is not None : <EOL> timeout = end_time - time . time ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> def getecho ( self ) : <EOL> '''<STR_LIT>''' <EOL> if not self . isatty ( ) : <EOL> return self . echo <EOL> try : <EOL> attr = termios . tcgetattr ( self . child_fd ) <EOL> except termios . error as err : <EOL> errmsg = '<STR_LIT>' <EOL> if err . args [ <NUM_LIT:0> ] == errno . EINVAL : <EOL> raise IOError ( err . args [ <NUM_LIT:0> ] , '<STR_LIT>' % ( err . args [ <NUM_LIT:1> ] , errmsg ) ) <EOL> raise <EOL> self . echo = bool ( attr [ <NUM_LIT:3> ] & termios . ECHO ) <EOL> return self . echo <EOL> def setecho ( self , state ) : <EOL> '''<STR_LIT>''' <EOL> if not self . isatty ( ) : <EOL> self . echo = state <EOL> return <EOL> errmsg = '<STR_LIT>' <EOL> try : <EOL> attr = termios . tcgetattr ( self . child_fd ) <EOL> except termios . error as err : <EOL> if err . args [ <NUM_LIT:0> ] == errno . EINVAL : <EOL> raise IOError ( err . args [ <NUM_LIT:0> ] , '<STR_LIT>' % ( err . args [ <NUM_LIT:1> ] , errmsg ) ) <EOL> raise <EOL> if state : <EOL> attr [ <NUM_LIT:3> ] = attr [ <NUM_LIT:3> ] | termios . ECHO <EOL> else : <EOL> attr [ <NUM_LIT:3> ] = attr [ <NUM_LIT:3> ] & ~ termios . ECHO <EOL> try : <EOL> termios . tcsetattr ( self . child_fd , termios . TCSANOW , attr ) <EOL> except IOError as err : <EOL> if err . args [ <NUM_LIT:0> ] == errno . EINVAL : <EOL> raise IOError ( err . args [ <NUM_LIT:0> ] , '<STR_LIT>' % ( err . args [ <NUM_LIT:1> ] , errmsg ) ) <EOL> raise <EOL> self . echo = state <EOL> def _log ( self , s , direction ) : <EOL> if self . logfile is not None : <EOL> self . logfile . write ( s ) <EOL> self . logfile . flush ( ) <EOL> second_log = self . logfile_send if ( direction == '<STR_LIT>' ) else self . logfile_read <EOL> if second_log is not None : <EOL> second_log . write ( s ) <EOL> second_log . flush ( ) <EOL> def _read_incoming ( self ) : <EOL> while <NUM_LIT:1> : <EOL> buf = '<STR_LIT>' <EOL> try : <EOL> buf = os . read ( self . child_fd , <NUM_LIT> ) <EOL> except OSError as e : <EOL> self . _log ( e , '<STR_LIT>' ) <EOL> if not buf : <EOL> self . flag_eof = True <EOL> return <EOL> self . _read_queue . put ( buf . decode ( '<STR_LIT:utf-8>' ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> def _winread ( self , size , timeout ) : <EOL> t0 = time . time ( ) <EOL> buf = '<STR_LIT>' <EOL> if timeout is None or timeout == - <NUM_LIT:1> : <EOL> timeout = <NUM_LIT:30> <EOL> while len ( buf ) < size : <EOL> try : <EOL> incoming = self . _read_queue . get ( True , <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> break <EOL> except KeyboardInterrupt : <EOL> self . sendintr ( ) <EOL> break <EOL> else : <EOL> if not incoming : <EOL> self . flag_eof = True <EOL> raise EOF ( '<STR_LIT>' ) <EOL> buf += incoming <EOL> if ( time . time ( ) - t0 ) >= timeout : <EOL> break <EOL> if len ( buf ) > size : <EOL> self . buffer = buf [ size : ] <EOL> buf = buf [ : size ] <EOL> buf = self . _coerce_read_string ( buf ) <EOL> self . _log ( buf , '<STR_LIT>' ) <EOL> return buf <EOL> def read_nonblocking ( self , size = <NUM_LIT:1> , timeout = - <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> if self . closed : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if timeout == - <NUM_LIT:1> : <EOL> timeout = self . timeout <EOL> if not pty : <EOL> return self . _winread ( size , timeout ) <EOL> if not self . isalive ( ) : <EOL> r , w , e = self . __select ( [ self . child_fd ] , [ ] , [ ] , <NUM_LIT:0> ) <EOL> if not r : <EOL> self . flag_eof = True <EOL> raise EOF ( '<STR_LIT>' ) <EOL> elif self . __irix_hack : <EOL> r , w , e = self . __select ( [ self . child_fd ] , [ ] , [ ] , <NUM_LIT:2> ) <EOL> if not r and not self . isalive ( ) : <EOL> self . flag_eof = True <EOL> raise EOF ( '<STR_LIT>' ) <EOL> r , w , e = self . __select ( [ self . child_fd ] , [ ] , [ ] , timeout ) <EOL> if not r : <EOL> if not self . isalive ( ) : <EOL> self . flag_eof = True <EOL> raise EOF ( '<STR_LIT>' ) <EOL> else : <EOL> raise TIMEOUT ( '<STR_LIT>' ) <EOL> if self . child_fd in r : <EOL> try : <EOL> s = os . read ( self . child_fd , size ) <EOL> except OSError as err : <EOL> if err . args [ <NUM_LIT:0> ] == errno . EIO : <EOL> self . flag_eof = True <EOL> raise EOF ( '<STR_LIT>' ) <EOL> raise <EOL> if s == b'<STR_LIT>' : <EOL> self . flag_eof = True <EOL> raise EOF ( '<STR_LIT>' ) <EOL> s = self . _coerce_read_string ( s ) <EOL> self . _log ( s , '<STR_LIT>' ) <EOL> return s <EOL> raise ExceptionPexpect ( '<STR_LIT>' ) <EOL> def read ( self , size = - <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> if size == <NUM_LIT:0> : <EOL> return self . string_type ( ) <EOL> if size < <NUM_LIT:0> : <EOL> self . expect ( self . delimiter ) <EOL> return self . before <EOL> cre = re . compile ( self . _coerce_expect_string ( '<STR_LIT>' % size ) , re . DOTALL ) <EOL> index = self . expect ( [ cre , self . delimiter ] ) <EOL> if index == <NUM_LIT:0> : <EOL> return self . after <EOL> return self . before <EOL> def readline ( self , size = - <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> if size == <NUM_LIT:0> : <EOL> return self . string_type ( ) <EOL> index = self . expect ( [ self . crlf , self . lf , self . delimiter ] ) <EOL> if index == <NUM_LIT:0> : <EOL> return self . before + self . crlf <EOL> elif index == <NUM_LIT:1> : <EOL> return self . before + '<STR_LIT:\n>' <EOL> else : <EOL> return self . before <EOL> def __iter__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> return iter ( self . readline , self . string_type ( ) ) <EOL> def readlines ( self , sizehint = - <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> lines = [ ] <EOL> while True : <EOL> line = self . readline ( ) <EOL> if not line : <EOL> break <EOL> lines . append ( line ) <EOL> return lines <EOL> def write ( self , s ) : <EOL> '''<STR_LIT>''' <EOL> self . send ( s ) <EOL> def writelines ( self , sequence ) : <EOL> '''<STR_LIT>''' <EOL> for s in sequence : <EOL> self . write ( s ) <EOL> def send ( self , s ) : <EOL> '''<STR_LIT>''' <EOL> time . sleep ( self . delaybeforesend ) <EOL> s = self . _coerce_send_string ( s ) <EOL> self . _log ( s , '<STR_LIT>' ) <EOL> return self . _send ( s ) <EOL> def _send ( self , s ) : <EOL> if not self . isalive ( ) : <EOL> raise EOF ( '<STR_LIT>' ) <EOL> if pty : <EOL> return os . write ( self . child_fd , s ) <EOL> else : <EOL> try : <EOL> return os . write ( self . write_fd , s ) <EOL> except OSError as e : <EOL> if e . errno == errno . EINVAL : <EOL> raise EOF ( '<STR_LIT>' ) <EOL> else : <EOL> raise ( e ) <EOL> def sendline ( self , s = '<STR_LIT>' ) : <EOL> '''<STR_LIT>''' <EOL> n = self . send ( s ) <EOL> n = n + self . send ( self . linesep ) <EOL> return n <EOL> def sendcontrol ( self , char ) : <EOL> '''<STR_LIT>''' <EOL> char = char . lower ( ) <EOL> a = ord ( char ) <EOL> if a >= <NUM_LIT> and a <= <NUM_LIT> : <EOL> a = a - ord ( '<STR_LIT:a>' ) + <NUM_LIT:1> <EOL> return self . send ( self . _chr ( a ) ) <EOL> d = { '<STR_LIT:@>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT:[>' : <NUM_LIT> , '<STR_LIT:{>' : <NUM_LIT> , <EOL> '<STR_LIT:\\>' : <NUM_LIT> , '<STR_LIT:|>' : <NUM_LIT> , <EOL> '<STR_LIT:]>' : <NUM_LIT> , '<STR_LIT:}>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:30> , '<STR_LIT>' : <NUM_LIT:30> , <EOL> '<STR_LIT:_>' : <NUM_LIT> , <EOL> '<STR_LIT:?>' : <NUM_LIT> } <EOL> if char not in d : <EOL> return <NUM_LIT:0> <EOL> return self . send ( self . _chr ( d [ char ] ) ) <EOL> def sendeof ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . send ( self . _chr ( self . _EOF ) ) <EOL> def sendintr ( self ) : <EOL> '''<STR_LIT>''' <EOL> if pty : <EOL> self . send ( self . _chr ( self . _INTR ) ) <EOL> else : <EOL> self . kill ( signal . CTRL_BREAK_EVENT ) <EOL> def eof ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . flag_eof <EOL> def terminate ( self , force = False ) : <EOL> '''<STR_LIT>''' <EOL> if not self . isalive ( ) : <EOL> return True <EOL> if not pty : <EOL> self . kill ( signal . SIGTERM ) <EOL> return True <EOL> try : <EOL> self . kill ( signal . SIGHUP ) <EOL> time . sleep ( self . delayafterterminate ) <EOL> if not self . isalive ( ) : <EOL> return True <EOL> self . kill ( signal . SIGCONT ) <EOL> time . sleep ( self . delayafterterminate ) <EOL> if not self . isalive ( ) : <EOL> return True <EOL> self . kill ( signal . SIGINT ) <EOL> time . sleep ( self . delayafterterminate ) <EOL> if not self . isalive ( ) : <EOL> return True <EOL> if force : <EOL> self . kill ( signal . SIGKILL ) <EOL> time . sleep ( self . delayafterterminate ) <EOL> if not self . isalive ( ) : <EOL> return True <EOL> else : <EOL> return False <EOL> return False <EOL> except OSError : <EOL> time . sleep ( self . delayafterterminate ) <EOL> if not self . isalive ( ) : <EOL> return True <EOL> else : <EOL> return False <EOL> def wait ( self ) : <EOL> '''<STR_LIT>''' <EOL> if self . isalive ( ) : <EOL> pid , status = os . waitpid ( self . pid , <NUM_LIT:0> ) <EOL> else : <EOL> raise ExceptionPexpect ( '<STR_LIT>' ) <EOL> self . exitstatus = os . WEXITSTATUS ( status ) <EOL> if os . WIFEXITED ( status ) : <EOL> self . status = status <EOL> self . exitstatus = os . WEXITSTATUS ( status ) <EOL> self . signalstatus = None <EOL> self . terminated = True <EOL> elif os . WIFSIGNALED ( status ) : <EOL> self . status = status <EOL> self . exitstatus = None <EOL> self . signalstatus = os . WTERMSIG ( status ) <EOL> self . terminated = True <EOL> elif os . WIFSTOPPED ( status ) : <EOL> raise ExceptionPexpect ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> return self . exitstatus <EOL> def isalive ( self ) : <EOL> '''<STR_LIT>''' <EOL> if self . terminated : <EOL> return False <EOL> if not pty : <EOL> return self . _win_isalive ( ) <EOL> if self . flag_eof : <EOL> waitpid_options = <NUM_LIT:0> <EOL> else : <EOL> waitpid_options = os . WNOHANG <EOL> try : <EOL> pid , status = os . waitpid ( self . pid , waitpid_options ) <EOL> except OSError : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> if err . errno == errno . ECHILD : <EOL> raise ExceptionPexpect ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> else : <EOL> raise err <EOL> if pid == <NUM_LIT:0> : <EOL> try : <EOL> pid , status = os . waitpid ( self . pid , waitpid_options ) <EOL> except OSError as e : <EOL> if e . errno == errno . ECHILD : <EOL> raise ExceptionPexpect ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> else : <EOL> raise <EOL> if pid == <NUM_LIT:0> : <EOL> return True <EOL> if pid == <NUM_LIT:0> : <EOL> return True <EOL> if os . WIFEXITED ( status ) : <EOL> self . status = status <EOL> self . exitstatus = os . WEXITSTATUS ( status ) <EOL> self . signalstatus = None <EOL> self . terminated = True <EOL> elif os . WIFSIGNALED ( status ) : <EOL> self . status = status <EOL> self . exitstatus = None <EOL> self . signalstatus = os . WTERMSIG ( status ) <EOL> self . terminated = True <EOL> elif os . WIFSTOPPED ( status ) : <EOL> raise ExceptionPexpect ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> return False <EOL> def _win_isalive ( self ) : <EOL> kernel32 = ctypes . windll . kernel32 <EOL> SYNCHRONIZE = <NUM_LIT> <EOL> process = kernel32 . OpenProcess ( SYNCHRONIZE , <NUM_LIT:0> , self . pid ) <EOL> if process != <NUM_LIT:0> : <EOL> kernel32 . CloseHandle ( process ) <EOL> return True <EOL> else : <EOL> return False <EOL> def kill ( self , sig ) : <EOL> '''<STR_LIT>''' <EOL> if not pty : <EOL> if sig in [ signal . SIGINT , signal . CTRL_C_EVENT ] : <EOL> sig = signal . CTRL_C_EVENT <EOL> elif sig in [ signal . SIGBREAK , signal . CTRL_BREAK_EVENT ] : <EOL> sig = signal . CTRL_BREAK_EVENT <EOL> else : <EOL> sig = signal . SIGTERM <EOL> if self . isalive ( ) : <EOL> os . kill ( self . pid , sig ) <EOL> def _pattern_type_err ( self , pattern ) : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( badtype = type ( pattern ) , <EOL> badobj = pattern , <EOL> goodtypes = '<STR_LIT:U+002CU+0020>' . join ( [ str ( ast ) for ast in self . allowed_string_types ] ) <EOL> ) <EOL> ) <EOL> def compile_pattern_list ( self , patterns ) : <EOL> '''<STR_LIT>''' <EOL> if patterns is None : <EOL> return [ ] <EOL> if not isinstance ( patterns , list ) : <EOL> patterns = [ patterns ] <EOL> compile_flags = re . DOTALL <EOL> if self . ignorecase : <EOL> compile_flags = compile_flags | re . IGNORECASE <EOL> compiled_pattern_list = [ ] <EOL> for idx , p in enumerate ( patterns ) : <EOL> if isinstance ( p , self . allowed_string_types ) : <EOL> p = self . _coerce_expect_string ( p ) <EOL> compiled_pattern_list . append ( re . compile ( p , compile_flags ) ) <EOL> elif p is EOF : <EOL> compiled_pattern_list . append ( EOF ) <EOL> elif p is TIMEOUT : <EOL> compiled_pattern_list . append ( TIMEOUT ) <EOL> elif isinstance ( p , type ( re . compile ( '<STR_LIT>' ) ) ) : <EOL> compiled_pattern_list . append ( p ) <EOL> else : <EOL> self . _pattern_type_err ( p ) <EOL> return compiled_pattern_list <EOL> def expect ( self , pattern , timeout = - <NUM_LIT:1> , searchwindowsize = - <NUM_LIT:1> , async = False ) : <EOL> '''<STR_LIT>''' <EOL> compiled_pattern_list = self . compile_pattern_list ( pattern ) <EOL> return self . expect_list ( compiled_pattern_list , <EOL> timeout , searchwindowsize , async ) <EOL> def expect_list ( self , pattern_list , timeout = - <NUM_LIT:1> , searchwindowsize = - <NUM_LIT:1> , <EOL> async = False ) : <EOL> '''<STR_LIT>''' <EOL> if timeout == - <NUM_LIT:1> : <EOL> timeout = self . timeout <EOL> exp = Expecter ( self , searcher_re ( pattern_list ) , searchwindowsize ) <EOL> return exp . expect_loop ( timeout ) <EOL> def expect_exact ( self , pattern_list , timeout = - <NUM_LIT:1> , searchwindowsize = - <NUM_LIT:1> , <EOL> async = False ) : <EOL> '''<STR_LIT>''' <EOL> if timeout == - <NUM_LIT:1> : <EOL> timeout = self . timeout <EOL> if ( isinstance ( pattern_list , self . allowed_string_types ) or <EOL> pattern_list in ( TIMEOUT , EOF ) ) : <EOL> pattern_list = [ pattern_list ] <EOL> def prepare_pattern ( pattern ) : <EOL> if pattern in ( TIMEOUT , EOF ) : <EOL> return pattern <EOL> if isinstance ( pattern , self . allowed_string_types ) : <EOL> return self . _coerce_expect_string ( pattern ) <EOL> self . _pattern_type_err ( pattern ) <EOL> try : <EOL> pattern_list = iter ( pattern_list ) <EOL> except TypeError : <EOL> self . _pattern_type_err ( pattern_list ) <EOL> pattern_list = [ prepare_pattern ( p ) for p in pattern_list ] <EOL> exp = Expecter ( self , searcher_string ( pattern_list ) , searchwindowsize ) <EOL> return exp . expect_loop ( timeout ) <EOL> def expect_loop ( self , searcher , timeout = - <NUM_LIT:1> , searchwindowsize = - <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> exp = Expecter ( self , searcher , searchwindowsize ) <EOL> return exp . expect_gloop ( timeout ) <EOL> def getwinsize ( self ) : <EOL> '''<STR_LIT>''' <EOL> if not self . isatty ( ) : <EOL> return ( - <NUM_LIT:1> , - <NUM_LIT:1> ) <EOL> TIOCGWINSZ = getattr ( termios , '<STR_LIT>' , <NUM_LIT> ) <EOL> s = struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> x = fcntl . ioctl ( self . child_fd , TIOCGWINSZ , s ) <EOL> return struct . unpack ( '<STR_LIT>' , x ) [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> def setwinsize ( self , rows , cols ) : <EOL> '''<STR_LIT>''' <EOL> if not self . isatty ( ) : <EOL> return <EOL> TIOCSWINSZ = getattr ( termios , '<STR_LIT>' , - <NUM_LIT> ) <EOL> s = struct . pack ( '<STR_LIT>' , rows , cols , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> fcntl . ioctl ( self . fileno ( ) , TIOCSWINSZ , s ) <EOL> def interact ( self , escape_character = chr ( <NUM_LIT> ) , <EOL> input_filter = None , output_filter = None ) : <EOL> '''<STR_LIT>''' <EOL> if not self . isatty ( ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> self . write_to_stdout ( self . buffer ) <EOL> self . stdout . flush ( ) <EOL> self . buffer = self . string_type ( ) <EOL> mode = tty . tcgetattr ( self . STDIN_FILENO ) <EOL> tty . setraw ( self . STDIN_FILENO ) <EOL> if PY3 : <EOL> escape_character = escape_character . encode ( '<STR_LIT>' ) <EOL> try : <EOL> self . __interact_copy ( escape_character , input_filter , output_filter ) <EOL> finally : <EOL> tty . tcsetattr ( self . STDIN_FILENO , tty . TCSAFLUSH , mode ) <EOL> def __interact_writen ( self , fd , data ) : <EOL> '''<STR_LIT>''' <EOL> while data != b'<STR_LIT>' and self . isalive ( ) : <EOL> n = os . write ( fd , data ) <EOL> data = data [ n : ] <EOL> def __interact_read ( self , fd ) : <EOL> '''<STR_LIT>''' <EOL> return os . read ( fd , <NUM_LIT:1000> ) <EOL> def __interact_copy ( self , escape_character = None , <EOL> input_filter = None , output_filter = None ) : <EOL> '''<STR_LIT>''' <EOL> while self . isalive ( ) : <EOL> r , w , e = self . __select ( [ self . child_fd , self . STDIN_FILENO ] , [ ] , [ ] ) <EOL> if self . child_fd in r : <EOL> try : <EOL> data = self . __interact_read ( self . child_fd ) <EOL> except OSError as err : <EOL> if err . args [ <NUM_LIT:0> ] == errno . EIO : <EOL> break <EOL> raise <EOL> if data == b'<STR_LIT>' : <EOL> break <EOL> if output_filter : <EOL> data = output_filter ( data ) <EOL> if self . logfile is not None : <EOL> self . logfile . write ( data ) <EOL> self . logfile . flush ( ) <EOL> os . write ( self . STDOUT_FILENO , data ) <EOL> if self . STDIN_FILENO in r : <EOL> data = self . __interact_read ( self . STDIN_FILENO ) <EOL> if input_filter : <EOL> data = input_filter ( data ) <EOL> i = data . rfind ( escape_character ) <EOL> if i != - <NUM_LIT:1> : <EOL> data = data [ : i ] <EOL> self . __interact_writen ( self . child_fd , data ) <EOL> break <EOL> self . __interact_writen ( self . child_fd , data ) <EOL> def __select ( self , iwtd , owtd , ewtd , timeout = None ) : <EOL> '''<STR_LIT>''' <EOL> if timeout is not None : <EOL> end_time = time . time ( ) + timeout <EOL> while True : <EOL> try : <EOL> return select . select ( iwtd , owtd , ewtd , timeout ) <EOL> except select . error : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> if err . args [ <NUM_LIT:0> ] == errno . EINTR : <EOL> if timeout is not None : <EOL> timeout = end_time - time . time ( ) <EOL> if timeout < <NUM_LIT:0> : <EOL> return ( [ ] , [ ] , [ ] ) <EOL> else : <EOL> raise <EOL> class spawnu ( spawn ) : <EOL> """<STR_LIT>""" <EOL> if PY3 : <EOL> string_type = str <EOL> allowed_string_types = ( str , ) <EOL> _chr = staticmethod ( chr ) <EOL> linesep = os . linesep <EOL> crlf = '<STR_LIT:\r\n>' <EOL> lf = '<STR_LIT:\n>' <EOL> else : <EOL> string_type = unicode <EOL> allowed_string_types = ( unicode , ) <EOL> _chr = staticmethod ( unichr ) <EOL> linesep = os . linesep . decode ( '<STR_LIT:ascii>' ) <EOL> crlf = '<STR_LIT:\r\n>' . decode ( '<STR_LIT:ascii>' ) <EOL> lf = '<STR_LIT:\n>' . decode ( '<STR_LIT:ascii>' ) <EOL> write_to_stdout = sys . stdout . write <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . encoding = kwargs . pop ( '<STR_LIT>' , '<STR_LIT:utf-8>' ) <EOL> self . errors = kwargs . pop ( '<STR_LIT>' , '<STR_LIT:strict>' ) <EOL> self . _decoder = codecs . getincrementaldecoder ( self . encoding ) ( errors = self . errors ) <EOL> super ( spawnu , self ) . __init__ ( * args , ** kwargs ) <EOL> @ staticmethod <EOL> def _coerce_expect_string ( s ) : <EOL> return s <EOL> @ staticmethod <EOL> def _coerce_send_string ( s ) : <EOL> return s <EOL> def _coerce_read_string ( self , s ) : <EOL> try : <EOL> return self . _decoder . decode ( s , final = False ) <EOL> except TypeError : <EOL> return s <EOL> def _send ( self , s ) : <EOL> return super ( spawnu , self ) . _send ( s . encode ( self . encoding , self . errors ) ) <EOL> class searcher_string ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , strings ) : <EOL> '''<STR_LIT>''' <EOL> self . eof_index = - <NUM_LIT:1> <EOL> self . timeout_index = - <NUM_LIT:1> <EOL> self . _strings = [ ] <EOL> for n , s in enumerate ( strings ) : <EOL> if s is EOF : <EOL> self . eof_index = n <EOL> continue <EOL> if s is TIMEOUT : <EOL> self . timeout_index = n <EOL> continue <EOL> self . _strings . append ( ( n , s ) ) <EOL> def __str__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> ss = [ ( ns [ <NUM_LIT:0> ] , '<STR_LIT>' % ns ) for ns in self . _strings ] <EOL> ss . append ( ( - <NUM_LIT:1> , '<STR_LIT>' ) ) <EOL> if self . eof_index >= <NUM_LIT:0> : <EOL> ss . append ( ( self . eof_index , '<STR_LIT>' % self . eof_index ) ) <EOL> if self . timeout_index >= <NUM_LIT:0> : <EOL> ss . append ( ( self . timeout_index , <EOL> '<STR_LIT>' % self . timeout_index ) ) <EOL> ss . sort ( ) <EOL> ss = list ( zip ( * ss ) ) [ <NUM_LIT:1> ] <EOL> return '<STR_LIT:\n>' . join ( ss ) <EOL> def search ( self , buffer , freshlen , searchwindowsize = None ) : <EOL> '''<STR_LIT>''' <EOL> first_match = None <EOL> for index , s in self . _strings : <EOL> if searchwindowsize is None : <EOL> offset = - ( freshlen + len ( s ) ) <EOL> else : <EOL> offset = - searchwindowsize <EOL> n = buffer . find ( s , offset ) <EOL> if n >= <NUM_LIT:0> and ( first_match is None or n < first_match ) : <EOL> first_match = n <EOL> best_index , best_match = index , s <EOL> if first_match is None : <EOL> return - <NUM_LIT:1> <EOL> self . match = best_match <EOL> self . start = first_match <EOL> self . end = self . start + len ( self . match ) <EOL> return best_index <EOL> class searcher_re ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , patterns ) : <EOL> '''<STR_LIT>''' <EOL> self . eof_index = - <NUM_LIT:1> <EOL> self . timeout_index = - <NUM_LIT:1> <EOL> self . _searches = [ ] <EOL> for n , s in zip ( list ( range ( len ( patterns ) ) ) , patterns ) : <EOL> if s is EOF : <EOL> self . eof_index = n <EOL> continue <EOL> if s is TIMEOUT : <EOL> self . timeout_index = n <EOL> continue <EOL> self . _searches . append ( ( n , s ) ) <EOL> def __str__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> ss = list ( ) <EOL> for n , s in self . _searches : <EOL> try : <EOL> ss . append ( ( n , '<STR_LIT>' % ( n , s . pattern ) ) ) <EOL> except UnicodeEncodeError : <EOL> ss . append ( ( n , '<STR_LIT>' % ( n , s . pattern ) ) ) <EOL> ss . append ( ( - <NUM_LIT:1> , '<STR_LIT>' ) ) <EOL> if self . eof_index >= <NUM_LIT:0> : <EOL> ss . append ( ( self . eof_index , '<STR_LIT>' % self . eof_index ) ) <EOL> if self . timeout_index >= <NUM_LIT:0> : <EOL> ss . append ( ( self . timeout_index , '<STR_LIT>' % <EOL> self . timeout_index ) ) <EOL> ss . sort ( ) <EOL> ss = list ( zip ( * ss ) ) [ <NUM_LIT:1> ] <EOL> return '<STR_LIT:\n>' . join ( ss ) <EOL> def search ( self , buffer , freshlen , searchwindowsize = None ) : <EOL> '''<STR_LIT>''' <EOL> first_match = None <EOL> if searchwindowsize is None : <EOL> searchstart = <NUM_LIT:0> <EOL> else : <EOL> searchstart = max ( <NUM_LIT:0> , len ( buffer ) - searchwindowsize ) <EOL> for index , s in self . _searches : <EOL> match = s . search ( buffer , searchstart ) <EOL> if match is None : <EOL> continue <EOL> n = match . start ( ) <EOL> if first_match is None or n < first_match : <EOL> first_match = n <EOL> the_match = match <EOL> best_index = index <EOL> if first_match is None : <EOL> return - <NUM_LIT:1> <EOL> self . start = first_match <EOL> self . match = the_match <EOL> self . end = self . match . end ( ) <EOL> return best_index <EOL> def is_executable_file ( path ) : <EOL> """<STR_LIT>""" <EOL> fpath = os . path . realpath ( path ) <EOL> if not os . path . isfile ( fpath ) : <EOL> return False <EOL> mode = os . stat ( fpath ) . st_mode <EOL> if mode & stat . S_IROTH and mode & stat . S_IXOTH : <EOL> return True <EOL> user_gids = os . getgroups ( ) + [ os . getgid ( ) ] <EOL> if ( os . stat ( fpath ) . st_gid in user_gids and <EOL> mode & stat . S_IRGRP and mode & stat . S_IXGRP ) : <EOL> return True <EOL> user_gids = os . getgroups ( ) + [ os . getgid ( ) ] <EOL> if ( os . stat ( fpath ) . st_uid == os . geteuid ( ) and <EOL> mode & stat . S_IRUSR and mode & stat . S_IXUSR ) : <EOL> return True <EOL> return False <EOL> def which ( filename ) : <EOL> '''<STR_LIT>''' <EOL> if os . path . dirname ( filename ) != '<STR_LIT>' and is_executable_file ( filename ) : <EOL> return filename <EOL> if '<STR_LIT>' not in os . environ or os . environ [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> p = os . defpath <EOL> else : <EOL> p = os . environ [ '<STR_LIT>' ] <EOL> pathlist = p . split ( os . pathsep ) <EOL> for path in pathlist : <EOL> ff = os . path . join ( path , filename ) <EOL> if pty : <EOL> if is_executable_file ( ff ) : <EOL> return ff <EOL> else : <EOL> pathext = os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> pathext = pathext . split ( os . pathsep ) + [ '<STR_LIT>' ] <EOL> for ext in pathext : <EOL> if os . access ( ff + ext , os . X_OK ) : <EOL> return ff + ext <EOL> return None <EOL> def split_command_line ( command_line ) : <EOL> '''<STR_LIT>''' <EOL> arg_list = [ ] <EOL> arg = '<STR_LIT>' <EOL> state_basic = <NUM_LIT:0> <EOL> state_esc = <NUM_LIT:1> <EOL> state_singlequote = <NUM_LIT:2> <EOL> state_doublequote = <NUM_LIT:3> <EOL> state_whitespace = <NUM_LIT:4> <EOL> state = state_basic <EOL> for c in command_line : <EOL> if state == state_basic or state == state_whitespace : <EOL> if c == '<STR_LIT:\\>' : <EOL> state = state_esc <EOL> elif c == r"<STR_LIT:'>" : <EOL> state = state_singlequote <EOL> elif c == r'<STR_LIT:">' : <EOL> state = state_doublequote <EOL> elif c . isspace ( ) : <EOL> if state == state_whitespace : <EOL> None <EOL> else : <EOL> arg_list . append ( arg ) <EOL> arg = '<STR_LIT>' <EOL> state = state_whitespace <EOL> else : <EOL> arg = arg + c <EOL> state = state_basic <EOL> elif state == state_esc : <EOL> arg = arg + c <EOL> state = state_basic <EOL> elif state == state_singlequote : <EOL> if c == r"<STR_LIT:'>" : <EOL> state = state_basic <EOL> else : <EOL> arg = arg + c <EOL> elif state == state_doublequote : <EOL> if c == r'<STR_LIT:">' : <EOL> state = state_basic <EOL> else : <EOL> arg = arg + c <EOL> if arg != '<STR_LIT>' : <EOL> arg_list . append ( arg ) <EOL> return arg_list </s>
