<s> import telnetlib <EOL> import time <EOL> import re <EOL> import inspect <EOL> from robot . version import get_version <EOL> from robot import utils <EOL> class Telnet : <EOL> """<STR_LIT>""" <EOL> ROBOT_LIBRARY_SCOPE = '<STR_LIT>' <EOL> ROBOT_LIBRARY_VERSION = get_version ( ) <EOL> def __init__ ( self , timeout = <NUM_LIT> , newline = '<STR_LIT>' , prompt = None , prompt_is_regexp = False ) : <EOL> """<STR_LIT>""" <EOL> self . _timeout = timeout == '<STR_LIT>' and <NUM_LIT> or timeout <EOL> self . _newline = newline == '<STR_LIT>' and '<STR_LIT>' or newline <EOL> self . _prompt = ( prompt , prompt_is_regexp ) <EOL> self . _cache = utils . ConnectionCache ( ) <EOL> self . _conn = None <EOL> self . _conn_kws = self . _lib_kws = None <EOL> def get_keyword_names ( self ) : <EOL> return self . _get_library_keywords ( ) + self . _get_connection_keywords ( ) <EOL> def _get_library_keywords ( self ) : <EOL> if self . _lib_kws is None : <EOL> self . _lib_kws = [ name for name in dir ( self ) <EOL> if not name . startswith ( '<STR_LIT:_>' ) and name != '<STR_LIT>' <EOL> and inspect . ismethod ( getattr ( self , name ) ) ] <EOL> return self . _lib_kws <EOL> def _get_connection_keywords ( self ) : <EOL> if self . _conn_kws is None : <EOL> conn = self . _get_connection ( ) <EOL> excluded = [ name for name in dir ( telnetlib . Telnet ( ) ) <EOL> if name not in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> self . _conn_kws = [ name for name in dir ( conn ) <EOL> if not name . startswith ( '<STR_LIT:_>' ) and name not in excluded <EOL> and inspect . ismethod ( getattr ( conn , name ) ) ] <EOL> return self . _conn_kws <EOL> def __getattr__ ( self , name ) : <EOL> if name not in self . _get_connection_keywords ( ) : <EOL> raise AttributeError ( name ) <EOL> conn = self . _conn is None and self . _get_connection ( ) or self . _conn <EOL> return getattr ( conn , name ) <EOL> def open_connection ( self , host , alias = None , port = <NUM_LIT> , timeout = None , <EOL> newline = None , prompt = None , prompt_is_regexp = False ) : <EOL> """<STR_LIT>""" <EOL> if timeout is None or timeout == '<STR_LIT>' : <EOL> timeout = self . _timeout <EOL> if newline is None : <EOL> newline = self . _newline <EOL> if prompt is None : <EOL> prompt , prompt_is_regexp = self . _prompt <EOL> print '<STR_LIT>' % ( host , port , self . _prompt ) <EOL> self . _conn = self . _get_connection ( host , port , timeout , newline , <EOL> prompt , prompt_is_regexp ) <EOL> return self . _cache . register ( self . _conn , alias ) <EOL> def _get_connection ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> return TelnetConnection ( * args ) <EOL> def switch_connection ( self , index_or_alias ) : <EOL> """<STR_LIT>""" <EOL> old_index = self . _cache . current_index <EOL> self . _conn = self . _cache . switch ( index_or_alias ) <EOL> return old_index <EOL> def close_all_connections ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _conn = self . _cache . close_all ( ) <EOL> class TelnetConnection ( telnetlib . Telnet ) : <EOL> def __init__ ( self , host = None , port = <NUM_LIT> , timeout = <NUM_LIT> , newline = '<STR_LIT>' , <EOL> prompt = None , prompt_is_regexp = False ) : <EOL> port = port == '<STR_LIT>' and <NUM_LIT> or int ( port ) <EOL> telnetlib . Telnet . __init__ ( self , host , port ) <EOL> self . set_timeout ( timeout ) <EOL> self . set_newline ( newline ) <EOL> self . set_prompt ( prompt , prompt_is_regexp ) <EOL> self . _default_log_level = '<STR_LIT>' <EOL> self . set_option_negotiation_callback ( self . _negotiate_echo_on ) <EOL> def set_timeout ( self , timeout ) : <EOL> """<STR_LIT>""" <EOL> old = getattr ( self , '<STR_LIT>' , <NUM_LIT> ) <EOL> self . _timeout = utils . timestr_to_secs ( timeout ) <EOL> return utils . secs_to_timestr ( old ) <EOL> def set_newline ( self , newline ) : <EOL> """<STR_LIT>""" <EOL> old = getattr ( self , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _newline = newline . upper ( ) . replace ( '<STR_LIT>' , '<STR_LIT:\n>' ) . replace ( '<STR_LIT>' , '<STR_LIT:\r>' ) <EOL> return old <EOL> def close_connection ( self , loglevel = None ) : <EOL> """<STR_LIT>""" <EOL> telnetlib . Telnet . close ( self ) <EOL> ret = self . read_all ( ) . decode ( '<STR_LIT>' , '<STR_LIT:ignore>' ) <EOL> self . _log ( ret , loglevel ) <EOL> return ret <EOL> def login ( self , username , password , login_prompt = '<STR_LIT>' , <EOL> password_prompt = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> ret = self . read_until ( login_prompt , '<STR_LIT>' ) . decode ( '<STR_LIT>' , '<STR_LIT:ignore>' ) <EOL> self . write_bare ( username + self . _newline ) <EOL> ret += username + '<STR_LIT:\n>' <EOL> ret += self . read_until ( password_prompt , '<STR_LIT>' ) . decode ( '<STR_LIT>' , '<STR_LIT:ignore>' ) <EOL> self . write_bare ( password + self . _newline ) <EOL> ret += '<STR_LIT:*>' * len ( password ) + '<STR_LIT:\n>' <EOL> if self . _prompt_is_set ( ) : <EOL> try : <EOL> ret += self . read_until_prompt ( '<STR_LIT>' ) <EOL> except AssertionError : <EOL> self . _verify_login ( ret ) <EOL> raise <EOL> else : <EOL> ret += self . _verify_login ( ret ) <EOL> self . _log ( ret ) <EOL> return ret <EOL> def _verify_login ( self , ret ) : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> while True : <EOL> try : <EOL> ret += self . read_until ( '<STR_LIT:\n>' , '<STR_LIT>' ) . decode ( '<STR_LIT>' , '<STR_LIT:ignore>' ) <EOL> except AssertionError : <EOL> return ret <EOL> else : <EOL> if '<STR_LIT>' in ret : <EOL> self . _log ( ret ) <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> def write ( self , text , loglevel = None ) : <EOL> """<STR_LIT>""" <EOL> if self . _newline in text : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> text += self . _newline <EOL> self . write_bare ( text ) <EOL> return self . read_until ( self . _newline , loglevel ) <EOL> def write_bare ( self , text ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> text = str ( text ) <EOL> except UnicodeError : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' % text ) <EOL> telnetlib . Telnet . write ( self , text ) <EOL> def write_until_expected_output ( self , text , expected , timeout , <EOL> retry_interval , loglevel = None ) : <EOL> """<STR_LIT>""" <EOL> timeout = utils . timestr_to_secs ( timeout ) <EOL> retry_interval = utils . timestr_to_secs ( retry_interval ) <EOL> starttime = time . time ( ) <EOL> while time . time ( ) - starttime < timeout : <EOL> self . write_bare ( text ) <EOL> self . read_until ( text , loglevel ) <EOL> ret = telnetlib . Telnet . read_until ( self , expected , <EOL> retry_interval ) . decode ( '<STR_LIT>' , '<STR_LIT:ignore>' ) <EOL> self . _log ( ret , loglevel ) <EOL> if ret . endswith ( expected ) : <EOL> return ret <EOL> raise AssertionError ( "<STR_LIT>" <EOL> % ( expected , utils . secs_to_timestr ( timeout ) ) ) <EOL> def read ( self , loglevel = None ) : <EOL> """<STR_LIT>""" <EOL> ret = self . read_very_eager ( ) . decode ( '<STR_LIT>' , '<STR_LIT:ignore>' ) <EOL> self . _log ( ret , loglevel ) <EOL> return ret <EOL> def read_until ( self , expected , loglevel = None ) : <EOL> """<STR_LIT>""" <EOL> ret = telnetlib . Telnet . read_until ( self , expected , <EOL> self . _timeout ) . decode ( '<STR_LIT>' , '<STR_LIT:ignore>' ) <EOL> self . _log ( ret , loglevel ) <EOL> if not ret . endswith ( expected ) : <EOL> raise AssertionError ( "<STR_LIT>" <EOL> % ( expected , utils . secs_to_timestr ( self . _timeout ) ) ) <EOL> return ret <EOL> def read_until_regexp ( self , * expected ) : <EOL> """<STR_LIT>""" <EOL> expected = list ( expected ) <EOL> if self . _is_valid_log_level ( expected [ - <NUM_LIT:1> ] ) : <EOL> loglevel = expected [ - <NUM_LIT:1> ] <EOL> expected = expected [ : - <NUM_LIT:1> ] <EOL> else : <EOL> loglevel = '<STR_LIT>' <EOL> try : <EOL> index , _ , ret = self . expect ( expected , self . _timeout ) <EOL> except TypeError : <EOL> index , ret = - <NUM_LIT:1> , '<STR_LIT>' <EOL> ret = ret . decode ( '<STR_LIT>' , '<STR_LIT:ignore>' ) <EOL> self . _log ( ret , loglevel ) <EOL> if index == - <NUM_LIT:1> : <EOL> expected = [ exp if isinstance ( exp , basestring ) else exp . pattern <EOL> for exp in expected ] <EOL> raise AssertionError ( "<STR_LIT>" <EOL> % ( utils . seq2str ( expected , lastsep = '<STR_LIT>' ) , <EOL> utils . secs_to_timestr ( self . _timeout ) ) ) <EOL> return ret <EOL> def read_until_prompt ( self , loglevel = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . _prompt_is_set ( ) : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> prompt , regexp = self . _prompt <EOL> if regexp : <EOL> return self . read_until_regexp ( prompt , loglevel ) <EOL> return self . read_until ( prompt , loglevel ) <EOL> def execute_command ( self , command , loglevel = None ) : <EOL> """<STR_LIT>""" <EOL> self . write ( command , loglevel ) <EOL> return self . read_until_prompt ( loglevel ) <EOL> def set_prompt ( self , prompt , prompt_is_regexp = False ) : <EOL> """<STR_LIT>""" <EOL> old = hasattr ( self , '<STR_LIT>' ) and self . _prompt or ( None , False ) <EOL> if prompt_is_regexp : <EOL> self . _prompt = ( re . compile ( prompt ) , True ) <EOL> else : <EOL> self . _prompt = ( prompt , False ) <EOL> if old [ <NUM_LIT:1> ] : <EOL> return old [ <NUM_LIT:0> ] . pattern , True <EOL> return old <EOL> def _prompt_is_set ( self ) : <EOL> return self . _prompt [ <NUM_LIT:0> ] is not None <EOL> def set_default_log_level ( self , level ) : <EOL> """<STR_LIT>""" <EOL> self . _is_valid_log_level ( level , raise_if_invalid = True ) <EOL> old = self . _default_log_level <EOL> self . _default_log_level = level . upper ( ) <EOL> return old <EOL> def _log ( self , msg , level = None ) : <EOL> self . _is_valid_log_level ( level , raise_if_invalid = True ) <EOL> msg = msg . strip ( ) <EOL> if level is None : <EOL> level = self . _default_log_level <EOL> if msg != '<STR_LIT>' : <EOL> print '<STR_LIT>' % ( level . upper ( ) , msg ) <EOL> def _is_valid_log_level ( self , level , raise_if_invalid = False ) : <EOL> if level is None : <EOL> return True <EOL> if isinstance ( level , basestring ) and level . upper ( ) in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> return True <EOL> if not raise_if_invalid : <EOL> return False <EOL> raise AssertionError ( "<STR_LIT>" % level ) <EOL> def _negotiate_echo_on ( self , sock , cmd , opt ) : <EOL> if opt == telnetlib . ECHO and cmd in ( telnetlib . WILL , telnetlib . WONT ) : <EOL> self . sock . sendall ( telnetlib . IAC + telnetlib . DO + opt ) <EOL> elif opt != telnetlib . NOOPT : <EOL> if cmd in ( telnetlib . DO , telnetlib . DONT ) : <EOL> self . sock . sendall ( telnetlib . IAC + telnetlib . WONT + opt ) <EOL> elif cmd in ( telnetlib . WILL , telnetlib . WONT ) : <EOL> self . sock . sendall ( telnetlib . IAC + telnetlib . DONT + opt ) </s>
