<s> import getpass <EOL> import os <EOL> import pexpect <EOL> from functools import wraps <EOL> from deployer . console import Console <EOL> from deployer . exceptions import ExecCommandFailed <EOL> from . base import Host , Stat <EOL> __all__ = ( <EOL> '<STR_LIT>' , <EOL> ) <EOL> class LocalStat ( Stat ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> _localhost_password = None <EOL> _localhost_start_path = os . getcwd ( ) <EOL> class LocalHost ( Host ) : <EOL> """<STR_LIT>""" <EOL> slug = '<STR_LIT:localhost>' <EOL> address = '<STR_LIT:localhost>' <EOL> start_path = os . getcwd ( ) <EOL> def run ( self , * a , ** kw ) : <EOL> if kw . get ( '<STR_LIT>' , False ) : <EOL> self . _ensure_password_is_known ( ) <EOL> return Host . run ( self , * a , ** kw ) <EOL> def _expand_tilde ( self , relative_path ) : <EOL> return os . path . expanduser ( relative_path ) <EOL> def expand_path ( self , path ) : <EOL> return os . path . expanduser ( path ) <EOL> def _ensure_password_is_known ( self ) : <EOL> global _localhost_password <EOL> tries = <NUM_LIT:0> <EOL> while _localhost_password is None : <EOL> _localhost_password = Console ( self . pty ) . input ( '<STR_LIT>' % <EOL> ( self . username , self . slug ) , is_password = True ) <EOL> try : <EOL> Host . _run_silent_sudo ( self , '<STR_LIT>' ) <EOL> except ExecCommandFailed : <EOL> print '<STR_LIT>' <EOL> self . _backend . password = None <EOL> tries += <NUM_LIT:1> <EOL> if tries >= <NUM_LIT:3> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> @ property <EOL> def password ( self ) : <EOL> return _localhost_password <EOL> @ property <EOL> def username ( self ) : <EOL> return getpass . getuser ( ) <EOL> def get_start_path ( self ) : <EOL> return _localhost_start_path <EOL> def get_ip_address ( self , interface = '<STR_LIT>' ) : <EOL> return '<STR_LIT:127.0.0.1>' <EOL> def _get_session ( self ) : <EOL> """<STR_LIT>""" <EOL> class channel ( object ) : <EOL> def __init__ ( self ) : <EOL> self . _spawn = None <EOL> self . _height , self . _width = None , None <EOL> class spawn_override ( pexpect . spawn ) : <EOL> def setwinsize ( s , rows = None , cols = None ) : <EOL> w = self . _width or cols <EOL> h = self . _height or rows <EOL> if w and h : <EOL> pexpect . spawn . setwinsize ( s , h , w ) <EOL> self . spawn_override = spawn_override <EOL> def get_pty ( self , term = None , width = None , height = None ) : <EOL> self . resize_pty ( width = width , height = height ) <EOL> def recv ( self , count = <NUM_LIT> ) : <EOL> try : <EOL> return self . _spawn . read_nonblocking ( count ) <EOL> except pexpect . EOF : <EOL> return '<STR_LIT>' <EOL> def read ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _spawn . read ( ) <EOL> def send ( self , data ) : <EOL> return self . _spawn . write ( data ) <EOL> def fileno ( self ) : <EOL> return self . _spawn . child_fd <EOL> def resize_pty ( self , width = None , height = None ) : <EOL> self . _width , self . _height = width , height <EOL> if self . _spawn and self . _height and self . _width : <EOL> self . _spawn . setwinsize ( self . _height , self . _width ) <EOL> def exec_command ( self , command ) : <EOL> self . _spawn = self . spawn_override ( '<STR_LIT>' , [ '<STR_LIT:-c>' , command ] ) <EOL> if self . _spawn and self . _height and self . _width : <EOL> self . _spawn . setwinsize ( self . _height , self . _width ) <EOL> def recv_exit_status ( self ) : <EOL> self . _spawn . close ( ) <EOL> return self . _spawn . exitstatus <EOL> def settimeout ( self , * a , ** kw ) : <EOL> pass <EOL> @ property <EOL> def status_event ( self ) : <EOL> class event ( object ) : <EOL> def isSet ( self ) : <EOL> return False <EOL> return event ( ) <EOL> return channel ( ) <EOL> def _open ( self , remote_path , mode ) : <EOL> return open ( remote_path , mode ) <EOL> @ wraps ( Host . stat ) <EOL> def stat ( self , path ) : <EOL> try : <EOL> full_path = os . path . join ( self . getcwd ( ) , path ) <EOL> filename = os . path . split ( full_path ) [ <NUM_LIT:1> ] <EOL> return LocalStat ( os . stat ( full_path ) , filename ) <EOL> except OSError as e : <EOL> raise IOError ( e . message ) <EOL> @ wraps ( Host . listdir ) <EOL> def listdir ( self , path = '<STR_LIT:.>' ) : <EOL> return os . listdir ( os . path . join ( * [ self . getcwd ( ) , path ] ) ) <EOL> @ wraps ( Host . listdir_stat ) <EOL> def listdir_stat ( self , path = '<STR_LIT:.>' ) : <EOL> return [ self . stat ( f ) for f in self . listdir ( ) ] <EOL> def _read_non_interactive ( self , chan ) : <EOL> return chan . read ( ) <EOL> def start_interactive_shell ( self , command = None , initial_input = None ) : <EOL> """<STR_LIT>""" <EOL> self . run ( command = '<STR_LIT>' , initial_input = initial_input ) </s>
