back to Reference (Gold) summary
Reference (Gold): pexpect
Pytest Summary for test tests
| status | count |
|---|---|
| passed | 245 |
| failed | 12 |
| total | 257 |
| collected | 257 |
Failed pytests:
test_ctrl_chars.py::TestCtrlChars::test_control_chars
test_ctrl_chars.py::TestCtrlChars::test_control_chars
self =def test_control_chars(self): '''This tests that we can send all 256 8-bit characters to a child process.''' child = pexpect.spawn(self.getch_cmd, echo=False, timeout=5) child.expect('READY') for i in range(1, 256): child.send(byte(i)) > child.expect ('%d ' % (i,)) tests/test_ctrl_chars.py:51: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ pexpect/spawnbase.py:354: in expect return self.expect_list(compiled_pattern_list, pexpect/spawnbase.py:383: in expect_list return exp.expect_loop(timeout) pexpect/expect.py:181: in expect_loop return self.timeout(e) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = err = TIMEOUT('Timeout exceeded.') def timeout(self, err=None): spawn = self.spawn spawn.before = spawn._before.getvalue() spawn.after = TIMEOUT index = self.searcher.timeout_index if index >= 0: spawn.match = TIMEOUT spawn.match_index = index return index else: spawn.match = None spawn.match_index = None msg = str(spawn) msg += '\nsearcher: %s' % self.searcher if err is not None: msg = str(err) + '\n' + msg exc = TIMEOUT(msg) exc.__cause__ = None # in Python 3.x we can use "raise exc from None" > raise exc E pexpect.exceptions.TIMEOUT: Timeout exceeded. E E command: /testbed/.venv/bin/python3 E args: ['/testbed/.venv/bin/python3', 'getch.py'] E buffer (last 100 chars): b'\r\n' E before (last 100 chars): b'\r\n' E after: E match: None E match_index: None E exitstatus: None E flag_eof: False E pid: 33 E child_fd: 13 E closed: False E timeout: 5 E delimiter: E logfile: None E logfile_read: None E logfile_send: None E maxread: 2000 E ignorecase: False E searchwindowsize: None E delaybeforesend: 0.05 E delayafterclose: 0.1 E delayafterterminate: 0.1 E searcher: searcher_re: E 0: re.compile(b'26 ') pexpect/expect.py:144: TIMEOUT
test_ctrl_chars.py::TestCtrlChars::test_sendcontrol
test_ctrl_chars.py::TestCtrlChars::test_sendcontrol
self =def test_sendcontrol(self): '''This tests that we can send all special control codes by name. ''' child = pexpect.spawn(self.getch_cmd, echo=False, timeout=5) child.expect('READY') for ctrl in 'abcdefghijklmnopqrstuvwxyz': assert child.sendcontrol(ctrl) == 1 val = ord(ctrl) - ord('a') + 1 > child.expect_exact(str(val)+' ') tests/test_ctrl_chars.py:99: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ pexpect/spawnbase.py:432: in expect_exact return exp.expect_loop(timeout) pexpect/expect.py:181: in expect_loop return self.timeout(e) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = err = TIMEOUT('Timeout exceeded.') def timeout(self, err=None): spawn = self.spawn spawn.before = spawn._before.getvalue() spawn.after = TIMEOUT index = self.searcher.timeout_index if index >= 0: spawn.match = TIMEOUT spawn.match_index = index return index else: spawn.match = None spawn.match_index = None msg = str(spawn) msg += '\nsearcher: %s' % self.searcher if err is not None: msg = str(err) + '\n' + msg exc = TIMEOUT(msg) exc.__cause__ = None # in Python 3.x we can use "raise exc from None" > raise exc E pexpect.exceptions.TIMEOUT: Timeout exceeded. E E command: /testbed/.venv/bin/python3 E args: ['/testbed/.venv/bin/python3', 'getch.py'] E buffer (last 100 chars): b'\r\n' E before (last 100 chars): b'\r\n' E after: E match: None E match_index: None E exitstatus: None E flag_eof: False E pid: 34 E child_fd: 14 E closed: False E timeout: 5 E delimiter: E logfile: None E logfile_read: None E logfile_send: None E maxread: 2000 E ignorecase: False E searchwindowsize: None E delaybeforesend: 0.05 E delayafterclose: 0.1 E delayafterterminate: 0.1 E searcher: searcher_string: E 0: b'26 ' pexpect/expect.py:144: TIMEOUT
test_expect.py::ExpectTestCase::test_expect
test_expect.py::ExpectTestCase::test_expect
self =def test_expect (self): the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'], stdout=subprocess.PIPE).communicate()[0].rstrip() p = pexpect.spawn('ls -l /bin') the_new_way = b'' while 1: i = p.expect ([b'\n', pexpect.EOF]) the_new_way = the_new_way + p.before if i == 1: break the_new_way = the_new_way.rstrip() the_new_way = the_new_way.replace(b'\r\n', b'\n' ).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip() the_old_way = the_old_way.replace(b'\r\n', b'\n' ).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip() > assert the_old_way == the_new_way, hex_diff(the_old_way, the_new_way) tests/test_expect.py:429: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ tests/test_expect.py:53: in hex_diff hex_dump(left).splitlines(), hex_dump(right).splitlines()) tests/test_expect.py:46: in hex_dump hexa = ' '.join(["%02X"%ord(x) for x in s]) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .0 = > hexa = ' '.join(["%02X"%ord(x) for x in s]) E TypeError: ord() expected string of length 1, but int found tests/test_expect.py:46: TypeError
test_expect.py::ExpectTestCase::test_expect_eof
test_expect.py::ExpectTestCase::test_expect_eof
self =def test_expect_eof (self): the_old_way = subprocess.Popen(args=['/bin/ls', '-l', '/bin'], stdout=subprocess.PIPE).communicate()[0].rstrip() p = pexpect.spawn('/bin/ls -l /bin') p.expect(pexpect.EOF) # This basically tells it to read everything. Same as pexpect.run() function. the_new_way = p.before the_new_way = the_new_way.replace(b'\r\n', b'\n' ).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip() the_old_way = the_old_way.replace(b'\r\n', b'\n' ).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip() > assert the_old_way == the_new_way, hex_diff(the_old_way, the_new_way) tests/test_expect.py:461: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ tests/test_expect.py:53: in hex_diff hex_dump(left).splitlines(), hex_dump(right).splitlines()) tests/test_expect.py:46: in hex_dump hexa = ' '.join(["%02X"%ord(x) for x in s]) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .0 = > hexa = ' '.join(["%02X"%ord(x) for x in s]) E TypeError: ord() expected string of length 1, but int found tests/test_expect.py:46: TypeError
test_expect.py::ExpectTestCase::test_expect_exact
test_expect.py::ExpectTestCase::test_expect_exact
self =def test_expect_exact (self): the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'], stdout=subprocess.PIPE).communicate()[0].rstrip() p = pexpect.spawn('ls -l /bin') the_new_way = b'' while 1: i = p.expect_exact ([b'\n', pexpect.EOF]) the_new_way = the_new_way + p.before if i == 1: break the_new_way = the_new_way.replace(b'\r\n', b'\n' ).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip() the_old_way = the_old_way.replace(b'\r\n', b'\n' ).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip() > assert the_old_way == the_new_way, hex_diff(the_old_way, the_new_way) tests/test_expect.py:445: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ tests/test_expect.py:53: in hex_diff hex_dump(left).splitlines(), hex_dump(right).splitlines()) tests/test_expect.py:46: in hex_dump hexa = ' '.join(["%02X"%ord(x) for x in s]) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .0 = > hexa = ' '.join(["%02X"%ord(x) for x in s]) E TypeError: ord() expected string of length 1, but int found tests/test_expect.py:46: TypeError
test_popen_spawn.py::ExpectTestCase::test_expect
test_popen_spawn.py::ExpectTestCase::test_expect
self =def test_expect(self): the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'], stdout=subprocess.PIPE).communicate()[0].rstrip() p = PopenSpawn('ls -l /bin') the_new_way = b'' while 1: i = p.expect([b'\n', pexpect.EOF]) the_new_way = the_new_way + p.before if i == 1: break the_new_way += b'\n' the_new_way = the_new_way.rstrip() > assert the_old_way == the_new_way, len(the_old_way) - len(the_new_way) E AssertionError: -30 E assert b'lrwxrwxrwx ...in -> usr/bin' == b'ls: /bin: B...in -> usr/bin' E E At index 1 diff: b'r' != b's' E Use -v to get more diff tests/test_popen_spawn.py:66: AssertionError
test_popen_spawn.py::ExpectTestCase::test_expect_eof
test_popen_spawn.py::ExpectTestCase::test_expect_eof
self =def test_expect_eof(self): the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'], stdout=subprocess.PIPE).communicate()[0].rstrip() p = PopenSpawn('ls -l /bin') # This basically tells it to read everything. Same as pexpect.run() # function. p.expect(pexpect.EOF) the_new_way = p.before.rstrip() > assert the_old_way == the_new_way, len(the_old_way) - len(the_new_way) E AssertionError: -30 E assert b'lrwxrwxrwx ...in -> usr/bin' == b'ls: /bin: B...in -> usr/bin' E E At index 1 diff: b'r' != b's' E Use -v to get more diff tests/test_popen_spawn.py:95: AssertionError
test_popen_spawn.py::ExpectTestCase::test_expect_exact
test_popen_spawn.py::ExpectTestCase::test_expect_exact
self =def test_expect_exact(self): the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'], stdout=subprocess.PIPE).communicate()[0].rstrip() p = PopenSpawn('ls -l /bin') the_new_way = b'' while 1: i = p.expect_exact([b'\n', pexpect.EOF]) the_new_way = the_new_way + p.before if i == 1: break the_new_way += b'\n' the_new_way = the_new_way.rstrip() > assert the_old_way == the_new_way, len(the_old_way) - len(the_new_way) E AssertionError: -30 E assert b'lrwxrwxrwx ...in -> usr/bin' == b'ls: /bin: B...in -> usr/bin' E E At index 1 diff: b'r' != b's' E Use -v to get more diff tests/test_popen_spawn.py:81: AssertionError
test_replwrap.py::REPLWrapTestCase::test_existing_spawn
test_replwrap.py::REPLWrapTestCase::test_existing_spawn
self =def test_existing_spawn(self): child = pexpect.spawn("bash", timeout=5, encoding='utf-8') repl = replwrap.REPLWrapper(child, re.compile('[$#]'), "PS1='{0}' PS2='{1}' " "PROMPT_COMMAND=''") print(repl) res = repl.run_command("echo $HOME") print(res) > assert res.startswith('/'), res E AssertionError: [?2004l /root E [?2004h E assert False E + where False = ('/') E + where = '\x1b[?2004l\r/root\r\n\x1b[?2004h'.startswith /testbed/tests/test_replwrap.py:98: AssertionError
test_replwrap.py::REPLWrapTestCase::test_pager_as_cat
test_replwrap.py::REPLWrapTestCase::test_pager_as_cat
self =def test_pager_as_cat(self): " PAGER is set to cat, to prevent timeout in ``man sleep``. " bash = replwrap.bash() res = bash.run_command('man sleep', timeout=5) > assert 'SLEEP' in res.upper(), res E AssertionError: This system has been minimized by removing packages and content that are E not required on a system that users do not log into. E E To restore this content, including manpages, you can run the 'unminimize' E command. You will still need to ensure the 'man-db' package is installed. E E assert 'SLEEP' in "THIS SYSTEM HAS BEEN MINIMIZED BY REMOVING PACKAGES AND CONTENT THAT ARE\r\nNOT REQUIRED ON A SYSTEM THAT USERS DO NO...ANPAGES, YOU CAN RUN THE 'UNMINIMIZE'\r\nCOMMAND. YOU WILL STILL NEED TO ENSURE THE 'MAN-DB' PACKAGE IS INSTALLED.\r\n" E + where "THIS SYSTEM HAS BEEN MINIMIZED BY REMOVING PACKAGES AND CONTENT THAT ARE\r\nNOT REQUIRED ON A SYSTEM THAT USERS DO NO...ANPAGES, YOU CAN RUN THE 'UNMINIMIZE'\r\nCOMMAND. YOU WILL STILL NEED TO ENSURE THE 'MAN-DB' PACKAGE IS INSTALLED.\r\n" = () E + where = "This system has been minimized by removing packages and content that are\r\nnot required on a system that users do no...anpages, you can run the 'unminimize'\r\ncommand. You will still need to ensure the 'man-db' package is installed.\r\n".upper /testbed/tests/test_replwrap.py:42: AssertionError
test_replwrap.py::REPLWrapTestCase::test_zsh
test_replwrap.py::REPLWrapTestCase::test_zsh
self =def test_zsh(self): > zsh = replwrap.zsh() /testbed/tests/test_replwrap.py:101: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ /testbed/pexpect/replwrap.py:136: in zsh return _repl_sh(command, list(args), non_printable_insert='%(!..)') /testbed/pexpect/replwrap.py:116: in _repl_sh child = pexpect.spawn(command, args, echo=False, encoding='utf-8') /testbed/pexpect/pty_spawn.py:205: in __init__ self._spawn(command, args, preexec_fn, dimensions) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = , command = 'zsh' args = ['--no-rcs', '-V', '+Z'], preexec_fn = None, dimensions = None def _spawn(self, command, args=[], preexec_fn=None, dimensions=None): '''This starts the given command in a child process. This does all the fork/exec type of stuff for a pty. This is called by __init__. If args is empty then command will be parsed (split on spaces) and args will be set to parsed arguments. ''' # The pid and child_fd of this object get set by this method. # Note that it is difficult for this method to fail. # You cannot detect if the child process cannot start. # So the only way you can tell if the child process started # or not is to try to read from the file descriptor. If you get # EOF immediately then it means that the child is already dead. # That may not necessarily be bad because you may have spawned a child # that performs some task; creates no stdout output; and then dies. # If command is an int type then it may represent a file descriptor. if isinstance(command, type(0)): raise ExceptionPexpect('Command is an int type. ' + 'If this is a file descriptor then maybe you want to ' + 'use fdpexpect.fdspawn which takes an existing ' + 'file descriptor instead of a command string.') if not isinstance(args, type([])): raise TypeError('The argument, args, must be a list.') if args == []: self.args = split_command_line(command) self.command = self.args[0] else: # Make a shallow copy of the args list. self.args = args[:] self.args.insert(0, command) self.command = command command_with_path = which(self.command, env=self.env) if command_with_path is None: > raise ExceptionPexpect('The command was not found or was not ' + 'executable: %s.' % self.command) E pexpect.exceptions.ExceptionPexpect: The command was not found or was not executable: zsh. /testbed/pexpect/pty_spawn.py:276: ExceptionPexpect
test_winsize.py::TestCaseWinsize::test_setwinsize
test_winsize.py::TestCaseWinsize::test_setwinsize
self =def test_setwinsize(self): """ Ensure method .setwinsize() sends signal caught by child. """ p = pexpect.spawn('{self.PYTHONBIN} sigwinch_report.py' .format(self=self), timeout=3) # Note that we must await the installation of the child process' # signal handler, p.expect_exact('READY') p.setwinsize(19, 84) > p.expect_exact('SIGWINCH: (19, 84)') /testbed/tests/test_winsize.py:52: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ /testbed/pexpect/spawnbase.py:432: in expect_exact return exp.expect_loop(timeout) /testbed/pexpect/expect.py:181: in expect_loop return self.timeout(e) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = err = TIMEOUT('Timeout exceeded.') def timeout(self, err=None): spawn = self.spawn spawn.before = spawn._before.getvalue() spawn.after = TIMEOUT index = self.searcher.timeout_index if index >= 0: spawn.match = TIMEOUT spawn.match_index = index return index else: spawn.match = None spawn.match_index = None msg = str(spawn) msg += '\nsearcher: %s' % self.searcher if err is not None: msg = str(err) + '\n' + msg exc = TIMEOUT(msg) exc.__cause__ = None # in Python 3.x we can use "raise exc from None" > raise exc E pexpect.exceptions.TIMEOUT: Timeout exceeded. E E command: /testbed/.venv/bin/python3 E args: ['/testbed/.venv/bin/python3', 'sigwinch_report.py'] E buffer (last 100 chars): b'\r\n' E before (last 100 chars): b'\r\n' E after: E match: None E match_index: None E exitstatus: None E flag_eof: False E pid: 355 E child_fd: 21 E closed: False E timeout: 3 E delimiter: E logfile: None E logfile_read: None E logfile_send: None E maxread: 2000 E ignorecase: False E searchwindowsize: None E delaybeforesend: 0.05 E delayafterclose: 0.1 E delayafterterminate: 0.1 E searcher: searcher_string: E 0: b'SIGWINCH: (19, 84)' /testbed/pexpect/expect.py:144: TIMEOUT
Patch diff
diff --git a/pexpect/ANSI.py b/pexpect/ANSI.py
index df126d4..1cd2e90 100644
--- a/pexpect/ANSI.py
+++ b/pexpect/ANSI.py
@@ -1,4 +1,4 @@
-"""This implements an ANSI (VT100) terminal emulator as a subclass of screen.
+'''This implements an ANSI (VT100) terminal emulator as a subclass of screen.
PEXPECT LICENSE
@@ -17,127 +17,335 @@ PEXPECT LICENSE
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-"""
+'''
+
+# references:
+# http://en.wikipedia.org/wiki/ANSI_escape_code
+# http://www.retards.org/terminals/vt102.html
+# http://vt100.net/docs/vt102-ug/contents.html
+# http://vt100.net/docs/vt220-rm/
+# http://www.termsys.demon.co.uk/vtansi.htm
+
from . import screen
from . import FSM
import string
+#
+# The 'Do.*' functions are helper functions for the ANSI class.
+#
+def DoEmit (fsm):
+
+ screen = fsm.memory[0]
+ screen.write_ch(fsm.input_symbol)
+
+def DoStartNumber (fsm):
+
+ fsm.memory.append (fsm.input_symbol)
+
+def DoBuildNumber (fsm):
+
+ ns = fsm.memory.pop()
+ ns = ns + fsm.input_symbol
+ fsm.memory.append (ns)
+
+def DoBackOne (fsm):
+
+ screen = fsm.memory[0]
+ screen.cursor_back ()
+
+def DoBack (fsm):
+
+ count = int(fsm.memory.pop())
+ screen = fsm.memory[0]
+ screen.cursor_back (count)
+
+def DoDownOne (fsm):
+
+ screen = fsm.memory[0]
+ screen.cursor_down ()
+
+def DoDown (fsm):
+
+ count = int(fsm.memory.pop())
+ screen = fsm.memory[0]
+ screen.cursor_down (count)
+
+def DoForwardOne (fsm):
+
+ screen = fsm.memory[0]
+ screen.cursor_forward ()
+
+def DoForward (fsm):
+
+ count = int(fsm.memory.pop())
+ screen = fsm.memory[0]
+ screen.cursor_forward (count)
+
+def DoUpReverse (fsm):
+
+ screen = fsm.memory[0]
+ screen.cursor_up_reverse()
+
+def DoUpOne (fsm):
+
+ screen = fsm.memory[0]
+ screen.cursor_up ()
+
+def DoUp (fsm):
+
+ count = int(fsm.memory.pop())
+ screen = fsm.memory[0]
+ screen.cursor_up (count)
+
+def DoHome (fsm):
+
+ c = int(fsm.memory.pop())
+ r = int(fsm.memory.pop())
+ screen = fsm.memory[0]
+ screen.cursor_home (r,c)
+
+def DoHomeOrigin (fsm):
+
+ c = 1
+ r = 1
+ screen = fsm.memory[0]
+ screen.cursor_home (r,c)
+
+def DoEraseDown (fsm):
+
+ screen = fsm.memory[0]
+ screen.erase_down()
+
+def DoErase (fsm):
+
+ arg = int(fsm.memory.pop())
+ screen = fsm.memory[0]
+ if arg == 0:
+ screen.erase_down()
+ elif arg == 1:
+ screen.erase_up()
+ elif arg == 2:
+ screen.erase_screen()
+
+def DoEraseEndOfLine (fsm):
-class term(screen.screen):
- """This class is an abstract, generic terminal.
+ screen = fsm.memory[0]
+ screen.erase_end_of_line()
+
+def DoEraseLine (fsm):
+
+ arg = int(fsm.memory.pop())
+ screen = fsm.memory[0]
+ if arg == 0:
+ screen.erase_end_of_line()
+ elif arg == 1:
+ screen.erase_start_of_line()
+ elif arg == 2:
+ screen.erase_line()
+
+def DoEnableScroll (fsm):
+
+ screen = fsm.memory[0]
+ screen.scroll_screen()
+
+def DoCursorSave (fsm):
+
+ screen = fsm.memory[0]
+ screen.cursor_save_attrs()
+
+def DoCursorRestore (fsm):
+
+ screen = fsm.memory[0]
+ screen.cursor_restore_attrs()
+
+def DoScrollRegion (fsm):
+
+ screen = fsm.memory[0]
+ r2 = int(fsm.memory.pop())
+ r1 = int(fsm.memory.pop())
+ screen.scroll_screen_rows (r1,r2)
+
+def DoMode (fsm):
+
+ screen = fsm.memory[0]
+ mode = fsm.memory.pop() # Should be 4
+ # screen.setReplaceMode ()
+
+def DoLog (fsm):
+
+ screen = fsm.memory[0]
+ fsm.memory = [screen]
+ fout = open ('log', 'a')
+ fout.write (fsm.input_symbol + ',' + fsm.current_state + '\n')
+ fout.close()
+
+class term (screen.screen):
+
+ '''This class is an abstract, generic terminal.
This does nothing. This is a placeholder that
provides a common base class for other terminals
- such as an ANSI terminal. """
+ such as an ANSI terminal. '''
- def __init__(self, r=24, c=80, *args, **kwargs):
- screen.screen.__init__(self, r, c, *args, **kwargs)
+ def __init__ (self, r=24, c=80, *args, **kwargs):
+ screen.screen.__init__(self, r,c,*args,**kwargs)
-class ANSI(term):
- """This class implements an ANSI (VT100) terminal.
+class ANSI (term):
+ '''This class implements an ANSI (VT100) terminal.
It is a stream filter that recognizes ANSI terminal
- escape sequences and maintains the state of a screen object. """
-
- def __init__(self, r=24, c=80, *args, **kwargs):
- term.__init__(self, r, c, *args, **kwargs)
- self.state = FSM.FSM('INIT', [self])
- self.state.set_default_transition(DoLog, 'INIT')
- self.state.add_transition_any('INIT', DoEmit, 'INIT')
- self.state.add_transition('\x1b', 'INIT', None, 'ESC')
- self.state.add_transition_any('ESC', DoLog, 'INIT')
- self.state.add_transition('(', 'ESC', None, 'G0SCS')
- self.state.add_transition(')', 'ESC', None, 'G1SCS')
- self.state.add_transition_list('AB012', 'G0SCS', None, 'INIT')
- self.state.add_transition_list('AB012', 'G1SCS', None, 'INIT')
- self.state.add_transition('7', 'ESC', DoCursorSave, 'INIT')
- self.state.add_transition('8', 'ESC', DoCursorRestore, 'INIT')
- self.state.add_transition('M', 'ESC', DoUpReverse, 'INIT')
- self.state.add_transition('>', 'ESC', DoUpReverse, 'INIT')
- self.state.add_transition('<', 'ESC', DoUpReverse, 'INIT')
- self.state.add_transition('=', 'ESC', None, 'INIT')
- self.state.add_transition('#', 'ESC', None, 'GRAPHICS_POUND')
- self.state.add_transition_any('GRAPHICS_POUND', None, 'INIT')
- self.state.add_transition('[', 'ESC', None, 'ELB')
- self.state.add_transition('H', 'ELB', DoHomeOrigin, 'INIT')
- self.state.add_transition('D', 'ELB', DoBackOne, 'INIT')
- self.state.add_transition('B', 'ELB', DoDownOne, 'INIT')
- self.state.add_transition('C', 'ELB', DoForwardOne, 'INIT')
- self.state.add_transition('A', 'ELB', DoUpOne, 'INIT')
- self.state.add_transition('J', 'ELB', DoEraseDown, 'INIT')
- self.state.add_transition('K', 'ELB', DoEraseEndOfLine, 'INIT')
- self.state.add_transition('r', 'ELB', DoEnableScroll, 'INIT')
- self.state.add_transition('m', 'ELB', self.do_sgr, 'INIT')
- self.state.add_transition('?', 'ELB', None, 'MODECRAP')
- self.state.add_transition_list(string.digits, 'ELB', DoStartNumber,
- 'NUMBER_1')
- self.state.add_transition_list(string.digits, 'NUMBER_1',
- DoBuildNumber, 'NUMBER_1')
- self.state.add_transition('D', 'NUMBER_1', DoBack, 'INIT')
- self.state.add_transition('B', 'NUMBER_1', DoDown, 'INIT')
- self.state.add_transition('C', 'NUMBER_1', DoForward, 'INIT')
- self.state.add_transition('A', 'NUMBER_1', DoUp, 'INIT')
- self.state.add_transition('J', 'NUMBER_1', DoErase, 'INIT')
- self.state.add_transition('K', 'NUMBER_1', DoEraseLine, 'INIT')
- self.state.add_transition('l', 'NUMBER_1', DoMode, 'INIT')
- self.state.add_transition('m', 'NUMBER_1', self.do_sgr, 'INIT')
- self.state.add_transition('q', 'NUMBER_1', self.do_decsca, 'INIT')
- self.state.add_transition_list(string.digits, 'MODECRAP',
- DoStartNumber, 'MODECRAP_NUM')
- self.state.add_transition_list(string.digits, 'MODECRAP_NUM',
- DoBuildNumber, 'MODECRAP_NUM')
- self.state.add_transition('l', 'MODECRAP_NUM', self.do_modecrap, 'INIT'
- )
- self.state.add_transition('h', 'MODECRAP_NUM', self.do_modecrap, 'INIT'
- )
- self.state.add_transition(';', 'NUMBER_1', None, 'SEMICOLON')
- self.state.add_transition_any('SEMICOLON', DoLog, 'INIT')
- self.state.add_transition_list(string.digits, 'SEMICOLON',
- DoStartNumber, 'NUMBER_2')
- self.state.add_transition_list(string.digits, 'NUMBER_2',
- DoBuildNumber, 'NUMBER_2')
- self.state.add_transition_any('NUMBER_2', DoLog, 'INIT')
- self.state.add_transition('H', 'NUMBER_2', DoHome, 'INIT')
- self.state.add_transition('f', 'NUMBER_2', DoHome, 'INIT')
- self.state.add_transition('r', 'NUMBER_2', DoScrollRegion, 'INIT')
- self.state.add_transition('m', 'NUMBER_2', self.do_sgr, 'INIT')
- self.state.add_transition('q', 'NUMBER_2', self.do_decsca, 'INIT')
- self.state.add_transition(';', 'NUMBER_2', None, 'SEMICOLON_X')
- self.state.add_transition_any('SEMICOLON_X', DoLog, 'INIT')
- self.state.add_transition_list(string.digits, 'SEMICOLON_X',
- DoStartNumber, 'NUMBER_X')
- self.state.add_transition_list(string.digits, 'NUMBER_X',
- DoBuildNumber, 'NUMBER_X')
- self.state.add_transition_any('NUMBER_X', DoLog, 'INIT')
- self.state.add_transition('m', 'NUMBER_X', self.do_sgr, 'INIT')
- self.state.add_transition('q', 'NUMBER_X', self.do_decsca, 'INIT')
- self.state.add_transition(';', 'NUMBER_X', None, 'SEMICOLON_X')
-
- def process(self, c):
+ escape sequences and maintains the state of a screen object. '''
+
+ def __init__ (self, r=24,c=80,*args,**kwargs):
+
+ term.__init__(self,r,c,*args,**kwargs)
+
+ #self.screen = screen (24,80)
+ self.state = FSM.FSM ('INIT',[self])
+ self.state.set_default_transition (DoLog, 'INIT')
+ self.state.add_transition_any ('INIT', DoEmit, 'INIT')
+ self.state.add_transition ('\x1b', 'INIT', None, 'ESC')
+ self.state.add_transition_any ('ESC', DoLog, 'INIT')
+ self.state.add_transition ('(', 'ESC', None, 'G0SCS')
+ self.state.add_transition (')', 'ESC', None, 'G1SCS')
+ self.state.add_transition_list ('AB012', 'G0SCS', None, 'INIT')
+ self.state.add_transition_list ('AB012', 'G1SCS', None, 'INIT')
+ self.state.add_transition ('7', 'ESC', DoCursorSave, 'INIT')
+ self.state.add_transition ('8', 'ESC', DoCursorRestore, 'INIT')
+ self.state.add_transition ('M', 'ESC', DoUpReverse, 'INIT')
+ self.state.add_transition ('>', 'ESC', DoUpReverse, 'INIT')
+ self.state.add_transition ('<', 'ESC', DoUpReverse, 'INIT')
+ self.state.add_transition ('=', 'ESC', None, 'INIT') # Selects application keypad.
+ self.state.add_transition ('#', 'ESC', None, 'GRAPHICS_POUND')
+ self.state.add_transition_any ('GRAPHICS_POUND', None, 'INIT')
+ self.state.add_transition ('[', 'ESC', None, 'ELB')
+ # ELB means Escape Left Bracket. That is ^[[
+ self.state.add_transition ('H', 'ELB', DoHomeOrigin, 'INIT')
+ self.state.add_transition ('D', 'ELB', DoBackOne, 'INIT')
+ self.state.add_transition ('B', 'ELB', DoDownOne, 'INIT')
+ self.state.add_transition ('C', 'ELB', DoForwardOne, 'INIT')
+ self.state.add_transition ('A', 'ELB', DoUpOne, 'INIT')
+ self.state.add_transition ('J', 'ELB', DoEraseDown, 'INIT')
+ self.state.add_transition ('K', 'ELB', DoEraseEndOfLine, 'INIT')
+ self.state.add_transition ('r', 'ELB', DoEnableScroll, 'INIT')
+ self.state.add_transition ('m', 'ELB', self.do_sgr, 'INIT')
+ self.state.add_transition ('?', 'ELB', None, 'MODECRAP')
+ self.state.add_transition_list (string.digits, 'ELB', DoStartNumber, 'NUMBER_1')
+ self.state.add_transition_list (string.digits, 'NUMBER_1', DoBuildNumber, 'NUMBER_1')
+ self.state.add_transition ('D', 'NUMBER_1', DoBack, 'INIT')
+ self.state.add_transition ('B', 'NUMBER_1', DoDown, 'INIT')
+ self.state.add_transition ('C', 'NUMBER_1', DoForward, 'INIT')
+ self.state.add_transition ('A', 'NUMBER_1', DoUp, 'INIT')
+ self.state.add_transition ('J', 'NUMBER_1', DoErase, 'INIT')
+ self.state.add_transition ('K', 'NUMBER_1', DoEraseLine, 'INIT')
+ self.state.add_transition ('l', 'NUMBER_1', DoMode, 'INIT')
+ ### It gets worse... the 'm' code can have infinite number of
+ ### number;number;number before it. I've never seen more than two,
+ ### but the specs say it's allowed. crap!
+ self.state.add_transition ('m', 'NUMBER_1', self.do_sgr, 'INIT')
+ ### LED control. Same implementation problem as 'm' code.
+ self.state.add_transition ('q', 'NUMBER_1', self.do_decsca, 'INIT')
+
+ # \E[?47h switch to alternate screen
+ # \E[?47l restores to normal screen from alternate screen.
+ self.state.add_transition_list (string.digits, 'MODECRAP', DoStartNumber, 'MODECRAP_NUM')
+ self.state.add_transition_list (string.digits, 'MODECRAP_NUM', DoBuildNumber, 'MODECRAP_NUM')
+ self.state.add_transition ('l', 'MODECRAP_NUM', self.do_modecrap, 'INIT')
+ self.state.add_transition ('h', 'MODECRAP_NUM', self.do_modecrap, 'INIT')
+
+#RM Reset Mode Esc [ Ps l none
+ self.state.add_transition (';', 'NUMBER_1', None, 'SEMICOLON')
+ self.state.add_transition_any ('SEMICOLON', DoLog, 'INIT')
+ self.state.add_transition_list (string.digits, 'SEMICOLON', DoStartNumber, 'NUMBER_2')
+ self.state.add_transition_list (string.digits, 'NUMBER_2', DoBuildNumber, 'NUMBER_2')
+ self.state.add_transition_any ('NUMBER_2', DoLog, 'INIT')
+ self.state.add_transition ('H', 'NUMBER_2', DoHome, 'INIT')
+ self.state.add_transition ('f', 'NUMBER_2', DoHome, 'INIT')
+ self.state.add_transition ('r', 'NUMBER_2', DoScrollRegion, 'INIT')
+ ### It gets worse... the 'm' code can have infinite number of
+ ### number;number;number before it. I've never seen more than two,
+ ### but the specs say it's allowed. crap!
+ self.state.add_transition ('m', 'NUMBER_2', self.do_sgr, 'INIT')
+ ### LED control. Same problem as 'm' code.
+ self.state.add_transition ('q', 'NUMBER_2', self.do_decsca, 'INIT')
+ self.state.add_transition (';', 'NUMBER_2', None, 'SEMICOLON_X')
+
+ # Create a state for 'q' and 'm' which allows an infinite number of ignored numbers
+ self.state.add_transition_any ('SEMICOLON_X', DoLog, 'INIT')
+ self.state.add_transition_list (string.digits, 'SEMICOLON_X', DoStartNumber, 'NUMBER_X')
+ self.state.add_transition_list (string.digits, 'NUMBER_X', DoBuildNumber, 'NUMBER_X')
+ self.state.add_transition_any ('NUMBER_X', DoLog, 'INIT')
+ self.state.add_transition ('m', 'NUMBER_X', self.do_sgr, 'INIT')
+ self.state.add_transition ('q', 'NUMBER_X', self.do_decsca, 'INIT')
+ self.state.add_transition (';', 'NUMBER_X', None, 'SEMICOLON_X')
+
+ def process (self, c):
"""Process a single character. Called by :meth:`write`."""
- pass
+ if isinstance(c, bytes):
+ c = self._decode(c)
+ self.state.process(c)
+
+ def process_list (self, l):
+
+ self.write(l)
- def write(self, s):
+ def write (self, s):
"""Process text, writing it to the virtual screen while handling
ANSI escape codes.
"""
+ if isinstance(s, bytes):
+ s = self._decode(s)
+ for c in s:
+ self.process(c)
+
+ def flush (self):
pass
- def write_ch(self, ch):
- """This puts a character at the current cursor position. The cursor
+ def write_ch (self, ch):
+ '''This puts a character at the current cursor position. The cursor
position is moved forward with wrap-around, but no scrolling is done if
- the cursor hits the lower-right corner of the screen. """
- pass
+ the cursor hits the lower-right corner of the screen. '''
- def do_sgr(self, fsm):
- """Select Graphic Rendition, e.g. color. """
- pass
+ if isinstance(ch, bytes):
+ ch = self._decode(ch)
- def do_decsca(self, fsm):
- """Select character protection attribute. """
- pass
+ #\r and \n both produce a call to cr() and lf(), respectively.
+ ch = ch[0]
- def do_modecrap(self, fsm):
- """Handler for [?<number>h and [?<number>l. If anyone
+ if ch == u'\r':
+ self.cr()
+ return
+ if ch == u'\n':
+ self.crlf()
+ return
+ if ch == chr(screen.BS):
+ self.cursor_back()
+ return
+ self.put_abs(self.cur_r, self.cur_c, ch)
+ old_r = self.cur_r
+ old_c = self.cur_c
+ self.cursor_forward()
+ if old_c == self.cur_c:
+ self.cursor_down()
+ if old_r != self.cur_r:
+ self.cursor_home (self.cur_r, 1)
+ else:
+ self.scroll_up ()
+ self.cursor_home (self.cur_r, 1)
+ self.erase_line()
+
+ def do_sgr (self, fsm):
+ '''Select Graphic Rendition, e.g. color. '''
+ screen = fsm.memory[0]
+ fsm.memory = [screen]
+
+ def do_decsca (self, fsm):
+ '''Select character protection attribute. '''
+ screen = fsm.memory[0]
+ fsm.memory = [screen]
+
+ def do_modecrap (self, fsm):
+ '''Handler for \x1b[?<number>h and \x1b[?<number>l. If anyone
wanted to actually use these, they'd need to add more states to the
- FSM rather than just improve or override this method. """
- pass
+ FSM rather than just improve or override this method. '''
+ screen = fsm.memory[0]
+ fsm.memory = [screen]
diff --git a/pexpect/FSM.py b/pexpect/FSM.py
index 5fc4095..46b392e 100644
--- a/pexpect/FSM.py
+++ b/pexpect/FSM.py
@@ -1,4 +1,6 @@
-"""This module implements a Finite State Machine (FSM). In addition to state
+#!/usr/bin/env python
+
+'''This module implements a Finite State Machine (FSM). In addition to state
this FSM also maintains a user defined "memory". So this FSM can be used as a
Push-down Automata (PDA) since a PDA is a FSM + memory.
@@ -80,11 +82,11 @@ PEXPECT LICENSE
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-"""
-
+'''
class ExceptionFSM(Exception):
- """This is the FSM Exception class."""
+
+ '''This is the FSM Exception class.'''
def __init__(self, value):
self.value = value
@@ -92,19 +94,24 @@ class ExceptionFSM(Exception):
def __str__(self):
return 'ExceptionFSM: ' + str(self.value)
-
class FSM:
- """This is a Finite State Machine (FSM).
- """
+
+ '''This is a Finite State Machine (FSM).
+ '''
def __init__(self, initial_state, memory=None):
- """This creates the FSM. You set the initial state here. The "memory"
+
+ '''This creates the FSM. You set the initial state here. The "memory"
attribute is any object that you want to pass along to the action
functions. It is not used by the FSM. For parsing you would typically
- pass a list to be used as a stack. """
+ pass a list to be used as a stack. '''
+
+ # Map (input_symbol, current_state) --> (action, next_state).
self.state_transitions = {}
+ # Map (current_state) --> (action, next_state).
self.state_transitions_any = {}
self.default_transition = None
+
self.input_symbol = None
self.initial_state = initial_state
self.current_state = self.initial_state
@@ -112,15 +119,18 @@ class FSM:
self.action = None
self.memory = memory
- def reset(self):
- """This sets the current_state to the initial_state and sets
+ def reset (self):
+
+ '''This sets the current_state to the initial_state and sets
input_symbol to None. The initial state was set by the constructor
- __init__(). """
- pass
+ __init__(). '''
- def add_transition(self, input_symbol, state, action=None, next_state=None
- ):
- """This adds a transition that associates:
+ self.current_state = self.initial_state
+ self.input_symbol = None
+
+ def add_transition (self, input_symbol, state, action=None, next_state=None):
+
+ '''This adds a transition that associates:
(input_symbol, current_state) --> (action, next_state)
@@ -129,23 +139,31 @@ class FSM:
set to None in which case the current state will be unchanged.
You can also set transitions for a list of symbols by using
- add_transition_list(). """
- pass
+ add_transition_list(). '''
+
+ if next_state is None:
+ next_state = state
+ self.state_transitions[(input_symbol, state)] = (action, next_state)
- def add_transition_list(self, list_input_symbols, state, action=None,
- next_state=None):
- """This adds the same transition for a list of input symbols.
+ def add_transition_list (self, list_input_symbols, state, action=None, next_state=None):
+
+ '''This adds the same transition for a list of input symbols.
You can pass a list or a string. Note that it is handy to use
string.digits, string.whitespace, string.letters, etc. to add
transitions that match character classes.
The action may be set to None in which case the process() method will
ignore the action and only set the next_state. The next_state may be
- set to None in which case the current state will be unchanged. """
- pass
+ set to None in which case the current state will be unchanged. '''
+
+ if next_state is None:
+ next_state = state
+ for input_symbol in list_input_symbols:
+ self.add_transition (input_symbol, state, action, next_state)
- def add_transition_any(self, state, action=None, next_state=None):
- """This adds a transition that associates:
+ def add_transition_any (self, state, action=None, next_state=None):
+
+ '''This adds a transition that associates:
(current_state) --> (action, next_state)
@@ -155,22 +173,28 @@ class FSM:
The action may be set to None in which case the process() method will
ignore the action and only set the next_state. The next_state may be
- set to None in which case the current state will be unchanged. """
- pass
+ set to None in which case the current state will be unchanged. '''
+
+ if next_state is None:
+ next_state = state
+ self.state_transitions_any [state] = (action, next_state)
- def set_default_transition(self, action, next_state):
- """This sets the default transition. This defines an action and
+ def set_default_transition (self, action, next_state):
+
+ '''This sets the default transition. This defines an action and
next_state if the FSM cannot find the input symbol and the current
state in the transition list and if the FSM cannot find the
current_state in the transition_any list. This is useful as a final
fall-through state for catching errors and undefined states.
The default transition can be removed by setting the attribute
- default_transition to None. """
- pass
+ default_transition to None. '''
+
+ self.default_transition = (action, next_state)
+
+ def get_transition (self, input_symbol, state):
- def get_transition(self, input_symbol, state):
- """This returns (action, next state) given an input_symbol and state.
+ '''This returns (action, next state) given an input_symbol and state.
This does not modify the FSM state, so calling this method has no side
effects. Normally you do not call this method directly. It is called by
process().
@@ -189,35 +213,121 @@ class FSM:
This is a handler for errors, undefined states, or defaults.
4. No transition was defined. If we get here then raise an exception.
- """
- pass
+ '''
+
+ if (input_symbol, state) in self.state_transitions:
+ return self.state_transitions[(input_symbol, state)]
+ elif state in self.state_transitions_any:
+ return self.state_transitions_any[state]
+ elif self.default_transition is not None:
+ return self.default_transition
+ else:
+ raise ExceptionFSM ('Transition is undefined: (%s, %s).' %
+ (str(input_symbol), str(state)) )
+
+ def process (self, input_symbol):
- def process(self, input_symbol):
- """This is the main method that you call to process input. This may
+ '''This is the main method that you call to process input. This may
cause the FSM to change state and call an action. This method calls
get_transition() to find the action and next_state associated with the
input_symbol and current_state. If the action is None then the action
is not called and only the current state is changed. This method
processes one complete input symbol. You can process a list of symbols
- (or a string) by calling process_list(). """
- pass
+ (or a string) by calling process_list(). '''
+
+ self.input_symbol = input_symbol
+ (self.action, self.next_state) = self.get_transition (self.input_symbol, self.current_state)
+ if self.action is not None:
+ self.action (self)
+ self.current_state = self.next_state
+ self.next_state = None
+
+ def process_list (self, input_symbols):
+
+ '''This takes a list and sends each element to process(). The list may
+ be a string or any iterable object. '''
- def process_list(self, input_symbols):
- """This takes a list and sends each element to process(). The list may
- be a string or any iterable object. """
- pass
+ for s in input_symbols:
+ self.process (s)
+##############################################################################
+# The following is an example that demonstrates the use of the FSM class to
+# process an RPN expression. Run this module from the command line. You will
+# get a prompt > for input. Enter an RPN Expression. Numbers may be integers.
+# Operators are * / + - Use the = sign to evaluate and print the expression.
+# For example:
+#
+# 167 3 2 2 * * * 1 - =
+#
+# will print:
+#
+# 2003
+##############################################################################
import sys
import string
-PY3 = sys.version_info[0] >= 3
+PY3 = (sys.version_info[0] >= 3)
+
+#
+# These define the actions.
+# Note that "memory" is a list being used as a stack.
+#
+
+def BeginBuildNumber (fsm):
+ fsm.memory.append (fsm.input_symbol)
+
+def BuildNumber (fsm):
+ s = fsm.memory.pop ()
+ s = s + fsm.input_symbol
+ fsm.memory.append (s)
+
+def EndBuildNumber (fsm):
+ s = fsm.memory.pop ()
+ fsm.memory.append (int(s))
+
+def DoOperator (fsm):
+ ar = fsm.memory.pop()
+ al = fsm.memory.pop()
+ if fsm.input_symbol == '+':
+ fsm.memory.append (al + ar)
+ elif fsm.input_symbol == '-':
+ fsm.memory.append (al - ar)
+ elif fsm.input_symbol == '*':
+ fsm.memory.append (al * ar)
+ elif fsm.input_symbol == '/':
+ fsm.memory.append (al / ar)
+
+def DoEqual (fsm):
+ print(str(fsm.memory.pop()))
+
+def Error (fsm):
+ print('That does not compute.')
+ print(str(fsm.input_symbol))
def main():
- """This is where the example starts and the FSM state transitions are
+
+ '''This is where the example starts and the FSM state transitions are
defined. Note that states are strings (such as 'INIT'). This is not
- necessary, but it makes the example easier to read. """
- pass
+ necessary, but it makes the example easier to read. '''
+
+ f = FSM ('INIT', [])
+ f.set_default_transition (Error, 'INIT')
+ f.add_transition_any ('INIT', None, 'INIT')
+ f.add_transition ('=', 'INIT', DoEqual, 'INIT')
+ f.add_transition_list (string.digits, 'INIT', BeginBuildNumber, 'BUILDING_NUMBER')
+ f.add_transition_list (string.digits, 'BUILDING_NUMBER', BuildNumber, 'BUILDING_NUMBER')
+ f.add_transition_list (string.whitespace, 'BUILDING_NUMBER', EndBuildNumber, 'INIT')
+ f.add_transition_list ('+-*/', 'INIT', DoOperator, 'INIT')
+
+ print()
+ print('Enter an RPN Expression.')
+ print('Numbers may be integers. Operators are * / + -')
+ print('Use the = sign to evaluate and print the expression.')
+ print('For example: ')
+ print(' 167 3 2 2 * * * 1 - =')
+ inputstr = (input if PY3 else raw_input)('> ') # analysis:ignore
+ f.process_list(inputstr)
if __name__ == '__main__':
diff --git a/pexpect/_async.py b/pexpect/_async.py
index 41e86a0..261720c 100644
--- a/pexpect/_async.py
+++ b/pexpect/_async.py
@@ -9,8 +9,20 @@ For Python versions later than 3.6, coroutines and objects that are defined via
Here the code is just imported, to provide the same interface to older code.
"""
+# pylint: disable=unused-import
+# flake8: noqa: F401
from sys import version_info as py_version_info
+
+# this assumes async def/await are more stable
if py_version_info >= (3, 6):
- from pexpect._async_w_await import PatternWaiter, expect_async, repl_run_command_async
+ from pexpect._async_w_await import (
+ PatternWaiter,
+ expect_async,
+ repl_run_command_async,
+ )
else:
- from pexpect._async_pre_await import PatternWaiter, expect_async, repl_run_command_async
+ from pexpect._async_pre_await import (
+ PatternWaiter,
+ expect_async,
+ repl_run_command_async,
+ )
diff --git a/pexpect/_async_pre_await.py b/pexpect/_async_pre_await.py
index 17921cf..81ece1b 100644
--- a/pexpect/_async_pre_await.py
+++ b/pexpect/_async_pre_await.py
@@ -5,8 +5,107 @@
import asyncio
import errno
import signal
+
from pexpect import EOF
+@asyncio.coroutine
+def expect_async(expecter, timeout=None):
+ # First process data that was previously read - if it maches, we don't need
+ # async stuff.
+ idx = expecter.existing_data()
+ if idx is not None:
+ return idx
+ if not expecter.spawn.async_pw_transport:
+ pw = PatternWaiter()
+ pw.set_expecter(expecter)
+ transport, pw = yield from asyncio.get_event_loop().connect_read_pipe(
+ lambda: pw, expecter.spawn
+ )
+ expecter.spawn.async_pw_transport = pw, transport
+ else:
+ pw, transport = expecter.spawn.async_pw_transport
+ pw.set_expecter(expecter)
+ transport.resume_reading()
+ try:
+ return (yield from asyncio.wait_for(pw.fut, timeout))
+ except asyncio.TimeoutError as e:
+ transport.pause_reading()
+ return expecter.timeout(e)
+
+
+@asyncio.coroutine
+def repl_run_command_async(repl, cmdlines, timeout=-1):
+ res = []
+ repl.child.sendline(cmdlines[0])
+ for line in cmdlines[1:]:
+ yield from repl._expect_prompt(timeout=timeout, async_=True)
+ res.append(repl.child.before)
+ repl.child.sendline(line)
+
+ # Command was fully submitted, now wait for the next prompt
+ prompt_idx = yield from repl._expect_prompt(timeout=timeout, async_=True)
+ if prompt_idx == 1:
+ # We got the continuation prompt - command was incomplete
+ repl.child.kill(signal.SIGINT)
+ yield from repl._expect_prompt(timeout=1, async_=True)
+ raise ValueError("Continuation prompt found - input was incomplete:")
+ return "".join(res + [repl.child.before])
+
+
class PatternWaiter(asyncio.Protocol):
transport = None
+
+ def set_expecter(self, expecter):
+ self.expecter = expecter
+ self.fut = asyncio.Future()
+
+ def found(self, result):
+ if not self.fut.done():
+ self.fut.set_result(result)
+ self.transport.pause_reading()
+
+ def error(self, exc):
+ if not self.fut.done():
+ self.fut.set_exception(exc)
+ self.transport.pause_reading()
+
+ def connection_made(self, transport):
+ self.transport = transport
+
+ def data_received(self, data):
+ spawn = self.expecter.spawn
+ s = spawn._decoder.decode(data)
+ spawn._log(s, "read")
+
+ if self.fut.done():
+ spawn._before.write(s)
+ spawn._buffer.write(s)
+ return
+
+ try:
+ index = self.expecter.new_data(s)
+ if index is not None:
+ # Found a match
+ self.found(index)
+ except Exception as e:
+ self.expecter.errored()
+ self.error(e)
+
+ def eof_received(self):
+ # N.B. If this gets called, async will close the pipe (the spawn object)
+ # for us
+ try:
+ self.expecter.spawn.flag_eof = True
+ index = self.expecter.eof()
+ except EOF as e:
+ self.error(e)
+ else:
+ self.found(index)
+
+ def connection_lost(self, exc):
+ if isinstance(exc, OSError) and exc.errno == errno.EIO:
+ # We may get here without eof_received being called, e.g on Linux
+ self.eof_received()
+ elif exc is not None:
+ self.error(exc)
diff --git a/pexpect/_async_w_await.py b/pexpect/_async_w_await.py
index fda878f..59cb1ef 100644
--- a/pexpect/_async_w_await.py
+++ b/pexpect/_async_w_await.py
@@ -7,12 +7,112 @@ import asyncio
import errno
import signal
from sys import version_info as py_version_info
+
from pexpect import EOF
+
if py_version_info >= (3, 7):
+ # get_running_loop, new in 3.7, is preferred to get_event_loop
_loop_getter = asyncio.get_running_loop
else:
+ # Deprecation warning since 3.10
_loop_getter = asyncio.get_event_loop
+async def expect_async(expecter, timeout=None):
+ # First process data that was previously read - if it maches, we don't need
+ # async stuff.
+ idx = expecter.existing_data()
+ if idx is not None:
+ return idx
+ if not expecter.spawn.async_pw_transport:
+ pattern_waiter = PatternWaiter()
+ pattern_waiter.set_expecter(expecter)
+ transport, pattern_waiter = await _loop_getter().connect_read_pipe(
+ lambda: pattern_waiter, expecter.spawn
+ )
+ expecter.spawn.async_pw_transport = pattern_waiter, transport
+ else:
+ pattern_waiter, transport = expecter.spawn.async_pw_transport
+ pattern_waiter.set_expecter(expecter)
+ transport.resume_reading()
+ try:
+ return await asyncio.wait_for(pattern_waiter.fut, timeout)
+ except asyncio.TimeoutError as exc:
+ transport.pause_reading()
+ return expecter.timeout(exc)
+
+
+async def repl_run_command_async(repl, cmdlines, timeout=-1):
+ res = []
+ repl.child.sendline(cmdlines[0])
+ for line in cmdlines[1:]:
+ await repl._expect_prompt(timeout=timeout, async_=True)
+ res.append(repl.child.before)
+ repl.child.sendline(line)
+
+ # Command was fully submitted, now wait for the next prompt
+ prompt_idx = await repl._expect_prompt(timeout=timeout, async_=True)
+ if prompt_idx == 1:
+ # We got the continuation prompt - command was incomplete
+ repl.child.kill(signal.SIGINT)
+ await repl._expect_prompt(timeout=1, async_=True)
+ raise ValueError("Continuation prompt found - input was incomplete:")
+ return "".join(res + [repl.child.before])
+
+
class PatternWaiter(asyncio.Protocol):
transport = None
+
+ def set_expecter(self, expecter):
+ self.expecter = expecter
+ self.fut = asyncio.Future()
+
+ def found(self, result):
+ if not self.fut.done():
+ self.fut.set_result(result)
+ self.transport.pause_reading()
+
+ def error(self, exc):
+ if not self.fut.done():
+ self.fut.set_exception(exc)
+ self.transport.pause_reading()
+
+ def connection_made(self, transport):
+ self.transport = transport
+
+ def data_received(self, data):
+ spawn = self.expecter.spawn
+ s = spawn._decoder.decode(data)
+ spawn._log(s, "read")
+
+ if self.fut.done():
+ spawn._before.write(s)
+ spawn._buffer.write(s)
+ return
+
+ try:
+ index = self.expecter.new_data(s)
+ if index is not None:
+ # Found a match
+ self.found(index)
+ except Exception as exc:
+ self.expecter.errored()
+ self.error(exc)
+
+ def eof_received(self):
+ # N.B. If this gets called, async will close the pipe (the spawn object)
+ # for us
+ try:
+ self.expecter.spawn.flag_eof = True
+ index = self.expecter.eof()
+ except EOF as exc:
+ self.error(exc)
+ else:
+ self.found(index)
+
+ def connection_lost(self, exc):
+ if isinstance(exc, OSError) and exc.errno == errno.EIO:
+ # We may get here without eof_received being called, e.g on Linux
+ self.eof_received()
+ elif exc is not None:
+ self.error(exc)
diff --git a/pexpect/exceptions.py b/pexpect/exceptions.py
index c66fe77..cb360f0 100644
--- a/pexpect/exceptions.py
+++ b/pexpect/exceptions.py
@@ -1,11 +1,11 @@
"""Exception classes used by Pexpect"""
+
import traceback
import sys
-
class ExceptionPexpect(Exception):
- """Base class for all exceptions raised by this module.
- """
+ '''Base class for all exceptions raised by this module.
+ '''
def __init__(self, value):
super(ExceptionPexpect, self).__init__(value)
@@ -15,16 +15,21 @@ class ExceptionPexpect(Exception):
return str(self.value)
def get_trace(self):
- """This returns an abbreviated stack trace with lines that only concern
+ '''This returns an abbreviated stack trace with lines that only concern
the caller. In other words, the stack trace inside the Pexpect module
- is not included. """
- pass
+ is not included. '''
+
+ tblist = traceback.extract_tb(sys.exc_info()[2])
+ tblist = [item for item in tblist if ('pexpect/__init__' not in item[0])
+ and ('pexpect/expect' not in item[0])]
+ tblist = traceback.format_list(tblist)
+ return ''.join(tblist)
class EOF(ExceptionPexpect):
- """Raised when EOF is read from a child.
- This usually means the child has exited."""
+ '''Raised when EOF is read from a child.
+ This usually means the child has exited.'''
class TIMEOUT(ExceptionPexpect):
- """Raised when a read time exceeds the timeout. """
+ '''Raised when a read time exceeds the timeout. '''
diff --git a/pexpect/expect.py b/pexpect/expect.py
index dabbe22..d3409db 100644
--- a/pexpect/expect.py
+++ b/pexpect/expect.py
@@ -1,12 +1,13 @@
import time
-from .exceptions import EOF, TIMEOUT
+from .exceptions import EOF, TIMEOUT
class Expecter(object):
-
def __init__(self, spawn, searcher, searchwindowsize=-1):
self.spawn = spawn
self.searcher = searcher
+ # A value of -1 means to use the figure from spawn, which should
+ # be None or a positive number.
if searchwindowsize == -1:
searchwindowsize = spawn.searchwindowsize
self.searchwindowsize = searchwindowsize
@@ -14,13 +15,177 @@ class Expecter(object):
if hasattr(searcher, 'longest_string'):
self.lookback = searcher.longest_string
+ def do_search(self, window, freshlen):
+ spawn = self.spawn
+ searcher = self.searcher
+ if freshlen > len(window):
+ freshlen = len(window)
+ index = searcher.search(window, freshlen, self.searchwindowsize)
+ if index >= 0:
+ spawn._buffer = spawn.buffer_type()
+ spawn._buffer.write(window[searcher.end:])
+ spawn.before = spawn._before.getvalue()[
+ 0:-(len(window) - searcher.start)]
+ spawn._before = spawn.buffer_type()
+ spawn._before.write(window[searcher.end:])
+ spawn.after = window[searcher.start:searcher.end]
+ spawn.match = searcher.match
+ spawn.match_index = index
+ # Found a match
+ return index
+ elif self.searchwindowsize or self.lookback:
+ maintain = self.searchwindowsize or self.lookback
+ if spawn._buffer.tell() > maintain:
+ spawn._buffer = spawn.buffer_type()
+ spawn._buffer.write(window[-maintain:])
+
+ def existing_data(self):
+ # First call from a new call to expect_loop or expect_async.
+ # self.searchwindowsize may have changed.
+ # Treat all data as fresh.
+ spawn = self.spawn
+ before_len = spawn._before.tell()
+ buf_len = spawn._buffer.tell()
+ freshlen = before_len
+ if before_len > buf_len:
+ if not self.searchwindowsize:
+ spawn._buffer = spawn.buffer_type()
+ window = spawn._before.getvalue()
+ spawn._buffer.write(window)
+ elif buf_len < self.searchwindowsize:
+ spawn._buffer = spawn.buffer_type()
+ spawn._before.seek(
+ max(0, before_len - self.searchwindowsize))
+ window = spawn._before.read()
+ spawn._buffer.write(window)
+ else:
+ spawn._buffer.seek(max(0, buf_len - self.searchwindowsize))
+ window = spawn._buffer.read()
+ else:
+ if self.searchwindowsize:
+ spawn._buffer.seek(max(0, buf_len - self.searchwindowsize))
+ window = spawn._buffer.read()
+ else:
+ window = spawn._buffer.getvalue()
+ return self.do_search(window, freshlen)
+
+ def new_data(self, data):
+ # A subsequent call, after a call to existing_data.
+ spawn = self.spawn
+ freshlen = len(data)
+ spawn._before.write(data)
+ if not self.searchwindowsize:
+ if self.lookback:
+ # search lookback + new data.
+ old_len = spawn._buffer.tell()
+ spawn._buffer.write(data)
+ spawn._buffer.seek(max(0, old_len - self.lookback))
+ window = spawn._buffer.read()
+ else:
+ # copy the whole buffer (really slow for large datasets).
+ spawn._buffer.write(data)
+ window = spawn.buffer
+ else:
+ if len(data) >= self.searchwindowsize or not spawn._buffer.tell():
+ window = data[-self.searchwindowsize:]
+ spawn._buffer = spawn.buffer_type()
+ spawn._buffer.write(window[-self.searchwindowsize:])
+ else:
+ spawn._buffer.write(data)
+ new_len = spawn._buffer.tell()
+ spawn._buffer.seek(max(0, new_len - self.searchwindowsize))
+ window = spawn._buffer.read()
+ return self.do_search(window, freshlen)
+
+ def eof(self, err=None):
+ spawn = self.spawn
+
+ spawn.before = spawn._before.getvalue()
+ spawn._buffer = spawn.buffer_type()
+ spawn._before = spawn.buffer_type()
+ spawn.after = EOF
+ index = self.searcher.eof_index
+ if index >= 0:
+ spawn.match = EOF
+ spawn.match_index = index
+ return index
+ else:
+ spawn.match = None
+ spawn.match_index = None
+ msg = str(spawn)
+ msg += '\nsearcher: %s' % self.searcher
+ if err is not None:
+ msg = str(err) + '\n' + msg
+
+ exc = EOF(msg)
+ exc.__cause__ = None # in Python 3.x we can use "raise exc from None"
+ raise exc
+
+ def timeout(self, err=None):
+ spawn = self.spawn
+
+ spawn.before = spawn._before.getvalue()
+ spawn.after = TIMEOUT
+ index = self.searcher.timeout_index
+ if index >= 0:
+ spawn.match = TIMEOUT
+ spawn.match_index = index
+ return index
+ else:
+ spawn.match = None
+ spawn.match_index = None
+ msg = str(spawn)
+ msg += '\nsearcher: %s' % self.searcher
+ if err is not None:
+ msg = str(err) + '\n' + msg
+
+ exc = TIMEOUT(msg)
+ exc.__cause__ = None # in Python 3.x we can use "raise exc from None"
+ raise exc
+
+ def errored(self):
+ spawn = self.spawn
+ spawn.before = spawn._before.getvalue()
+ spawn.after = None
+ spawn.match = None
+ spawn.match_index = None
+
def expect_loop(self, timeout=-1):
"""Blocking expect"""
- pass
+ spawn = self.spawn
+
+ if timeout is not None:
+ end_time = time.time() + timeout
+
+ try:
+ idx = self.existing_data()
+ if idx is not None:
+ return idx
+ while True:
+ # No match at this point
+ if (timeout is not None) and (timeout < 0):
+ return self.timeout()
+ # Still have time left, so read more data
+ incoming = spawn.read_nonblocking(spawn.maxread, timeout)
+ if self.spawn.delayafterread is not None:
+ time.sleep(self.spawn.delayafterread)
+ idx = self.new_data(incoming)
+ # Keep reading until exception or return.
+ if idx is not None:
+ return idx
+ if timeout is not None:
+ timeout = end_time - time.time()
+ except EOF as e:
+ return self.eof(e)
+ except TIMEOUT as e:
+ return self.timeout(e)
+ except:
+ self.errored()
+ raise
class searcher_string(object):
- """This is a plain string search helper for the spawn.expect_any() method.
+ '''This is a plain string search helper for the spawn.expect_any() method.
This helper class is for speed. For more powerful regex patterns
see the helper class, searcher_re.
@@ -36,11 +201,12 @@ class searcher_string(object):
end - index into the buffer, first byte after match
match - the matching string itself
- """
+ '''
def __init__(self, strings):
- """This creates an instance of searcher_string. This argument 'strings'
- may be a list; a sequence of strings; or the EOF or TIMEOUT types. """
+ '''This creates an instance of searcher_string. This argument 'strings'
+ may be a list; a sequence of strings; or the EOF or TIMEOUT types. '''
+
self.eof_index = -1
self.timeout_index = -1
self._strings = []
@@ -57,21 +223,22 @@ class searcher_string(object):
self.longest_string = len(s)
def __str__(self):
- """This returns a human-readable string that represents the state of
- the object."""
+ '''This returns a human-readable string that represents the state of
+ the object.'''
+
ss = [(ns[0], ' %d: %r' % ns) for ns in self._strings]
ss.append((-1, 'searcher_string:'))
if self.eof_index >= 0:
ss.append((self.eof_index, ' %d: EOF' % self.eof_index))
if self.timeout_index >= 0:
- ss.append((self.timeout_index, ' %d: TIMEOUT' % self.
- timeout_index))
+ ss.append((self.timeout_index,
+ ' %d: TIMEOUT' % self.timeout_index))
ss.sort()
ss = list(zip(*ss))[1]
return '\n'.join(ss)
def search(self, buffer, freshlen, searchwindowsize=None):
- """This searches 'buffer' for the first occurrence of one of the search
+ '''This searches 'buffer' for the first occurrence of one of the search
strings. 'freshlen' must indicate the number of bytes at the end of
'buffer' which have not been searched before. It helps to avoid
searching the same, possibly big, buffer over and over again.
@@ -79,12 +246,44 @@ class searcher_string(object):
See class spawn for the 'searchwindowsize' argument.
If there is a match this returns the index of that string, and sets
- 'start', 'end' and 'match'. Otherwise, this returns -1. """
- pass
+ 'start', 'end' and 'match'. Otherwise, this returns -1. '''
+
+ first_match = None
+
+ # 'freshlen' helps a lot here. Further optimizations could
+ # possibly include:
+ #
+ # using something like the Boyer-Moore Fast String Searching
+ # Algorithm; pre-compiling the search through a list of
+ # strings into something that can scan the input once to
+ # search for all N strings; realize that if we search for
+ # ['bar', 'baz'] and the input is '...foo' we need not bother
+ # rescanning until we've read three more bytes.
+ #
+ # Sadly, I don't know enough about this interesting topic. /grahn
+
+ for index, s in self._strings:
+ if searchwindowsize is None:
+ # the match, if any, can only be in the fresh data,
+ # or at the very end of the old data
+ offset = -(freshlen + len(s))
+ else:
+ # better obey searchwindowsize
+ offset = -searchwindowsize
+ n = buffer.find(s, offset)
+ if n >= 0 and (first_match is None or n < first_match):
+ first_match = n
+ best_index, best_match = index, s
+ if first_match is None:
+ return -1
+ self.match = best_match
+ self.start = first_match
+ self.end = self.start + len(self.match)
+ return best_index
class searcher_re(object):
- """This is regular expression string search helper for the
+ '''This is regular expression string search helper for the
spawn.expect_any() method. This helper class is for powerful
pattern matching. For speed, see the helper class, searcher_string.
@@ -100,12 +299,13 @@ class searcher_re(object):
end - index into the buffer, first byte after match
match - the re.match object returned by a successful re.search
- """
+ '''
def __init__(self, patterns):
- """This creates an instance that searches for 'patterns' Where
+ '''This creates an instance that searches for 'patterns' Where
'patterns' may be a list or other sequence of compiled regular
- expressions, or the EOF or TIMEOUT types."""
+ expressions, or the EOF or TIMEOUT types.'''
+
self.eof_index = -1
self.timeout_index = -1
self._searches = []
@@ -119,8 +319,11 @@ class searcher_re(object):
self._searches.append((n, s))
def __str__(self):
- """This returns a human-readable string that represents the state of
- the object."""
+ '''This returns a human-readable string that represents the state of
+ the object.'''
+
+ #ss = [(n, ' %d: re.compile("%s")' %
+ # (n, repr(s.pattern))) for n, s in self._searches]
ss = list()
for n, s in self._searches:
ss.append((n, ' %d: re.compile(%r)' % (n, s.pattern)))
@@ -128,19 +331,41 @@ class searcher_re(object):
if self.eof_index >= 0:
ss.append((self.eof_index, ' %d: EOF' % self.eof_index))
if self.timeout_index >= 0:
- ss.append((self.timeout_index, ' %d: TIMEOUT' % self.
- timeout_index))
+ ss.append((self.timeout_index, ' %d: TIMEOUT' %
+ self.timeout_index))
ss.sort()
ss = list(zip(*ss))[1]
return '\n'.join(ss)
def search(self, buffer, freshlen, searchwindowsize=None):
- """This searches 'buffer' for the first occurrence of one of the regular
+ '''This searches 'buffer' for the first occurrence of one of the regular
expressions. 'freshlen' must indicate the number of bytes at the end of
'buffer' which have not been searched before.
See class spawn for the 'searchwindowsize' argument.
If there is a match this returns the index of that string, and sets
- 'start', 'end' and 'match'. Otherwise, returns -1."""
- pass
+ 'start', 'end' and 'match'. Otherwise, returns -1.'''
+
+ first_match = None
+ # 'freshlen' doesn't help here -- we cannot predict the
+ # length of a match, and the re module provides no help.
+ if searchwindowsize is None:
+ searchstart = 0
+ else:
+ searchstart = max(0, len(buffer) - searchwindowsize)
+ for index, s in self._searches:
+ match = s.search(buffer, searchstart)
+ if match is None:
+ continue
+ n = match.start()
+ if first_match is None or n < first_match:
+ first_match = n
+ the_match = match
+ best_index = index
+ if first_match is None:
+ return -1
+ self.start = first_match
+ self.match = the_match
+ self.end = self.match.end()
+ return best_index
diff --git a/pexpect/fdpexpect.py b/pexpect/fdpexpect.py
index 4f4c4d9..140bdfe 100644
--- a/pexpect/fdpexpect.py
+++ b/pexpect/fdpexpect.py
@@ -1,4 +1,4 @@
-"""This is like :mod:`pexpect`, but it will work with any file descriptor that you
+'''This is like :mod:`pexpect`, but it will work with any file descriptor that you
pass it. You are responsible for opening and close the file descriptor.
This allows you to use Pexpect with sockets and named pipes (FIFOs).
@@ -23,78 +23,101 @@ PEXPECT LICENSE
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-"""
+'''
+
from .spawnbase import SpawnBase
from .exceptions import ExceptionPexpect, TIMEOUT
from .utils import select_ignore_interrupts, poll_ignore_interrupts
import os
-__all__ = ['fdspawn']
+__all__ = ['fdspawn']
class fdspawn(SpawnBase):
- """This is like pexpect.spawn but allows you to supply your own open file
+ '''This is like pexpect.spawn but allows you to supply your own open file
descriptor. For example, you could use it to read through a file looking
- for patterns, or to control a modem or serial device. """
+ for patterns, or to control a modem or serial device. '''
- def __init__(self, fd, args=None, timeout=30, maxread=2000,
- searchwindowsize=None, logfile=None, encoding=None, codec_errors=
- 'strict', use_poll=False):
- """This takes a file descriptor (an int) or an object that support the
+ def __init__ (self, fd, args=None, timeout=30, maxread=2000, searchwindowsize=None,
+ logfile=None, encoding=None, codec_errors='strict', use_poll=False):
+ '''This takes a file descriptor (an int) or an object that support the
fileno() method (returning an int). All Python file-like objects
- support fileno(). """
+ support fileno(). '''
+
if type(fd) != type(0) and hasattr(fd, 'fileno'):
fd = fd.fileno()
+
if type(fd) != type(0):
- raise ExceptionPexpect(
- 'The fd argument is not an int. If this is a command string then maybe you want to use pexpect.spawn.'
- )
- try:
+ raise ExceptionPexpect('The fd argument is not an int. If this is a command string then maybe you want to use pexpect.spawn.')
+
+ try: # make sure fd is a valid file descriptor
os.fstat(fd)
except OSError:
- raise ExceptionPexpect(
- 'The fd argument is not a valid file descriptor.')
+ raise ExceptionPexpect('The fd argument is not a valid file descriptor.')
+
self.args = None
self.command = None
- SpawnBase.__init__(self, timeout, maxread, searchwindowsize,
- logfile, encoding=encoding, codec_errors=codec_errors)
+ SpawnBase.__init__(self, timeout, maxread, searchwindowsize, logfile,
+ encoding=encoding, codec_errors=codec_errors)
self.child_fd = fd
self.own_fd = False
self.closed = False
self.name = '<file descriptor %d>' % fd
self.use_poll = use_poll
- def close(self):
+ def close (self):
"""Close the file descriptor.
Calling this method a second time does nothing, but if the file
descriptor was closed elsewhere, :class:`OSError` will be raised.
"""
- pass
+ if self.child_fd == -1:
+ return
- def isalive(self):
- """This checks if the file descriptor is still valid. If :func:`os.fstat`
- does not raise an exception then we assume it is alive. """
- pass
+ self.flush()
+ os.close(self.child_fd)
+ self.child_fd = -1
+ self.closed = True
- def terminate(self, force=False):
- """Deprecated and invalid. Just raises an exception."""
- pass
+ def isalive (self):
+ '''This checks if the file descriptor is still valid. If :func:`os.fstat`
+ does not raise an exception then we assume it is alive. '''
+ if self.child_fd == -1:
+ return False
+ try:
+ os.fstat(self.child_fd)
+ return True
+ except:
+ return False
+
+ def terminate (self, force=False): # pragma: no cover
+ '''Deprecated and invalid. Just raises an exception.'''
+ raise ExceptionPexpect('This method is not valid for file descriptors.')
+
+ # These four methods are left around for backwards compatibility, but not
+ # documented as part of fdpexpect. You're encouraged to use os.write
+ # directly.
def send(self, s):
- """Write to fd, return number of bytes written"""
- pass
+ "Write to fd, return number of bytes written"
+ s = self._coerce_send_string(s)
+ self._log(s, 'send')
+
+ b = self._encoder.encode(s, final=False)
+ return os.write(self.child_fd, b)
def sendline(self, s):
- """Write to fd with trailing newline, return number of bytes written"""
- pass
+ "Write to fd with trailing newline, return number of bytes written"
+ s = self._coerce_send_string(s)
+ return self.send(s + self.linesep)
def write(self, s):
- """Write to fd, return None"""
- pass
+ "Write to fd, return None"
+ self.send(s)
def writelines(self, sequence):
- """Call self.write() for each item in sequence"""
- pass
+ "Call self.write() for each item in sequence"
+ for s in sequence:
+ self.write(s)
def read_nonblocking(self, size=1, timeout=-1):
"""
@@ -112,4 +135,18 @@ class fdspawn(SpawnBase):
ready to read. When -1 (default), use self.timeout. When 0, poll.
:return: String containing the bytes read
"""
- pass
+ if os.name == 'posix':
+ if timeout == -1:
+ timeout = self.timeout
+ rlist = [self.child_fd]
+ wlist = []
+ xlist = []
+ if self.use_poll:
+ rlist = poll_ignore_interrupts(rlist, timeout)
+ else:
+ rlist, wlist, xlist = select_ignore_interrupts(
+ rlist, wlist, xlist, timeout
+ )
+ if self.child_fd not in rlist:
+ raise TIMEOUT('Timeout exceeded.')
+ return super(fdspawn, self).read_nonblocking(size)
diff --git a/pexpect/popen_spawn.py b/pexpect/popen_spawn.py
index 5f7d56a..e6bdf07 100644
--- a/pexpect/popen_spawn.py
+++ b/pexpect/popen_spawn.py
@@ -7,91 +7,182 @@ import sys
import time
import signal
import shlex
+
try:
- from queue import Queue, Empty
+ from queue import Queue, Empty # Python 3
except ImportError:
- from Queue import Queue, Empty
+ from Queue import Queue, Empty # Python 2
+
from .spawnbase import SpawnBase, PY3
from .exceptions import EOF
from .utils import string_types
-
class PopenSpawn(SpawnBase):
-
def __init__(self, cmd, timeout=30, maxread=2000, searchwindowsize=None,
- logfile=None, cwd=None, env=None, encoding=None, codec_errors=
- 'strict', preexec_fn=None):
+ logfile=None, cwd=None, env=None, encoding=None,
+ codec_errors='strict', preexec_fn=None):
super(PopenSpawn, self).__init__(timeout=timeout, maxread=maxread,
- searchwindowsize=searchwindowsize, logfile=logfile, encoding=
- encoding, codec_errors=codec_errors)
+ searchwindowsize=searchwindowsize, logfile=logfile,
+ encoding=encoding, codec_errors=codec_errors)
+
+ # Note that `SpawnBase` initializes `self.crlf` to `\r\n`
+ # because the default behaviour for a PTY is to convert
+ # incoming LF to `\r\n` (see the `onlcr` flag and
+ # https://stackoverflow.com/a/35887657/5397009). Here we set
+ # it to `os.linesep` because that is what the spawned
+ # application outputs by default and `popen` doesn't translate
+ # anything.
if encoding is None:
- self.crlf = os.linesep.encode('ascii')
+ self.crlf = os.linesep.encode ("ascii")
else:
- self.crlf = self.string_type(os.linesep)
- kwargs = dict(bufsize=0, stdin=subprocess.PIPE, stderr=subprocess.
- STDOUT, stdout=subprocess.PIPE, cwd=cwd, preexec_fn=preexec_fn,
- env=env)
+ self.crlf = self.string_type (os.linesep)
+
+ kwargs = dict(bufsize=0, stdin=subprocess.PIPE,
+ stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
+ cwd=cwd, preexec_fn=preexec_fn, env=env)
+
if sys.platform == 'win32':
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
kwargs['startupinfo'] = startupinfo
kwargs['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP
+
if isinstance(cmd, string_types) and sys.platform != 'win32':
cmd = shlex.split(cmd, posix=os.name == 'posix')
+
self.proc = subprocess.Popen(cmd, **kwargs)
self.pid = self.proc.pid
self.closed = False
self._buf = self.string_type()
+
self._read_queue = Queue()
self._read_thread = threading.Thread(target=self._read_incoming)
self._read_thread.daemon = True
self._read_thread.start()
+
_read_reached_eof = False
+ def read_nonblocking(self, size, timeout):
+ buf = self._buf
+ if self._read_reached_eof:
+ # We have already finished reading. Use up any buffered data,
+ # then raise EOF
+ if buf:
+ self._buf = buf[size:]
+ return buf[:size]
+ else:
+ self.flag_eof = True
+ raise EOF('End Of File (EOF).')
+
+ if timeout == -1:
+ timeout = self.timeout
+ elif timeout is None:
+ timeout = 1e6
+
+ t0 = time.time()
+ while (time.time() - t0) < timeout and size and len(buf) < size:
+ try:
+ incoming = self._read_queue.get_nowait()
+ except Empty:
+ break
+ else:
+ if incoming is None:
+ self._read_reached_eof = True
+ break
+
+ buf += self._decoder.decode(incoming, final=False)
+
+ r, self._buf = buf[:size], buf[size:]
+
+ self._log(r, 'read')
+ return r
+
def _read_incoming(self):
"""Run in a thread to move output from a pipe to a queue."""
- pass
+ fileno = self.proc.stdout.fileno()
+ while 1:
+ buf = b''
+ try:
+ buf = os.read(fileno, 1024)
+ except OSError as e:
+ self._log(e, 'read')
+
+ if not buf:
+ # This indicates we have reached EOF
+ self._read_queue.put(None)
+ return
+
+ self._read_queue.put(buf)
def write(self, s):
- """This is similar to send() except that there is no return value.
- """
- pass
+ '''This is similar to send() except that there is no return value.
+ '''
+ self.send(s)
def writelines(self, sequence):
- """This calls write() for each element in the sequence.
+ '''This calls write() for each element in the sequence.
The sequence can be any iterable object producing strings, typically a
list of strings. This does not add line separators. There is no return
value.
- """
- pass
+ '''
+ for s in sequence:
+ self.send(s)
def send(self, s):
- """Send data to the subprocess' stdin.
+ '''Send data to the subprocess' stdin.
Returns the number of bytes written.
- """
- pass
+ '''
+ s = self._coerce_send_string(s)
+ self._log(s, 'send')
+
+ b = self._encoder.encode(s, final=False)
+ if PY3:
+ return self.proc.stdin.write(b)
+ else:
+ # On Python 2, .write() returns None, so we return the length of
+ # bytes written ourselves. This assumes they all got written.
+ self.proc.stdin.write(b)
+ return len(b)
def sendline(self, s=''):
- """Wraps send(), sending string ``s`` to child process, with os.linesep
- automatically appended. Returns number of bytes written. """
- pass
+ '''Wraps send(), sending string ``s`` to child process, with os.linesep
+ automatically appended. Returns number of bytes written. '''
+
+ n = self.send(s)
+ return n + self.send(self.linesep)
def wait(self):
- """Wait for the subprocess to finish.
+ '''Wait for the subprocess to finish.
Returns the exit code.
- """
- pass
+ '''
+ status = self.proc.wait()
+ if status >= 0:
+ self.exitstatus = status
+ self.signalstatus = None
+ else:
+ self.exitstatus = None
+ self.signalstatus = -status
+ self.terminated = True
+ return status
def kill(self, sig):
- """Sends a Unix signal to the subprocess.
+ '''Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
- """
- pass
+ '''
+ if sys.platform == 'win32':
+ if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
+ sig = signal.CTRL_C_EVENT
+ elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
+ sig = signal.CTRL_BREAK_EVENT
+ else:
+ sig = signal.SIGTERM
+
+ os.kill(self.proc.pid, sig)
def sendeof(self):
- """Closes the stdin pipe from the writing end."""
- pass
+ '''Closes the stdin pipe from the writing end.'''
+ self.proc.stdin.close()
diff --git a/pexpect/pty_spawn.py b/pexpect/pty_spawn.py
index 4fc030b..8e28ca7 100644
--- a/pexpect/pty_spawn.py
+++ b/pexpect/pty_spawn.py
@@ -6,32 +6,39 @@ import tty
import errno
import signal
from contextlib import contextmanager
+
import ptyprocess
from ptyprocess.ptyprocess import use_native_pty_fork
+
from .exceptions import ExceptionPexpect, EOF, TIMEOUT
from .spawnbase import SpawnBase
-from .utils import which, split_command_line, select_ignore_interrupts, poll_ignore_interrupts
-
+from .utils import (
+ which, split_command_line, select_ignore_interrupts, poll_ignore_interrupts
+)
@contextmanager
def _wrap_ptyprocess_err():
"""Turn ptyprocess errors into our own ExceptionPexpect errors"""
- pass
-
-
-PY3 = sys.version_info[0] >= 3
+ try:
+ yield
+ except ptyprocess.PtyProcessError as e:
+ raise ExceptionPexpect(*e.args)
+PY3 = (sys.version_info[0] >= 3)
class spawn(SpawnBase):
- """This is the main class interface for Pexpect. Use this class to start
- and control child applications. """
+ '''This is the main class interface for Pexpect. Use this class to start
+ and control child applications. '''
+
+ # This is purely informational now - changing it has no effect
use_native_pty_fork = use_native_pty_fork
def __init__(self, command, args=[], timeout=30, maxread=2000,
- searchwindowsize=None, logfile=None, cwd=None, env=None,
- ignore_sighup=False, echo=True, preexec_fn=None, encoding=None,
- codec_errors='strict', dimensions=None, use_poll=False):
- """This is the constructor. The command parameter may be a string that
+ searchwindowsize=None, logfile=None, cwd=None, env=None,
+ ignore_sighup=False, echo=True, preexec_fn=None,
+ encoding=None, codec_errors='strict', dimensions=None,
+ use_poll=False):
+ '''This is the constructor. The command parameter may be a string that
includes a command and any arguments to the command. For example::
child = pexpect.spawn('/usr/bin/ftp')
@@ -178,10 +185,9 @@ class spawn(SpawnBase):
The use_poll attribute enables using select.poll() over select.select()
for socket handling. This is handy if your system could have > 1024 fds
- """
- super(spawn, self).__init__(timeout=timeout, maxread=maxread,
- searchwindowsize=searchwindowsize, logfile=logfile, encoding=
- encoding, codec_errors=codec_errors)
+ '''
+ super(spawn, self).__init__(timeout=timeout, maxread=maxread, searchwindowsize=searchwindowsize,
+ logfile=logfile, encoding=encoding, codec_errors=codec_errors)
self.STDIN_FILENO = pty.STDIN_FILENO
self.STDOUT_FILENO = pty.STDOUT_FILENO
self.STDERR_FILENO = pty.STDERR_FILENO
@@ -200,16 +206,15 @@ class spawn(SpawnBase):
self.use_poll = use_poll
def __str__(self):
- """This returns a human-readable string that represents the state of
- the object. """
+ '''This returns a human-readable string that represents the state of
+ the object. '''
+
s = []
s.append(repr(self))
s.append('command: ' + str(self.command))
s.append('args: %r' % (self.args,))
- s.append('buffer (last %s chars): %r' % (self.str_last_chars, self.
- buffer[-self.str_last_chars:]))
- s.append('before (last %s chars): %r' % (self.str_last_chars, self.
- before[-self.str_last_chars:] if self.before else ''))
+ s.append('buffer (last %s chars): %r' % (self.str_last_chars,self.buffer[-self.str_last_chars:]))
+ s.append('before (last %s chars): %r' % (self.str_last_chars,self.before[-self.str_last_chars:] if self.before else ''))
s.append('after: %r' % (self.after,))
s.append('match: %r' % (self.match,))
s.append('match_index: ' + str(self.match_index))
@@ -233,36 +238,111 @@ class spawn(SpawnBase):
return '\n'.join(s)
def _spawn(self, command, args=[], preexec_fn=None, dimensions=None):
- """This starts the given command in a child process. This does all the
+ '''This starts the given command in a child process. This does all the
fork/exec type of stuff for a pty. This is called by __init__. If args
is empty then command will be parsed (split on spaces) and args will be
- set to parsed arguments. """
- pass
+ set to parsed arguments. '''
+
+ # The pid and child_fd of this object get set by this method.
+ # Note that it is difficult for this method to fail.
+ # You cannot detect if the child process cannot start.
+ # So the only way you can tell if the child process started
+ # or not is to try to read from the file descriptor. If you get
+ # EOF immediately then it means that the child is already dead.
+ # That may not necessarily be bad because you may have spawned a child
+ # that performs some task; creates no stdout output; and then dies.
+
+ # If command is an int type then it may represent a file descriptor.
+ if isinstance(command, type(0)):
+ raise ExceptionPexpect('Command is an int type. ' +
+ 'If this is a file descriptor then maybe you want to ' +
+ 'use fdpexpect.fdspawn which takes an existing ' +
+ 'file descriptor instead of a command string.')
+
+ if not isinstance(args, type([])):
+ raise TypeError('The argument, args, must be a list.')
+
+ if args == []:
+ self.args = split_command_line(command)
+ self.command = self.args[0]
+ else:
+ # Make a shallow copy of the args list.
+ self.args = args[:]
+ self.args.insert(0, command)
+ self.command = command
+
+ command_with_path = which(self.command, env=self.env)
+ if command_with_path is None:
+ raise ExceptionPexpect('The command was not found or was not ' +
+ 'executable: %s.' % self.command)
+ self.command = command_with_path
+ self.args[0] = self.command
+
+ self.name = '<' + ' '.join(self.args) + '>'
+
+ assert self.pid is None, 'The pid member must be None.'
+ assert self.command is not None, 'The command member must not be None.'
+
+ kwargs = {'echo': self.echo, 'preexec_fn': preexec_fn}
+ if self.ignore_sighup:
+ def preexec_wrapper():
+ "Set SIGHUP to be ignored, then call the real preexec_fn"
+ signal.signal(signal.SIGHUP, signal.SIG_IGN)
+ if preexec_fn is not None:
+ preexec_fn()
+ kwargs['preexec_fn'] = preexec_wrapper
+
+ if dimensions is not None:
+ kwargs['dimensions'] = dimensions
+
+ if self.encoding is not None:
+ # Encode command line using the specified encoding
+ self.args = [a if isinstance(a, bytes) else a.encode(self.encoding)
+ for a in self.args]
+
+ self.ptyproc = self._spawnpty(self.args, env=self.env,
+ cwd=self.cwd, **kwargs)
+
+ self.pid = self.ptyproc.pid
+ self.child_fd = self.ptyproc.fd
+
+
+ self.terminated = False
+ self.closed = False
def _spawnpty(self, args, **kwargs):
- """Spawn a pty and return an instance of PtyProcess."""
- pass
+ '''Spawn a pty and return an instance of PtyProcess.'''
+ return ptyprocess.PtyProcess.spawn(args, **kwargs)
def close(self, force=True):
- """This closes the connection with the child application. Note that
+ '''This closes the connection with the child application. Note that
calling close() more than once is valid. This emulates standard Python
behavior with files. Set force to True if you want to make sure that
the child is terminated (SIGKILL is sent if the child ignores SIGHUP
- and SIGINT). """
- pass
+ and SIGINT). '''
+
+ self.flush()
+ with _wrap_ptyprocess_err():
+ # PtyProcessError may be raised if it is not possible to terminate
+ # the child.
+ self.ptyproc.close(force=force)
+ self.isalive() # Update exit status from ptyproc
+ self.child_fd = -1
+ self.closed = True
def isatty(self):
- """This returns True if the file descriptor is open and connected to a
+ '''This returns True if the file descriptor is open and connected to a
tty(-like) device, else False.
On SVR4-style platforms implementing streams, such as SunOS and HP-UX,
the child pty may not appear as a terminal device. This means
methods such as setecho(), setwinsize(), getwinsize() may raise an
- IOError. """
- pass
+ IOError. '''
+
+ return os.isatty(self.child_fd)
def waitnoecho(self, timeout=-1):
- """This waits until the terminal ECHO flag is set False. This returns
+ '''This waits until the terminal ECHO flag is set False. This returns
True if the echo mode is off. This returns False if the ECHO flag was
not set False before the timeout. This can be used to detect when the
child is waiting for a password. Usually a child application will turn
@@ -276,19 +356,31 @@ class spawn(SpawnBase):
If timeout==-1 then this method will use the value in self.timeout.
If timeout==None then this method to block until ECHO flag is False.
- """
- pass
+ '''
+
+ if timeout == -1:
+ timeout = self.timeout
+ if timeout is not None:
+ end_time = time.time() + timeout
+ while True:
+ if not self.getecho():
+ return True
+ if timeout < 0 and timeout is not None:
+ return False
+ if timeout is not None:
+ timeout = end_time - time.time()
+ time.sleep(0.1)
def getecho(self):
- """This returns the terminal echo mode. This returns True if echo is
+ '''This returns the terminal echo mode. This returns True if echo is
on or False if echo is off. Child applications that are expecting you
to enter a password often set ECHO False. See waitnoecho().
- Not supported on platforms where ``isatty()`` returns False. """
- pass
+ Not supported on platforms where ``isatty()`` returns False. '''
+ return self.ptyproc.getecho()
def setecho(self, state):
- """This sets the terminal echo mode on or off. Note that anything the
+ '''This sets the terminal echo mode on or off. Note that anything the
child sent before the echo will be lost, so you should be sure that
your input buffer is empty before you call setecho(). For example, the
following will work as expected::
@@ -318,11 +410,11 @@ class spawn(SpawnBase):
Not supported on platforms where ``isatty()`` returns False.
- """
- pass
+ '''
+ return self.ptyproc.setecho(state)
def read_nonblocking(self, size=1, timeout=-1):
- """This reads at most size characters from the child application. It
+ '''This reads at most size characters from the child application. It
includes a timeout. If the read does not complete within the timeout
period then a TIMEOUT exception is raised. If the end of file is read
then an EOF exception will be raised. If a logfile is specified, a
@@ -345,23 +437,95 @@ class spawn(SpawnBase):
to read, the buffer will be filled, regardless of timeout.
This is a wrapper around os.read(). It uses select.select() or
- select.poll() to implement the timeout. """
- pass
+ select.poll() to implement the timeout. '''
+
+ if self.closed:
+ raise ValueError('I/O operation on closed file.')
+
+ if self.use_poll:
+ def select(timeout):
+ return poll_ignore_interrupts([self.child_fd], timeout)
+ else:
+ def select(timeout):
+ return select_ignore_interrupts([self.child_fd], [], [], timeout)[0]
+
+ # If there is data available to read right now, read as much as
+ # we can. We do this to increase performance if there are a lot
+ # of bytes to be read. This also avoids calling isalive() too
+ # often. See also:
+ # * https://github.com/pexpect/pexpect/pull/304
+ # * http://trac.sagemath.org/ticket/10295
+ if select(0):
+ try:
+ incoming = super(spawn, self).read_nonblocking(size)
+ except EOF:
+ # Maybe the child is dead: update some attributes in that case
+ self.isalive()
+ raise
+ while len(incoming) < size and select(0):
+ try:
+ incoming += super(spawn, self).read_nonblocking(size - len(incoming))
+ except EOF:
+ # Maybe the child is dead: update some attributes in that case
+ self.isalive()
+ # Don't raise EOF, just return what we read so far.
+ return incoming
+ return incoming
+
+ if timeout == -1:
+ timeout = self.timeout
+
+ if not self.isalive():
+ # The process is dead, but there may or may not be data
+ # available to read. Note that some systems such as Solaris
+ # do not give an EOF when the child dies. In fact, you can
+ # still try to read from the child_fd -- it will block
+ # forever or until TIMEOUT. For that reason, it's important
+ # to do this check before calling select() with timeout.
+ if select(0):
+ return super(spawn, self).read_nonblocking(size)
+ self.flag_eof = True
+ raise EOF('End Of File (EOF). Braindead platform.')
+ elif self.__irix_hack:
+ # Irix takes a long time before it realizes a child was terminated.
+ # Make sure that the timeout is at least 2 seconds.
+ # FIXME So does this mean Irix systems are forced to always have
+ # FIXME a 2 second delay when calling read_nonblocking? That sucks.
+ if timeout is not None and timeout < 2:
+ timeout = 2
+
+ # Because of the select(0) check above, we know that no data
+ # is available right now. But if a non-zero timeout is given
+ # (possibly timeout=None), we call select() with a timeout.
+ if (timeout != 0) and select(timeout):
+ return super(spawn, self).read_nonblocking(size)
+
+ if not self.isalive():
+ # Some platforms, such as Irix, will claim that their
+ # processes are alive; timeout on the select; and
+ # then finally admit that they are not alive.
+ self.flag_eof = True
+ raise EOF('End of File (EOF). Very slow platform.')
+ else:
+ raise TIMEOUT('Timeout exceeded.')
def write(self, s):
- """This is similar to send() except that there is no return value.
- """
- pass
+ '''This is similar to send() except that there is no return value.
+ '''
+
+ self.send(s)
def writelines(self, sequence):
- """This calls write() for each element in the sequence. The sequence
+ '''This calls write() for each element in the sequence. The sequence
can be any iterable object producing strings, typically a list of
strings. This does not add line separators. There is no return value.
- """
- pass
+ '''
+
+ for s in sequence:
+ self.write(s)
def send(self, s):
- """Sends string ``s`` to the child process, returning the number of
+ '''Sends string ``s`` to the child process, returning the number of
bytes written. If a logfile is specified, a copy is written to that
log.
@@ -381,7 +545,7 @@ class spawn(SpawnBase):
256
On such a system, only 256 bytes may be received per line. Any
- subsequent bytes received will be discarded. BEL (``''``) is then
+ subsequent bytes received will be discarded. BEL (``'\a'``) is then
sent to output if IMAXBEL (termios.h) is set by the tty driver.
This is usually enabled by default. Linux does not honor this as
an option -- it behaves as though it is always set on.
@@ -393,62 +557,120 @@ class spawn(SpawnBase):
>>> bash.sendline('stty -icanon')
>>> bash.sendline('base64')
>>> bash.sendline('x' * 5000)
- """
- pass
+ '''
+
+ if self.delaybeforesend is not None:
+ time.sleep(self.delaybeforesend)
+
+ s = self._coerce_send_string(s)
+ self._log(s, 'send')
+
+ b = self._encoder.encode(s, final=False)
+ return os.write(self.child_fd, b)
def sendline(self, s=''):
- """Wraps send(), sending string ``s`` to child process, with
+ '''Wraps send(), sending string ``s`` to child process, with
``os.linesep`` automatically appended. Returns number of bytes
written. Only a limited number of bytes may be sent for each
line in the default terminal mode, see docstring of :meth:`send`.
- """
- pass
+ '''
+ s = self._coerce_send_string(s)
+ return self.send(s + self.linesep)
def _log_control(self, s):
"""Write control characters to the appropriate log files"""
- pass
+ if self.encoding is not None:
+ s = s.decode(self.encoding, 'replace')
+ self._log(s, 'send')
def sendcontrol(self, char):
- """Helper method that wraps send() with mnemonic access for sending control
+ '''Helper method that wraps send() with mnemonic access for sending control
character to the child (such as Ctrl-C or Ctrl-D). For example, to send
- Ctrl-G (ASCII 7, bell, '')::
+ Ctrl-G (ASCII 7, bell, '\a')::
child.sendcontrol('g')
See also, sendintr() and sendeof().
- """
- pass
+ '''
+ n, byte = self.ptyproc.sendcontrol(char)
+ self._log_control(byte)
+ return n
def sendeof(self):
- """This sends an EOF to the child. This sends a character which causes
+ '''This sends an EOF to the child. This sends a character which causes
the pending parent output buffer to be sent to the waiting child
program without waiting for end-of-line. If it is the first character
of the line, the read() in the user program returns 0, which signifies
end-of-file. This means to work as expected a sendeof() has to be
called at the beginning of a line. This method does not send a newline.
It is the responsibility of the caller to ensure the eof is sent at the
- beginning of a line. """
- pass
+ beginning of a line. '''
+
+ n, byte = self.ptyproc.sendeof()
+ self._log_control(byte)
def sendintr(self):
- """This sends a SIGINT to the child. It does not require
- the SIGINT to be the first character on a line. """
- pass
+ '''This sends a SIGINT to the child. It does not require
+ the SIGINT to be the first character on a line. '''
+
+ n, byte = self.ptyproc.sendintr()
+ self._log_control(byte)
+
+ @property
+ def flag_eof(self):
+ return self.ptyproc.flag_eof
+
+ @flag_eof.setter
+ def flag_eof(self, value):
+ self.ptyproc.flag_eof = value
def eof(self):
- """This returns True if the EOF exception was ever raised.
- """
- pass
+ '''This returns True if the EOF exception was ever raised.
+ '''
+ return self.flag_eof
def terminate(self, force=False):
- """This forces a child process to terminate. It starts nicely with
+ '''This forces a child process to terminate. It starts nicely with
SIGHUP and SIGINT. If "force" is True then moves onto SIGKILL. This
returns True if the child was terminated. This returns False if the
- child could not be terminated. """
- pass
+ child could not be terminated. '''
+
+ if not self.isalive():
+ return True
+ try:
+ self.kill(signal.SIGHUP)
+ time.sleep(self.delayafterterminate)
+ if not self.isalive():
+ return True
+ self.kill(signal.SIGCONT)
+ time.sleep(self.delayafterterminate)
+ if not self.isalive():
+ return True
+ self.kill(signal.SIGINT)
+ time.sleep(self.delayafterterminate)
+ if not self.isalive():
+ return True
+ if force:
+ self.kill(signal.SIGKILL)
+ time.sleep(self.delayafterterminate)
+ if not self.isalive():
+ return True
+ else:
+ return False
+ return False
+ except OSError:
+ # I think there are kernel timing issues that sometimes cause
+ # this to happen. I think isalive() reports True, but the
+ # process is dead to the kernel.
+ # Make one last attempt to see if the kernel is up to date.
+ time.sleep(self.delayafterterminate)
+ if not self.isalive():
+ return True
+ else:
+ return False
def wait(self):
- """This waits until the child exits. This is a blocking call. This will
+ '''This waits until the child exits. This is a blocking call. This will
not read any data from the child, so this will block forever if the
child has unread output and has terminated. In other words, the child
may have printed output then called exit(), but, the child is
@@ -457,39 +679,67 @@ class spawn(SpawnBase):
This method is non-blocking if :meth:`wait` has already been called
previously or :meth:`isalive` method returns False. It simply returns
the previously determined exit status.
- """
- pass
+ '''
+
+ ptyproc = self.ptyproc
+ with _wrap_ptyprocess_err():
+ # exception may occur if "Is some other process attempting
+ # "job control with our child pid?"
+ exitstatus = ptyproc.wait()
+ self.status = ptyproc.status
+ self.exitstatus = ptyproc.exitstatus
+ self.signalstatus = ptyproc.signalstatus
+ self.terminated = True
+
+ return exitstatus
def isalive(self):
- """This tests if the child process is running or not. This is
+ '''This tests if the child process is running or not. This is
non-blocking. If the child was terminated then this will read the
exitstatus or signalstatus of the child. This returns True if the child
process appears to be running or False if not. It can take literally
- SECONDS for Solaris to return the right status. """
- pass
+ SECONDS for Solaris to return the right status. '''
+
+ ptyproc = self.ptyproc
+ with _wrap_ptyprocess_err():
+ alive = ptyproc.isalive()
+
+ if not alive:
+ self.status = ptyproc.status
+ self.exitstatus = ptyproc.exitstatus
+ self.signalstatus = ptyproc.signalstatus
+ self.terminated = True
+
+ return alive
def kill(self, sig):
- """This sends the given signal to the child application. In keeping
+
+ '''This sends the given signal to the child application. In keeping
with UNIX tradition it has a misleading name. It does not necessarily
- kill the child unless you send the right signal. """
- pass
+ kill the child unless you send the right signal. '''
+
+ # Same as os.kill, but the pid is given for you.
+ if self.isalive():
+ os.kill(self.pid, sig)
def getwinsize(self):
- """This returns the terminal window size of the child tty. The return
- value is a tuple of (rows, cols). """
- pass
+ '''This returns the terminal window size of the child tty. The return
+ value is a tuple of (rows, cols). '''
+ return self.ptyproc.getwinsize()
def setwinsize(self, rows, cols):
- """This sets the terminal window size of the child tty. This will cause
+ '''This sets the terminal window size of the child tty. This will cause
a SIGWINCH signal to be sent to the child. This does not change the
physical window size. It changes the size reported to TTY-aware
applications like vi or curses -- applications that respond to the
- SIGWINCH signal. """
- pass
+ SIGWINCH signal. '''
+ return self.ptyproc.setwinsize(rows, cols)
+
+
+ def interact(self, escape_character=chr(29),
+ input_filter=None, output_filter=None):
- def interact(self, escape_character=chr(29), input_filter=None,
- output_filter=None):
- """This gives control of the child process to the interactive user (the
+ '''This gives control of the child process to the interactive user (the
human at the keyboard). Keystrokes are sent to the child process, and
the stdout and stderr output of the child process is printed. This
simply echos the child stdout and child stderr to the real stdout and
@@ -529,26 +779,82 @@ class spawn(SpawnBase):
p = pexpect.spawn('/bin/bash')
signal.signal(signal.SIGWINCH, sigwinch_passthrough)
p.interact()
- """
- pass
+ '''
+
+ # Flush the buffer.
+ self.write_to_stdout(self.buffer)
+ self.stdout.flush()
+ self._buffer = self.buffer_type()
+ mode = tty.tcgetattr(self.STDIN_FILENO)
+ tty.setraw(self.STDIN_FILENO)
+ if escape_character is not None and PY3:
+ escape_character = escape_character.encode('latin-1')
+ try:
+ self.__interact_copy(escape_character, input_filter, output_filter)
+ finally:
+ tty.tcsetattr(self.STDIN_FILENO, tty.TCSAFLUSH, mode)
def __interact_writen(self, fd, data):
- """This is used by the interact() method.
- """
- pass
+ '''This is used by the interact() method.
+ '''
- def __interact_read(self, fd):
- """This is used by the interact() method.
- """
- pass
+ while data != b'' and self.isalive():
+ n = os.write(fd, data)
+ data = data[n:]
- def __interact_copy(self, escape_character=None, input_filter=None,
- output_filter=None):
- """This is used by the interact() method.
- """
- pass
+ def __interact_read(self, fd):
+ '''This is used by the interact() method.
+ '''
+
+ return os.read(fd, 1000)
+
+ def __interact_copy(
+ self, escape_character=None, input_filter=None, output_filter=None
+ ):
+
+ '''This is used by the interact() method.
+ '''
+
+ while self.isalive():
+ if self.use_poll:
+ r = poll_ignore_interrupts([self.child_fd, self.STDIN_FILENO])
+ else:
+ r, w, e = select_ignore_interrupts(
+ [self.child_fd, self.STDIN_FILENO], [], []
+ )
+ if self.child_fd in r:
+ try:
+ data = self.__interact_read(self.child_fd)
+ except OSError as err:
+ if err.args[0] == errno.EIO:
+ # Linux-style EOF
+ break
+ raise
+ if data == b'':
+ # BSD-style EOF
+ break
+ if output_filter:
+ data = output_filter(data)
+ self._log(data, 'read')
+ os.write(self.STDOUT_FILENO, data)
+ if self.STDIN_FILENO in r:
+ data = self.__interact_read(self.STDIN_FILENO)
+ if input_filter:
+ data = input_filter(data)
+ i = -1
+ if escape_character is not None:
+ i = data.rfind(escape_character)
+ if i != -1:
+ data = data[:i]
+ if data:
+ self._log(data, 'send')
+ self.__interact_writen(self.child_fd, data)
+ break
+ self._log(data, 'send')
+ self.__interact_writen(self.child_fd, data)
def spawnu(*args, **kwargs):
"""Deprecated: pass encoding to spawn() instead."""
- pass
+ kwargs.setdefault('encoding', 'utf-8')
+ return spawn(*args, **kwargs)
diff --git a/pexpect/pxssh.py b/pexpect/pxssh.py
index b1c2a86..742f59e 100644
--- a/pexpect/pxssh.py
+++ b/pexpect/pxssh.py
@@ -1,4 +1,4 @@
-"""This class extends pexpect.spawn to specialize setting up SSH connections.
+'''This class extends pexpect.spawn to specialize setting up SSH connections.
This adds methods for login, logout, and expecting the shell prompt.
PEXPECT LICENSE
@@ -18,32 +18,39 @@ PEXPECT LICENSE
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-"""
+'''
+
from pexpect import ExceptionPexpect, TIMEOUT, EOF, spawn
import time
import os
import sys
import re
-__all__ = ['ExceptionPxssh', 'pxssh']
+__all__ = ['ExceptionPxssh', 'pxssh']
+# Exception classes used by this module.
class ExceptionPxssh(ExceptionPexpect):
- """Raised for pxssh exceptions.
- """
-
+ '''Raised for pxssh exceptions.
+ '''
if sys.version_info > (3, 0):
from shlex import quote
else:
- _find_unsafe = re.compile('[^\\w@%+=:,./-]').search
+ _find_unsafe = re.compile(r'[^\w@%+=:,./-]').search
def quote(s):
"""Return a shell-escaped version of the string *s*."""
- pass
+ if not s:
+ return "''"
+ if _find_unsafe(s) is None:
+ return s
+ # use single quotes, and put single quotes into double quotes
+ # the string $'b is then quoted as '$'"'"'b'
+ return "'" + s.replace("'", "'\"'\"'") + "'"
-class pxssh(spawn):
- """This class extends pexpect.spawn to specialize setting up SSH
+class pxssh (spawn):
+ '''This class extends pexpect.spawn to specialize setting up SSH
connections. This adds methods for login, logout, and expecting the shell
prompt. It does various tricky things to handle many situations in the SSH
login process. For example, if the session is your first login, then pxssh
@@ -106,59 +113,154 @@ class pxssh(spawn):
`debug_command_string` is only for the test suite to confirm that the string
generated for SSH is correct, using this will not allow you to do
anything other than get a string back from `pxssh.pxssh.login()`.
- """
+ '''
+
+ def __init__ (self, timeout=30, maxread=2000, searchwindowsize=None,
+ logfile=None, cwd=None, env=None, ignore_sighup=True, echo=True,
+ options={}, encoding=None, codec_errors='strict',
+ debug_command_string=False, use_poll=False):
- def __init__(self, timeout=30, maxread=2000, searchwindowsize=None,
- logfile=None, cwd=None, env=None, ignore_sighup=True, echo=True,
- options={}, encoding=None, codec_errors='strict',
- debug_command_string=False, use_poll=False):
spawn.__init__(self, None, timeout=timeout, maxread=maxread,
- searchwindowsize=searchwindowsize, logfile=logfile, cwd=cwd,
- env=env, ignore_sighup=ignore_sighup, echo=echo, encoding=
- encoding, codec_errors=codec_errors, use_poll=use_poll)
+ searchwindowsize=searchwindowsize, logfile=logfile,
+ cwd=cwd, env=env, ignore_sighup=ignore_sighup, echo=echo,
+ encoding=encoding, codec_errors=codec_errors, use_poll=use_poll)
+
self.name = '<pxssh>'
- self.UNIQUE_PROMPT = '\\[PEXPECT\\][\\$\\#] '
+
+ #SUBTLE HACK ALERT! Note that the command that SETS the prompt uses a
+ #slightly different string than the regular expression to match it. This
+ #is because when you set the prompt the command will echo back, but we
+ #don't want to match the echoed command. So if we make the set command
+ #slightly different than the regex we eliminate the problem. To make the
+ #set command different we add a backslash in front of $. The $ doesn't
+ #need to be escaped, but it doesn't hurt and serves to make the set
+ #prompt command different than the regex.
+
+ # used to match the command-line prompt
+ self.UNIQUE_PROMPT = r"\[PEXPECT\][\$\#] "
self.PROMPT = self.UNIQUE_PROMPT
- self.PROMPT_SET_SH = "PS1='[PEXPECT]\\$ '"
- self.PROMPT_SET_CSH = "set prompt='[PEXPECT]\\$ '"
+
+ # used to set shell command-line prompt to UNIQUE_PROMPT.
+ self.PROMPT_SET_SH = r"PS1='[PEXPECT]\$ '"
+ self.PROMPT_SET_CSH = r"set prompt='[PEXPECT]\$ '"
self.PROMPT_SET_ZSH = "prompt restore;\nPS1='[PEXPECT]%(!.#.$) '"
- self.SSH_OPTS = " -o 'PubkeyAuthentication=no'"
+ self.SSH_OPTS = (" -o 'PubkeyAuthentication=no'")
+# Disabling host key checking, makes you vulnerable to MITM attacks.
+# + " -o 'StrictHostKeyChecking=no'"
+# + " -o 'UserKnownHostsFile /dev/null' ")
+ # Disabling X11 forwarding gets rid of the annoying SSH_ASKPASS from
+ # displaying a GUI password dialog. I have not figured out how to
+ # disable only SSH_ASKPASS without also disabling X11 forwarding.
+ # Unsetting SSH_ASKPASS on the remote side doesn't disable it! Annoying!
+ #self.SSH_OPTS = "-x -o 'PubkeyAuthentication=no'"
self.force_password = False
+
self.debug_command_string = debug_command_string
+
+ # User defined SSH options, eg,
+ # ssh.otions = dict(StrictHostKeyChecking="no",UserKnownHostsFile="/dev/null")
self.options = options
def levenshtein_distance(self, a, b):
- """This calculates the Levenshtein distance between a and b.
- """
- pass
+ '''This calculates the Levenshtein distance between a and b.
+ '''
+
+ n, m = len(a), len(b)
+ if n > m:
+ a,b = b,a
+ n,m = m,n
+ current = range(n+1)
+ for i in range(1,m+1):
+ previous, current = current, [i]+[0]*n
+ for j in range(1,n+1):
+ add, delete = previous[j]+1, current[j-1]+1
+ change = previous[j-1]
+ if a[j-1] != b[i-1]:
+ change = change + 1
+ current[j] = min(add, delete, change)
+ return current[n]
def try_read_prompt(self, timeout_multiplier):
- """This facilitates using communication timeouts to perform
+ '''This facilitates using communication timeouts to perform
synchronization as quickly as possible, while supporting high latency
connections with a tunable worst case performance. Fast connections
should be read almost immediately. Worst case performance for this
method is timeout_multiplier * 3 seconds.
- """
- pass
+ '''
+
+ # maximum time allowed to read the first response
+ first_char_timeout = timeout_multiplier * 0.5
- def sync_original_prompt(self, sync_multiplier=1.0):
- """This attempts to find the prompt. Basically, press enter and record
+ # maximum time allowed between subsequent characters
+ inter_char_timeout = timeout_multiplier * 0.1
+
+ # maximum time for reading the entire prompt
+ total_timeout = timeout_multiplier * 3.0
+
+ prompt = self.string_type()
+ begin = time.time()
+ expired = 0.0
+ timeout = first_char_timeout
+
+ while expired < total_timeout:
+ try:
+ prompt += self.read_nonblocking(size=1, timeout=timeout)
+ expired = time.time() - begin # updated total time expired
+ timeout = inter_char_timeout
+ except TIMEOUT:
+ break
+
+ return prompt
+
+ def sync_original_prompt (self, sync_multiplier=1.0):
+ '''This attempts to find the prompt. Basically, press enter and record
the response; press enter again and record the response; if the two
responses are similar then assume we are at the original prompt.
This can be a slow function. Worst case with the default sync_multiplier
can take 12 seconds. Low latency connections are more likely to fail
with a low sync_multiplier. Best case sync time gets worse with a
- high sync multiplier (500 ms with default). """
- pass
-
- def login(self, server, username=None, password='', terminal_type=
- 'ansi', original_prompt='[#$]', login_timeout=10, port=None,
- auto_prompt_reset=True, ssh_key=None, quiet=True, sync_multiplier=1,
- check_local_ip=True, password_regex=
- '(?i)(?:password:)|(?:passphrase for key)', ssh_tunnels={},
- spawn_local_ssh=True, sync_original_prompt=True, ssh_config=None,
- cmd='ssh'):
- """This logs the user into the given server.
+ high sync multiplier (500 ms with default). '''
+
+ # All of these timing pace values are magic.
+ # I came up with these based on what seemed reliable for
+ # connecting to a heavily loaded machine I have.
+ self.sendline()
+ time.sleep(0.1)
+
+ try:
+ # Clear the buffer before getting the prompt.
+ self.try_read_prompt(sync_multiplier)
+ except TIMEOUT:
+ pass
+
+ self.sendline()
+ x = self.try_read_prompt(sync_multiplier)
+
+ self.sendline()
+ a = self.try_read_prompt(sync_multiplier)
+
+ self.sendline()
+ b = self.try_read_prompt(sync_multiplier)
+
+ ld = self.levenshtein_distance(a,b)
+ len_a = len(a)
+ if len_a == 0:
+ return False
+ if float(ld)/len_a < 0.4:
+ return True
+ return False
+
+ ### TODO: This is getting messy and I'm pretty sure this isn't perfect.
+ ### TODO: I need to draw a flow chart for this.
+ ### TODO: Unit tests for SSH tunnels, remote SSH command exec, disabling original prompt sync
+ def login (self, server, username=None, password='', terminal_type='ansi',
+ original_prompt=r"[#$]", login_timeout=10, port=None,
+ auto_prompt_reset=True, ssh_key=None, quiet=True,
+ sync_multiplier=1, check_local_ip=True,
+ password_regex=r'(?i)(?:password:)|(?:passphrase for key)',
+ ssh_tunnels={}, spawn_local_ssh=True,
+ sync_original_prompt=True, ssh_config=None, cmd='ssh'):
+ '''This logs the user into the given server.
It uses 'original_prompt' to try to find the prompt right after login.
When it finds the prompt it immediately tries to reset the prompt to
@@ -204,18 +306,186 @@ class pxssh(spawn):
Alter the ``cmd`` to change the ssh client used, or to prepend it with network
namespaces. For example ```cmd="ip netns exec vlan2 ssh"``` to execute the ssh in
network namespace named ```vlan```.
- """
- pass
-
- def logout(self):
- """Sends exit to the remote shell.
+ '''
+
+ session_regex_array = ["(?i)are you sure you want to continue connecting", original_prompt, password_regex, "(?i)permission denied", "(?i)terminal type", TIMEOUT]
+ session_init_regex_array = []
+ session_init_regex_array.extend(session_regex_array)
+ session_init_regex_array.extend(["(?i)connection closed by remote host", EOF])
+
+ ssh_options = ''.join([" -o '%s=%s'" % (o, v) for (o, v) in self.options.items()])
+ if quiet:
+ ssh_options = ssh_options + ' -q'
+ if not check_local_ip:
+ ssh_options = ssh_options + " -o'NoHostAuthenticationForLocalhost=yes'"
+ if self.force_password:
+ ssh_options = ssh_options + ' ' + self.SSH_OPTS
+ if ssh_config is not None:
+ if spawn_local_ssh and not os.path.isfile(ssh_config):
+ raise ExceptionPxssh('SSH config does not exist or is not a file.')
+ ssh_options = ssh_options + ' -F ' + ssh_config
+ if port is not None:
+ ssh_options = ssh_options + ' -p %s'%(str(port))
+ if ssh_key is not None:
+ # Allow forwarding our SSH key to the current session
+ if ssh_key==True:
+ ssh_options = ssh_options + ' -A'
+ else:
+ if spawn_local_ssh and not os.path.isfile(ssh_key):
+ raise ExceptionPxssh('private ssh key does not exist or is not a file.')
+ ssh_options = ssh_options + ' -i %s' % (ssh_key)
+
+ # SSH tunnels, make sure you know what you're putting into the lists
+ # under each heading. Do not expect these to open 100% of the time,
+ # The port you're requesting might be bound.
+ #
+ # The structure should be like this:
+ # { 'local': ['2424:localhost:22'], # Local SSH tunnels
+ # 'remote': ['2525:localhost:22'], # Remote SSH tunnels
+ # 'dynamic': [8888] } # Dynamic/SOCKS tunnels
+ if ssh_tunnels!={} and isinstance({},type(ssh_tunnels)):
+ tunnel_types = {
+ 'local':'L',
+ 'remote':'R',
+ 'dynamic':'D'
+ }
+ for tunnel_type in tunnel_types:
+ cmd_type = tunnel_types[tunnel_type]
+ if tunnel_type in ssh_tunnels:
+ tunnels = ssh_tunnels[tunnel_type]
+ for tunnel in tunnels:
+ if spawn_local_ssh==False:
+ tunnel = quote(str(tunnel))
+ ssh_options = ssh_options + ' -' + cmd_type + ' ' + str(tunnel)
+
+ if username is not None:
+ ssh_options = ssh_options + ' -l ' + username
+ elif ssh_config is None:
+ raise TypeError('login() needs either a username or an ssh_config')
+ else: # make sure ssh_config has an entry for the server with a username
+ with open(ssh_config, 'rt') as f:
+ lines = [l.strip() for l in f.readlines()]
+
+ server_regex = r'^Host\s+%s\s*$' % server
+ user_regex = r'^User\s+\w+\s*$'
+ config_has_server = False
+ server_has_username = False
+ for line in lines:
+ if not config_has_server and re.match(server_regex, line, re.IGNORECASE):
+ config_has_server = True
+ elif config_has_server and 'hostname' in line.lower():
+ pass
+ elif config_has_server and 'host' in line.lower():
+ server_has_username = False # insurance
+ break # we have left the relevant section
+ elif config_has_server and re.match(user_regex, line, re.IGNORECASE):
+ server_has_username = True
+ break
+
+ if lines:
+ del line
+
+ del lines
+
+ if not config_has_server:
+ raise TypeError('login() ssh_config has no Host entry for %s' % server)
+ elif not server_has_username:
+ raise TypeError('login() ssh_config has no user entry for %s' % server)
+
+ cmd += " %s %s" % (ssh_options, server)
+ if self.debug_command_string:
+ return(cmd)
+
+ # Are we asking for a local ssh command or to spawn one in another session?
+ if spawn_local_ssh:
+ spawn._spawn(self, cmd)
+ else:
+ self.sendline(cmd)
+
+ # This does not distinguish between a remote server 'password' prompt
+ # and a local ssh 'passphrase' prompt (for unlocking a private key).
+ i = self.expect(session_init_regex_array, timeout=login_timeout)
+
+ # First phase
+ if i==0:
+ # New certificate -- always accept it.
+ # This is what you get if SSH does not have the remote host's
+ # public key stored in the 'known_hosts' cache.
+ self.sendline("yes")
+ i = self.expect(session_regex_array)
+ if i==2: # password or passphrase
+ self.sendline(password)
+ i = self.expect(session_regex_array)
+ if i==4:
+ self.sendline(terminal_type)
+ i = self.expect(session_regex_array)
+ if i==7:
+ self.close()
+ raise ExceptionPxssh('Could not establish connection to host')
+
+ # Second phase
+ if i==0:
+ # This is weird. This should not happen twice in a row.
+ self.close()
+ raise ExceptionPxssh('Weird error. Got "are you sure" prompt twice.')
+ elif i==1: # can occur if you have a public key pair set to authenticate.
+ ### TODO: May NOT be OK if expect() got tricked and matched a false prompt.
+ pass
+ elif i==2: # password prompt again
+ # For incorrect passwords, some ssh servers will
+ # ask for the password again, others return 'denied' right away.
+ # If we get the password prompt again then this means
+ # we didn't get the password right the first time.
+ self.close()
+ raise ExceptionPxssh('password refused')
+ elif i==3: # permission denied -- password was bad.
+ self.close()
+ raise ExceptionPxssh('permission denied')
+ elif i==4: # terminal type again? WTF?
+ self.close()
+ raise ExceptionPxssh('Weird error. Got "terminal type" prompt twice.')
+ elif i==5: # Timeout
+ #This is tricky... I presume that we are at the command-line prompt.
+ #It may be that the shell prompt was so weird that we couldn't match
+ #it. Or it may be that we couldn't log in for some other reason. I
+ #can't be sure, but it's safe to guess that we did login because if
+ #I presume wrong and we are not logged in then this should be caught
+ #later when I try to set the shell prompt.
+ pass
+ elif i==6: # Connection closed by remote host
+ self.close()
+ raise ExceptionPxssh('connection closed')
+ else: # Unexpected
+ self.close()
+ raise ExceptionPxssh('unexpected login response')
+ if sync_original_prompt:
+ if not self.sync_original_prompt(sync_multiplier):
+ self.close()
+ raise ExceptionPxssh('could not synchronize with original prompt')
+ # We appear to be in.
+ # set shell prompt to something unique.
+ if auto_prompt_reset:
+ if not self.set_unique_prompt():
+ self.close()
+ raise ExceptionPxssh('could not set shell prompt '
+ '(received: %r, expected: %r).' % (
+ self.before, self.PROMPT,))
+ return True
+
+ def logout (self):
+ '''Sends exit to the remote shell.
If there are stopped jobs then this automatically sends exit twice.
- """
- pass
+ '''
+ self.sendline("exit")
+ index = self.expect([EOF, "(?i)there are stopped jobs"])
+ if index==1:
+ self.sendline("exit")
+ self.expect(EOF)
+ self.close()
def prompt(self, timeout=-1):
- """Match the next shell prompt.
+ '''Match the next shell prompt.
This is little more than a short-cut to the :meth:`~pexpect.spawn.expect`
method. Note that if you called :meth:`login` with
@@ -229,11 +499,17 @@ class pxssh(spawn):
:return: True if the shell prompt was matched, False if the timeout was
reached.
- """
- pass
+ '''
+
+ if timeout == -1:
+ timeout = self.timeout
+ i = self.expect([self.PROMPT, TIMEOUT], timeout=timeout)
+ if i==1:
+ return False
+ return True
def set_unique_prompt(self):
- """This sets the remote prompt to something more unique than ``#`` or ``$``.
+ '''This sets the remote prompt to something more unique than ``#`` or ``$``.
This makes it easier for the :meth:`prompt` method to match the shell prompt
unambiguously. This method is called automatically by the :meth:`login`
method, but you may want to call it manually if you somehow reset the
@@ -246,5 +522,19 @@ class pxssh(spawn):
should call :meth:`login` with ``auto_prompt_reset=False``; then set the
:attr:`PROMPT` attribute to a regular expression. After that, the
:meth:`prompt` method will try to match your prompt pattern.
- """
- pass
+ '''
+
+ self.sendline("unset PROMPT_COMMAND")
+ self.sendline(self.PROMPT_SET_SH) # sh-style
+ i = self.expect ([TIMEOUT, self.PROMPT], timeout=10)
+ if i == 0: # csh-style
+ self.sendline(self.PROMPT_SET_CSH)
+ i = self.expect([TIMEOUT, self.PROMPT], timeout=10)
+ if i == 0: # zsh-style
+ self.sendline(self.PROMPT_SET_ZSH)
+ i = self.expect([TIMEOUT, self.PROMPT], timeout=10)
+ if i == 0:
+ return False
+ return True
+
+# vi:ts=4:sw=4:expandtab:ft=python:
diff --git a/pexpect/replwrap.py b/pexpect/replwrap.py
index 07a3d64..08dbd5e 100644
--- a/pexpect/replwrap.py
+++ b/pexpect/replwrap.py
@@ -3,14 +3,17 @@
import os.path
import signal
import sys
+
import pexpect
-PY3 = sys.version_info[0] >= 3
+
+PY3 = (sys.version_info[0] >= 3)
+
if PY3:
basestring = str
+
PEXPECT_PROMPT = u'[PEXPECT_PROMPT>'
PEXPECT_CONTINUATION_PROMPT = u'[PEXPECT_PROMPT+'
-
class REPLWrapper(object):
"""Wrapper for a REPL.
@@ -27,29 +30,41 @@ class REPLWrapper(object):
:param str extra_init_cmd: Commands to do extra initialisation, such as
disabling pagers.
"""
-
- def __init__(self, cmd_or_spawn, orig_prompt, prompt_change, new_prompt
- =PEXPECT_PROMPT, continuation_prompt=PEXPECT_CONTINUATION_PROMPT,
- extra_init_cmd=None):
+ def __init__(self, cmd_or_spawn, orig_prompt, prompt_change,
+ new_prompt=PEXPECT_PROMPT,
+ continuation_prompt=PEXPECT_CONTINUATION_PROMPT,
+ extra_init_cmd=None):
if isinstance(cmd_or_spawn, basestring):
- self.child = pexpect.spawn(cmd_or_spawn, echo=False, encoding=
- 'utf-8')
+ self.child = pexpect.spawn(cmd_or_spawn, echo=False, encoding='utf-8')
else:
self.child = cmd_or_spawn
if self.child.echo:
+ # Existing spawn instance has echo enabled, disable it
+ # to prevent our input from being repeated to output.
self.child.setecho(False)
self.child.waitnoecho()
+
if prompt_change is None:
self.prompt = orig_prompt
else:
- self.set_prompt(orig_prompt, prompt_change.format(new_prompt,
- continuation_prompt))
+ self.set_prompt(orig_prompt,
+ prompt_change.format(new_prompt, continuation_prompt))
self.prompt = new_prompt
self.continuation_prompt = continuation_prompt
+
self._expect_prompt()
+
if extra_init_cmd is not None:
self.run_command(extra_init_cmd)
+ def set_prompt(self, orig_prompt, prompt_change):
+ self.child.expect(orig_prompt)
+ self.child.sendline(prompt_change)
+
+ def _expect_prompt(self, timeout=-1, async_=False):
+ return self.child.expect_exact([self.prompt, self.continuation_prompt],
+ timeout=timeout, async_=async_)
+
def run_command(self, command, timeout=-1, async_=False):
"""Send a command to the REPL, wait for and return output.
@@ -65,19 +80,57 @@ class REPLWrapper(object):
:mod:`asyncio` Future, which you can yield from to get the same
result that this method would normally give directly.
"""
- pass
+ # Split up multiline commands and feed them in bit-by-bit
+ cmdlines = command.splitlines()
+ # splitlines ignores trailing newlines - add it back in manually
+ if command.endswith('\n'):
+ cmdlines.append('')
+ if not cmdlines:
+ raise ValueError("No command was given")
+
+ if async_:
+ from ._async import repl_run_command_async
+ return repl_run_command_async(self, cmdlines, timeout)
+ res = []
+ self.child.sendline(cmdlines[0])
+ for line in cmdlines[1:]:
+ self._expect_prompt(timeout=timeout)
+ res.append(self.child.before)
+ self.child.sendline(line)
+
+ # Command was fully submitted, now wait for the next prompt
+ if self._expect_prompt(timeout=timeout) == 1:
+ # We got the continuation prompt - command was incomplete
+ self.child.kill(signal.SIGINT)
+ self._expect_prompt(timeout=1)
+ raise ValueError("Continuation prompt found - input was incomplete:\n"
+ + command)
+ return u''.join(res + [self.child.before])
def python(command=sys.executable):
"""Start a Python shell and return a :class:`REPLWrapper` object."""
- pass
+ return REPLWrapper(command, u">>> ", u"import sys; sys.ps1={0!r}; sys.ps2={1!r}")
+def _repl_sh(command, args, non_printable_insert):
+ child = pexpect.spawn(command, args, echo=False, encoding='utf-8')
-def bash(command='bash'):
- """Start a bash shell and return a :class:`REPLWrapper` object."""
- pass
+ # If the user runs 'env', the value of PS1 will be in the output. To avoid
+ # replwrap seeing that as the next prompt, we'll embed the marker characters
+ # for invisible characters in the prompt; these show up when inspecting the
+ # environment variable, but not when bash displays the prompt.
+ ps1 = PEXPECT_PROMPT[:5] + non_printable_insert + PEXPECT_PROMPT[5:]
+ ps2 = PEXPECT_CONTINUATION_PROMPT[:5] + non_printable_insert + PEXPECT_CONTINUATION_PROMPT[5:]
+ prompt_change = u"PS1='{0}' PS2='{1}' PROMPT_COMMAND=''".format(ps1, ps2)
+ return REPLWrapper(child, u'\\$', prompt_change,
+ extra_init_cmd="export PAGER=cat")
+
+def bash(command="bash"):
+ """Start a bash shell and return a :class:`REPLWrapper` object."""
+ bashrc = os.path.join(os.path.dirname(__file__), 'bashrc.sh')
+ return _repl_sh(command, ['--rcfile', bashrc], non_printable_insert='\\[\\]')
-def zsh(command='zsh', args=('--no-rcs', '-V', '+Z')):
+def zsh(command="zsh", args=("--no-rcs", "-V", "+Z")):
"""Start a zsh shell and return a :class:`REPLWrapper` object."""
- pass
+ return _repl_sh(command, list(args), non_printable_insert='%(!..)')
diff --git a/pexpect/run.py b/pexpect/run.py
index 8710741..5695ab7 100644
--- a/pexpect/run.py
+++ b/pexpect/run.py
@@ -1,12 +1,13 @@
import sys
import types
+
from .exceptions import EOF, TIMEOUT
from .pty_spawn import spawn
+def run(command, timeout=30, withexitstatus=False, events=None,
+ extra_args=None, logfile=None, cwd=None, env=None, **kwargs):
-def run(command, timeout=30, withexitstatus=False, events=None, extra_args=
- None, logfile=None, cwd=None, env=None, **kwargs):
- """
+ '''
This function runs the given command; waits for it to finish; then
returns all output as a string. STDERR is included in output. If the full
path to the command is not given then the path is searched.
@@ -90,12 +91,67 @@ def run(command, timeout=30, withexitstatus=False, events=None, extra_args=
Like :class:`spawn`, passing *encoding* will make it work with unicode
instead of bytes. You can pass *codec_errors* to control how errors in
encoding and decoding are handled.
- """
- pass
-
-
-def runu(command, timeout=30, withexitstatus=False, events=None, extra_args
- =None, logfile=None, cwd=None, env=None, **kwargs):
+ '''
+ if timeout == -1:
+ child = spawn(command, maxread=2000, logfile=logfile, cwd=cwd, env=env,
+ **kwargs)
+ else:
+ child = spawn(command, timeout=timeout, maxread=2000, logfile=logfile,
+ cwd=cwd, env=env, **kwargs)
+ if isinstance(events, list):
+ patterns= [x for x,y in events]
+ responses = [y for x,y in events]
+ elif isinstance(events, dict):
+ patterns = list(events.keys())
+ responses = list(events.values())
+ else:
+ # This assumes EOF or TIMEOUT will eventually cause run to terminate.
+ patterns = None
+ responses = None
+ child_result_list = []
+ event_count = 0
+ while True:
+ try:
+ index = child.expect(patterns)
+ if isinstance(child.after, child.allowed_string_types):
+ child_result_list.append(child.before + child.after)
+ else:
+ # child.after may have been a TIMEOUT or EOF,
+ # which we don't want appended to the list.
+ child_result_list.append(child.before)
+ if isinstance(responses[index], child.allowed_string_types):
+ child.send(responses[index])
+ elif (isinstance(responses[index], types.FunctionType) or
+ isinstance(responses[index], types.MethodType)):
+ callback_result = responses[index](locals())
+ sys.stdout.flush()
+ if isinstance(callback_result, child.allowed_string_types):
+ child.send(callback_result)
+ elif callback_result:
+ break
+ else:
+ raise TypeError("parameter `event' at index {index} must be "
+ "a string, method, or function: {value!r}"
+ .format(index=index, value=responses[index]))
+ event_count = event_count + 1
+ except TIMEOUT:
+ child_result_list.append(child.before)
+ break
+ except EOF:
+ child_result_list.append(child.before)
+ break
+ child_result = child.string_type().join(child_result_list)
+ if withexitstatus:
+ child.close()
+ return (child_result, child.exitstatus)
+ else:
+ return child_result
+
+def runu(command, timeout=30, withexitstatus=False, events=None,
+ extra_args=None, logfile=None, cwd=None, env=None, **kwargs):
"""Deprecated: pass encoding to run() instead.
"""
- pass
+ kwargs.setdefault('encoding', 'utf-8')
+ return run(command, timeout=timeout, withexitstatus=withexitstatus,
+ events=events, extra_args=extra_args, logfile=logfile, cwd=cwd,
+ env=env, **kwargs)
diff --git a/pexpect/screen.py b/pexpect/screen.py
index c1cb2a3..79f95c4 100644
--- a/pexpect/screen.py
+++ b/pexpect/screen.py
@@ -1,4 +1,4 @@
-"""This implements a virtual screen. This is used to support ANSI terminal
+'''This implements a virtual screen. This is used to support ANSI terminal
emulation. The screen representation and state is implemented in this class.
Most of the methods are inspired by ANSI screen control codes. The
:class:`~pexpect.ANSI.ANSI` class extends this class to add parsing of ANSI
@@ -21,44 +21,54 @@ PEXPECT LICENSE
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-"""
+'''
+
import codecs
import copy
import sys
+
import warnings
-warnings.warn(
- 'pexpect.screen and pexpect.ANSI are deprecated. We recommend using pyte to emulate a terminal screen: https://pypi.python.org/pypi/pyte'
- , stacklevel=2)
-NUL = 0
-ENQ = 5
-BEL = 7
-BS = 8
-HT = 9
-LF = 10
-VT = 11
-FF = 12
-CR = 13
-SO = 14
-SI = 15
-XON = 17
-XOFF = 19
-CAN = 24
-SUB = 26
-ESC = 27
-DEL = 127
-SPACE = u' '
-PY3 = sys.version_info[0] >= 3
+
+warnings.warn(("pexpect.screen and pexpect.ANSI are deprecated. "
+ "We recommend using pyte to emulate a terminal screen: "
+ "https://pypi.python.org/pypi/pyte"),
+ stacklevel=2)
+
+NUL = 0 # Fill character; ignored on input.
+ENQ = 5 # Transmit answerback message.
+BEL = 7 # Ring the bell.
+BS = 8 # Move cursor left.
+HT = 9 # Move cursor to next tab stop.
+LF = 10 # Line feed.
+VT = 11 # Same as LF.
+FF = 12 # Same as LF.
+CR = 13 # Move cursor to left margin or newline.
+SO = 14 # Invoke G1 character set.
+SI = 15 # Invoke G0 character set.
+XON = 17 # Resume transmission.
+XOFF = 19 # Halt transmission.
+CAN = 24 # Cancel escape sequence.
+SUB = 26 # Same as CAN.
+ESC = 27 # Introduce a control sequence.
+DEL = 127 # Fill character; ignored on input.
+SPACE = u' ' # Space or blank character.
+
+PY3 = (sys.version_info[0] >= 3)
if PY3:
unicode = str
+def constrain (n, min, max):
-def constrain(n, min, max):
- """This returns a number, n constrained to the min and max bounds. """
- pass
+ '''This returns a number, n constrained to the min and max bounds. '''
+ if n < min:
+ return min
+ if n > max:
+ return max
+ return n
class screen:
- """This object maintains the state of a virtual text screen as a
+ '''This object maintains the state of a virtual text screen as a
rectangular array. This maintains a virtual cursor position and handles
scrolling as characters are added. This supports most of the methods needed
by an ANSI text screen. Row and column indexes are 1-based (not zero-based,
@@ -71,18 +81,16 @@ class screen:
unicode strings, with the exception of __str__() under Python 2. Passing
``encoding=None`` limits the API to only accept unicode input, so passing
bytes in will raise :exc:`TypeError`.
- """
+ '''
+ def __init__(self, r=24, c=80, encoding='latin-1', encoding_errors='replace'):
+ '''This initializes a blank screen of the given dimensions.'''
- def __init__(self, r=24, c=80, encoding='latin-1', encoding_errors=
- 'replace'):
- """This initializes a blank screen of the given dimensions."""
self.rows = r
self.cols = c
self.encoding = encoding
self.encoding_errors = encoding_errors
if encoding is not None:
- self.decoder = codecs.getincrementaldecoder(encoding)(
- encoding_errors)
+ self.decoder = codecs.getincrementaldecoder(encoding)(encoding_errors)
else:
self.decoder = None
self.cur_r = 1
@@ -91,164 +99,333 @@ class screen:
self.cur_saved_c = 1
self.scroll_row_start = 1
self.scroll_row_end = self.rows
- self.w = [([SPACE] * self.cols) for _ in range(self.rows)]
+ self.w = [ [SPACE] * self.cols for _ in range(self.rows)]
def _decode(self, s):
- """This converts from the external coding system (as passed to
- the constructor) to the internal one (unicode). """
- pass
+ '''This converts from the external coding system (as passed to
+ the constructor) to the internal one (unicode). '''
+ if self.decoder is not None:
+ return self.decoder.decode(s)
+ else:
+ raise TypeError("This screen was constructed with encoding=None, "
+ "so it does not handle bytes.")
def _unicode(self):
- """This returns a printable representation of the screen as a unicode
+ '''This returns a printable representation of the screen as a unicode
string (which, under Python 3.x, is the same as 'str'). The end of each
- screen line is terminated by a newline."""
- pass
+ screen line is terminated by a newline.'''
+
+ return u'\n'.join ([ u''.join(c) for c in self.w ])
+
if PY3:
__str__ = _unicode
else:
__unicode__ = _unicode
def __str__(self):
- """This returns a printable representation of the screen. The end of
- each screen line is terminated by a newline. """
+ '''This returns a printable representation of the screen. The end of
+ each screen line is terminated by a newline. '''
encoding = self.encoding or 'ascii'
return self._unicode().encode(encoding, 'replace')
- def dump(self):
- """This returns a copy of the screen as a unicode string. This is similar to
+ def dump (self):
+ '''This returns a copy of the screen as a unicode string. This is similar to
__str__/__unicode__ except that lines are not terminated with line
- feeds."""
- pass
+ feeds.'''
+
+ return u''.join ([ u''.join(c) for c in self.w ])
- def pretty(self):
- """This returns a copy of the screen as a unicode string with an ASCII
+ def pretty (self):
+ '''This returns a copy of the screen as a unicode string with an ASCII
text box around the screen border. This is similar to
- __str__/__unicode__ except that it adds a box."""
- pass
+ __str__/__unicode__ except that it adds a box.'''
- def cr(self):
- """This moves the cursor to the beginning (col 1) of the current row.
- """
- pass
+ top_bot = u'+' + u'-'*self.cols + u'+\n'
+ return top_bot + u'\n'.join([u'|'+line+u'|' for line in unicode(self).split(u'\n')]) + u'\n' + top_bot
- def lf(self):
- """This moves the cursor down with scrolling.
- """
- pass
+ def fill (self, ch=SPACE):
- def crlf(self):
- """This advances the cursor with CRLF properties.
+ if isinstance(ch, bytes):
+ ch = self._decode(ch)
+
+ self.fill_region (1,1,self.rows,self.cols, ch)
+
+ def fill_region (self, rs,cs, re,ce, ch=SPACE):
+
+ if isinstance(ch, bytes):
+ ch = self._decode(ch)
+
+ rs = constrain (rs, 1, self.rows)
+ re = constrain (re, 1, self.rows)
+ cs = constrain (cs, 1, self.cols)
+ ce = constrain (ce, 1, self.cols)
+ if rs > re:
+ rs, re = re, rs
+ if cs > ce:
+ cs, ce = ce, cs
+ for r in range (rs, re+1):
+ for c in range (cs, ce + 1):
+ self.put_abs (r,c,ch)
+
+ def cr (self):
+ '''This moves the cursor to the beginning (col 1) of the current row.
+ '''
+
+ self.cursor_home (self.cur_r, 1)
+
+ def lf (self):
+ '''This moves the cursor down with scrolling.
+ '''
+
+ old_r = self.cur_r
+ self.cursor_down()
+ if old_r == self.cur_r:
+ self.scroll_up ()
+ self.erase_line()
+
+ def crlf (self):
+ '''This advances the cursor with CRLF properties.
The cursor will line wrap and the screen may scroll.
- """
- pass
+ '''
- def newline(self):
- """This is an alias for crlf().
- """
- pass
+ self.cr ()
+ self.lf ()
- def put_abs(self, r, c, ch):
- """Screen array starts at 1 index."""
- pass
+ def newline (self):
+ '''This is an alias for crlf().
+ '''
- def put(self, ch):
- """This puts a characters at the current cursor position.
- """
- pass
+ self.crlf()
+
+ def put_abs (self, r, c, ch):
+ '''Screen array starts at 1 index.'''
+
+ r = constrain (r, 1, self.rows)
+ c = constrain (c, 1, self.cols)
+ if isinstance(ch, bytes):
+ ch = self._decode(ch)[0]
+ else:
+ ch = ch[0]
+ self.w[r-1][c-1] = ch
+
+ def put (self, ch):
+ '''This puts a characters at the current cursor position.
+ '''
+
+ if isinstance(ch, bytes):
+ ch = self._decode(ch)
- def insert_abs(self, r, c, ch):
- """This inserts a character at (r,c). Everything under
+ self.put_abs (self.cur_r, self.cur_c, ch)
+
+ def insert_abs (self, r, c, ch):
+ '''This inserts a character at (r,c). Everything under
and to the right is shifted right one character.
The last character of the line is lost.
- """
- pass
+ '''
- def get_region(self, rs, cs, re, ce):
- """This returns a list of lines representing the region.
- """
- pass
+ if isinstance(ch, bytes):
+ ch = self._decode(ch)
- def cursor_constrain(self):
- """This keeps the cursor within the screen area.
- """
- pass
+ r = constrain (r, 1, self.rows)
+ c = constrain (c, 1, self.cols)
+ for ci in range (self.cols, c, -1):
+ self.put_abs (r,ci, self.get_abs(r,ci-1))
+ self.put_abs (r,c,ch)
- def cursor_force_position(self, r, c):
- """Identical to Cursor Home."""
- pass
+ def insert (self, ch):
- def cursor_save(self):
- """Save current cursor position."""
- pass
+ if isinstance(ch, bytes):
+ ch = self._decode(ch)
- def cursor_unsave(self):
- """Restores cursor position after a Save Cursor."""
- pass
+ self.insert_abs (self.cur_r, self.cur_c, ch)
- def cursor_save_attrs(self):
- """Save current cursor position."""
- pass
+ def get_abs (self, r, c):
- def cursor_restore_attrs(self):
- """Restores cursor position after a Save Cursor."""
- pass
+ r = constrain (r, 1, self.rows)
+ c = constrain (c, 1, self.cols)
+ return self.w[r-1][c-1]
- def scroll_constrain(self):
- """This keeps the scroll region within the screen region."""
- pass
+ def get (self):
- def scroll_screen(self):
- """Enable scrolling for entire display."""
- pass
+ self.get_abs (self.cur_r, self.cur_c)
- def scroll_screen_rows(self, rs, re):
- """Enable scrolling from row {start} to row {end}."""
- pass
+ def get_region (self, rs,cs, re,ce):
+ '''This returns a list of lines representing the region.
+ '''
- def scroll_down(self):
- """Scroll display down one line."""
- pass
+ rs = constrain (rs, 1, self.rows)
+ re = constrain (re, 1, self.rows)
+ cs = constrain (cs, 1, self.cols)
+ ce = constrain (ce, 1, self.cols)
+ if rs > re:
+ rs, re = re, rs
+ if cs > ce:
+ cs, ce = ce, cs
+ sc = []
+ for r in range (rs, re+1):
+ line = u''
+ for c in range (cs, ce + 1):
+ ch = self.get_abs (r,c)
+ line = line + ch
+ sc.append (line)
+ return sc
- def scroll_up(self):
- """Scroll display up one line."""
- pass
+ def cursor_constrain (self):
+ '''This keeps the cursor within the screen area.
+ '''
- def erase_end_of_line(self):
- """Erases from the current cursor position to the end of the current
- line."""
- pass
+ self.cur_r = constrain (self.cur_r, 1, self.rows)
+ self.cur_c = constrain (self.cur_c, 1, self.cols)
- def erase_start_of_line(self):
- """Erases from the current cursor position to the start of the current
- line."""
- pass
+ def cursor_home (self, r=1, c=1): # <ESC>[{ROW};{COLUMN}H
- def erase_line(self):
- """Erases the entire current line."""
- pass
+ self.cur_r = r
+ self.cur_c = c
+ self.cursor_constrain ()
- def erase_down(self):
- """Erases the screen from the current line down to the bottom of the
- screen."""
- pass
+ def cursor_back (self,count=1): # <ESC>[{COUNT}D (not confused with down)
- def erase_up(self):
- """Erases the screen from the current line up to the top of the
- screen."""
- pass
+ self.cur_c = self.cur_c - count
+ self.cursor_constrain ()
- def erase_screen(self):
- """Erases the screen with the background color."""
- pass
+ def cursor_down (self,count=1): # <ESC>[{COUNT}B (not confused with back)
+
+ self.cur_r = self.cur_r + count
+ self.cursor_constrain ()
+
+ def cursor_forward (self,count=1): # <ESC>[{COUNT}C
+
+ self.cur_c = self.cur_c + count
+ self.cursor_constrain ()
+
+ def cursor_up (self,count=1): # <ESC>[{COUNT}A
+
+ self.cur_r = self.cur_r - count
+ self.cursor_constrain ()
+
+ def cursor_up_reverse (self): # <ESC> M (called RI -- Reverse Index)
+
+ old_r = self.cur_r
+ self.cursor_up()
+ if old_r == self.cur_r:
+ self.scroll_up()
+
+ def cursor_force_position (self, r, c): # <ESC>[{ROW};{COLUMN}f
+ '''Identical to Cursor Home.'''
+
+ self.cursor_home (r, c)
+
+ def cursor_save (self): # <ESC>[s
+ '''Save current cursor position.'''
+
+ self.cursor_save_attrs()
+
+ def cursor_unsave (self): # <ESC>[u
+ '''Restores cursor position after a Save Cursor.'''
+
+ self.cursor_restore_attrs()
+
+ def cursor_save_attrs (self): # <ESC>7
+ '''Save current cursor position.'''
+
+ self.cur_saved_r = self.cur_r
+ self.cur_saved_c = self.cur_c
+
+ def cursor_restore_attrs (self): # <ESC>8
+ '''Restores cursor position after a Save Cursor.'''
+
+ self.cursor_home (self.cur_saved_r, self.cur_saved_c)
+
+ def scroll_constrain (self):
+ '''This keeps the scroll region within the screen region.'''
+
+ if self.scroll_row_start <= 0:
+ self.scroll_row_start = 1
+ if self.scroll_row_end > self.rows:
+ self.scroll_row_end = self.rows
+
+ def scroll_screen (self): # <ESC>[r
+ '''Enable scrolling for entire display.'''
+
+ self.scroll_row_start = 1
+ self.scroll_row_end = self.rows
+
+ def scroll_screen_rows (self, rs, re): # <ESC>[{start};{end}r
+ '''Enable scrolling from row {start} to row {end}.'''
+
+ self.scroll_row_start = rs
+ self.scroll_row_end = re
+ self.scroll_constrain()
+
+ def scroll_down (self): # <ESC>D
+ '''Scroll display down one line.'''
+
+ # Screen is indexed from 1, but arrays are indexed from 0.
+ s = self.scroll_row_start - 1
+ e = self.scroll_row_end - 1
+ self.w[s+1:e+1] = copy.deepcopy(self.w[s:e])
+
+ def scroll_up (self): # <ESC>M
+ '''Scroll display up one line.'''
+
+ # Screen is indexed from 1, but arrays are indexed from 0.
+ s = self.scroll_row_start - 1
+ e = self.scroll_row_end - 1
+ self.w[s:e] = copy.deepcopy(self.w[s+1:e+1])
+
+ def erase_end_of_line (self): # <ESC>[0K -or- <ESC>[K
+ '''Erases from the current cursor position to the end of the current
+ line.'''
+
+ self.fill_region (self.cur_r, self.cur_c, self.cur_r, self.cols)
+
+ def erase_start_of_line (self): # <ESC>[1K
+ '''Erases from the current cursor position to the start of the current
+ line.'''
+
+ self.fill_region (self.cur_r, 1, self.cur_r, self.cur_c)
+
+ def erase_line (self): # <ESC>[2K
+ '''Erases the entire current line.'''
+
+ self.fill_region (self.cur_r, 1, self.cur_r, self.cols)
+
+ def erase_down (self): # <ESC>[0J -or- <ESC>[J
+ '''Erases the screen from the current line down to the bottom of the
+ screen.'''
+
+ self.erase_end_of_line ()
+ self.fill_region (self.cur_r + 1, 1, self.rows, self.cols)
+
+ def erase_up (self): # <ESC>[1J
+ '''Erases the screen from the current line up to the top of the
+ screen.'''
+
+ self.erase_start_of_line ()
+ self.fill_region (self.cur_r-1, 1, 1, self.cols)
+
+ def erase_screen (self): # <ESC>[2J
+ '''Erases the screen with the background color.'''
+
+ self.fill ()
+
+ def set_tab (self): # <ESC>H
+ '''Sets a tab at the current position.'''
- def set_tab(self):
- """Sets a tab at the current position."""
pass
- def clear_tab(self):
- """Clears tab at the current position."""
+ def clear_tab (self): # <ESC>[g
+ '''Clears tab at the current position.'''
+
pass
- def clear_all_tabs(self):
- """Clears all tabs."""
+ def clear_all_tabs (self): # <ESC>[3g
+ '''Clears all tabs.'''
+
pass
+
+# Insert line Esc [ Pn L
+# Delete line Esc [ Pn M
+# Delete character Esc [ Pn P
+# Scrolling region Esc [ Pn(top);Pn(bot) r
+
diff --git a/pexpect/socket_pexpect.py b/pexpect/socket_pexpect.py
index 5d7ca43..cb11ac2 100644
--- a/pexpect/socket_pexpect.py
+++ b/pexpect/socket_pexpect.py
@@ -19,11 +19,14 @@ PEXPECT LICENSE
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""
+
import socket
from contextlib import contextmanager
+
from .exceptions import TIMEOUT, EOF
from .spawnbase import SpawnBase
-__all__ = ['SocketSpawn']
+
+__all__ = ["SocketSpawn"]
class SocketSpawn(SpawnBase):
@@ -31,18 +34,35 @@ class SocketSpawn(SpawnBase):
rather than the unix-specific file descriptor api. Thus, it works with
remote connections on both unix and windows."""
- def __init__(self, socket: socket.socket, args=None, timeout=30,
- maxread=2000, searchwindowsize=None, logfile=None, encoding=None,
- codec_errors='strict', use_poll=False):
+ def __init__(
+ self,
+ socket: socket.socket,
+ args=None,
+ timeout=30,
+ maxread=2000,
+ searchwindowsize=None,
+ logfile=None,
+ encoding=None,
+ codec_errors="strict",
+ use_poll=False,
+ ):
"""This takes an open socket."""
+
self.args = None
self.command = None
- SpawnBase.__init__(self, timeout, maxread, searchwindowsize,
- logfile, encoding=encoding, codec_errors=codec_errors)
+ SpawnBase.__init__(
+ self,
+ timeout,
+ maxread,
+ searchwindowsize,
+ logfile,
+ encoding=encoding,
+ codec_errors=codec_errors,
+ )
self.socket = socket
self.child_fd = socket.fileno()
self.closed = False
- self.name = '<socket %s>' % socket
+ self.name = "<socket %s>" % socket
self.use_poll = use_poll
def close(self):
@@ -51,27 +71,50 @@ class SocketSpawn(SpawnBase):
Calling this method a second time does nothing, but if the file
descriptor was closed elsewhere, :class:`OSError` will be raised.
"""
- pass
+ if self.child_fd == -1:
+ return
+
+ self.flush()
+ self.socket.shutdown(socket.SHUT_RDWR)
+ self.socket.close()
+ self.child_fd = -1
+ self.closed = True
def isalive(self):
""" Alive if the fileno is valid """
- pass
+ return self.socket.fileno() >= 0
- def send(self, s) ->int:
+ def send(self, s) -> int:
"""Write to socket, return number of bytes written"""
- pass
+ s = self._coerce_send_string(s)
+ self._log(s, "send")
+
+ b = self._encoder.encode(s, final=False)
+ self.socket.sendall(b)
+ return len(b)
- def sendline(self, s) ->int:
+ def sendline(self, s) -> int:
"""Write to socket with trailing newline, return number of bytes written"""
- pass
+ s = self._coerce_send_string(s)
+ return self.send(s + self.linesep)
def write(self, s):
"""Write to socket, return None"""
- pass
+ self.send(s)
def writelines(self, sequence):
- """Call self.write() for each item in sequence"""
- pass
+ "Call self.write() for each item in sequence"
+ for s in sequence:
+ self.write(s)
+
+ @contextmanager
+ def _timeout(self, timeout):
+ saved_timeout = self.socket.gettimeout()
+ try:
+ self.socket.settimeout(timeout)
+ yield
+ finally:
+ self.socket.settimeout(saved_timeout)
def read_nonblocking(self, size=1, timeout=-1):
"""
@@ -89,4 +132,14 @@ class SocketSpawn(SpawnBase):
ready to read. When -1 (default), use self.timeout. When 0, poll.
:return: String containing the bytes read
"""
- pass
+ if timeout == -1:
+ timeout = self.timeout
+ try:
+ with self._timeout(timeout):
+ s = self.socket.recv(size)
+ if s == b'':
+ self.flag_eof = True
+ raise EOF("Socket closed")
+ return s
+ except socket.timeout:
+ raise TIMEOUT("Timeout exceeded.")
diff --git a/pexpect/spawnbase.py b/pexpect/spawnbase.py
index abe78e6..abf8071 100644
--- a/pexpect/spawnbase.py
+++ b/pexpect/spawnbase.py
@@ -6,13 +6,19 @@ import re
import errno
from .exceptions import ExceptionPexpect, EOF, TIMEOUT
from .expect import Expecter, searcher_string, searcher_re
-PY3 = sys.version_info[0] >= 3
-text_type = str if PY3 else unicode
+PY3 = (sys.version_info[0] >= 3)
+text_type = str if PY3 else unicode
class _NullCoder(object):
"""Pass bytes through unchanged."""
+ @staticmethod
+ def encode(b, final=False):
+ return b
+ @staticmethod
+ def decode(b, final=False):
+ return b
class SpawnBase(object):
"""A base class providing the backwards-compatible spawn API for Pexpect.
@@ -25,10 +31,11 @@ class SpawnBase(object):
flag_eof = False
def __init__(self, timeout=30, maxread=2000, searchwindowsize=None,
- logfile=None, encoding=None, codec_errors='strict'):
+ logfile=None, encoding=None, codec_errors='strict'):
self.stdin = sys.stdin
self.stdout = sys.stdout
self.stderr = sys.stderr
+
self.searcher = None
self.ignorecase = False
self.before = None
@@ -38,60 +45,126 @@ class SpawnBase(object):
self.terminated = True
self.exitstatus = None
self.signalstatus = None
+ # status returned by os.waitpid
self.status = None
+ # the child file descriptor is initially closed
self.child_fd = -1
self.timeout = timeout
self.delimiter = EOF
self.logfile = logfile
+ # input from child (read_nonblocking)
self.logfile_read = None
+ # output to send (send, sendline)
self.logfile_send = None
+ # max bytes to read at one time into buffer
self.maxread = maxread
+ # Data before searchwindowsize point is preserved, but not searched.
self.searchwindowsize = searchwindowsize
+ # Delay used before sending data to child. Time in seconds.
+ # Set this to None to skip the time.sleep() call completely.
self.delaybeforesend = 0.05
+ # Used by close() to give kernel time to update process status.
+ # Time in seconds.
self.delayafterclose = 0.1
+ # Used by terminate() to give kernel time to update process status.
+ # Time in seconds.
self.delayafterterminate = 0.1
+ # Delay in seconds to sleep after each call to read_nonblocking().
+ # Set this to None to skip the time.sleep() call completely: that
+ # would restore the behavior from pexpect-2.0 (for performance
+ # reasons or because you don't want to release Python's global
+ # interpreter lock).
self.delayafterread = 0.0001
self.softspace = False
self.name = '<' + repr(self) + '>'
self.closed = True
+
+ # Unicode interface
self.encoding = encoding
self.codec_errors = codec_errors
if encoding is None:
+ # bytes mode (accepts some unicode for backwards compatibility)
self._encoder = self._decoder = _NullCoder()
self.string_type = bytes
self.buffer_type = BytesIO
self.crlf = b'\r\n'
if PY3:
- self.allowed_string_types = bytes, str
+ self.allowed_string_types = (bytes, str)
self.linesep = os.linesep.encode('ascii')
-
def write_to_stdout(b):
try:
return sys.stdout.buffer.write(b)
except AttributeError:
+ # If stdout has been replaced, it may not have .buffer
return sys.stdout.write(b.decode('ascii', 'replace'))
self.write_to_stdout = write_to_stdout
else:
- self.allowed_string_types = basestring,
+ self.allowed_string_types = (basestring,) # analysis:ignore
self.linesep = os.linesep
self.write_to_stdout = sys.stdout.write
else:
- self._encoder = codecs.getincrementalencoder(encoding)(codec_errors
- )
- self._decoder = codecs.getincrementaldecoder(encoding)(codec_errors
- )
+ # unicode mode
+ self._encoder = codecs.getincrementalencoder(encoding)(codec_errors)
+ self._decoder = codecs.getincrementaldecoder(encoding)(codec_errors)
self.string_type = text_type
self.buffer_type = StringIO
self.crlf = u'\r\n'
- self.allowed_string_types = text_type,
+ self.allowed_string_types = (text_type, )
if PY3:
self.linesep = os.linesep
else:
self.linesep = os.linesep.decode('ascii')
+ # This can handle unicode in both Python 2 and 3
self.write_to_stdout = sys.stdout.write
+ # storage for async transport
self.async_pw_transport = None
+ # This is the read buffer. See maxread.
self._buffer = self.buffer_type()
+ # The buffer may be trimmed for efficiency reasons. This is the
+ # untrimmed buffer, used to create the before attribute.
self._before = self.buffer_type()
+
+ def _log(self, s, direction):
+ if self.logfile is not None:
+ self.logfile.write(s)
+ self.logfile.flush()
+ second_log = self.logfile_send if (direction=='send') else self.logfile_read
+ if second_log is not None:
+ second_log.write(s)
+ second_log.flush()
+
+ # For backwards compatibility, in bytes mode (when encoding is None)
+ # unicode is accepted for send and expect. Unicode mode is strictly unicode
+ # only.
+ def _coerce_expect_string(self, s):
+ if self.encoding is None and not isinstance(s, bytes):
+ return s.encode('ascii')
+ return s
+
+ # In bytes mode, regex patterns should also be of bytes type
+ def _coerce_expect_re(self, r):
+ p = r.pattern
+ if self.encoding is None and not isinstance(p, bytes):
+ return re.compile(p.encode('utf-8'))
+ # And vice-versa
+ elif self.encoding is not None and isinstance(p, bytes):
+ return re.compile(p.decode('utf-8'))
+ return r
+
+ def _coerce_send_string(self, s):
+ if self.encoding is None and not isinstance(s, bytes):
+ return s.encode('utf-8')
+ return s
+
+ def _get_buffer(self):
+ return self._buffer.getvalue()
+
+ def _set_buffer(self, value):
+ self._buffer = self.buffer_type()
+ self._buffer.write(value)
+
+ # This property is provided for backwards compatibility (self.buffer used
+ # to be a string/bytes object)
buffer = property(_get_buffer, _set_buffer)
def read_nonblocking(self, size=1, timeout=None):
@@ -101,10 +174,36 @@ class SpawnBase(object):
The timeout parameter is ignored.
"""
- pass
+
+ try:
+ s = os.read(self.child_fd, size)
+ except OSError as err:
+ if err.args[0] == errno.EIO:
+ # Linux-style EOF
+ self.flag_eof = True
+ raise EOF('End Of File (EOF). Exception style platform.')
+ raise
+ if s == b'':
+ # BSD-style EOF
+ self.flag_eof = True
+ raise EOF('End Of File (EOF). Empty string style platform.')
+
+ s = self._decoder.decode(s, final=False)
+ self._log(s, 'read')
+ return s
+
+ def _pattern_type_err(self, pattern):
+ raise TypeError('got {badtype} ({badobj!r}) as pattern, must be one'
+ ' of: {goodtypes}, pexpect.EOF, pexpect.TIMEOUT'\
+ .format(badtype=type(pattern),
+ badobj=pattern,
+ goodtypes=', '.join([str(ast)\
+ for ast in self.allowed_string_types])
+ )
+ )
def compile_pattern_list(self, patterns):
- """This compiles a pattern-string or a list of pattern-strings.
+ '''This compiles a pattern-string or a list of pattern-strings.
Patterns must be a StringType, EOF, TIMEOUT, SRE_Pattern, or a list of
those. Patterns may also be None which results in an empty list (you
might do this if waiting for an EOF or TIMEOUT condition without
@@ -125,12 +224,35 @@ class SpawnBase(object):
...
i = self.expect_list(cpl, timeout)
...
- """
- pass
+ '''
+
+ if patterns is None:
+ return []
+ if not isinstance(patterns, list):
+ patterns = [patterns]
+
+ # Allow dot to match \n
+ compile_flags = re.DOTALL
+ if self.ignorecase:
+ compile_flags = compile_flags | re.IGNORECASE
+ compiled_pattern_list = []
+ for idx, p in enumerate(patterns):
+ if isinstance(p, self.allowed_string_types):
+ p = self._coerce_expect_string(p)
+ compiled_pattern_list.append(re.compile(p, compile_flags))
+ elif p is EOF:
+ compiled_pattern_list.append(EOF)
+ elif p is TIMEOUT:
+ compiled_pattern_list.append(TIMEOUT)
+ elif isinstance(p, type(re.compile(''))):
+ p = self._coerce_expect_re(p)
+ compiled_pattern_list.append(p)
+ else:
+ self._pattern_type_err(p)
+ return compiled_pattern_list
- def expect(self, pattern, timeout=-1, searchwindowsize=-1, async_=False,
- **kw):
- """This seeks through the stream until a pattern is matched. The
+ def expect(self, pattern, timeout=-1, searchwindowsize=-1, async_=False, **kw):
+ '''This seeks through the stream until a pattern is matched. The
pattern is overloaded and may take several types. The pattern can be a
StringType, EOF, a compiled re, or a list of any of those types.
Strings will be compiled to re types. This returns the index into the
@@ -222,12 +344,19 @@ class SpawnBase(object):
With this non-blocking form::
index = yield from p.expect(patterns, async_=True)
- """
- pass
+ '''
+ if 'async' in kw:
+ async_ = kw.pop('async')
+ if kw:
+ raise TypeError("Unknown keyword arguments: {}".format(kw))
+
+ compiled_pattern_list = self.compile_pattern_list(pattern)
+ return self.expect_list(compiled_pattern_list,
+ timeout, searchwindowsize, async_)
def expect_list(self, pattern_list, timeout=-1, searchwindowsize=-1,
- async_=False, **kw):
- """This takes a list of compiled regular expressions and returns the
+ async_=False, **kw):
+ '''This takes a list of compiled regular expressions and returns the
index into the pattern_list that matched the child output. The list may
also contain EOF or TIMEOUT(which are not compiled regular
expressions). This method is similar to the expect() method except that
@@ -238,12 +367,25 @@ class SpawnBase(object):
Like :meth:`expect`, passing ``async_=True`` will make this return an
asyncio coroutine.
- """
- pass
+ '''
+ if timeout == -1:
+ timeout = self.timeout
+ if 'async' in kw:
+ async_ = kw.pop('async')
+ if kw:
+ raise TypeError("Unknown keyword arguments: {}".format(kw))
+
+ exp = Expecter(self, searcher_re(pattern_list), searchwindowsize)
+ if async_:
+ from ._async import expect_async
+ return expect_async(exp, timeout)
+ else:
+ return exp.expect_loop(timeout)
def expect_exact(self, pattern_list, timeout=-1, searchwindowsize=-1,
- async_=False, **kw):
- """This is similar to expect(), but uses plain string matching instead
+ async_=False, **kw):
+
+ '''This is similar to expect(), but uses plain string matching instead
of compiled regular expressions in 'pattern_list'. The 'pattern_list'
may be a string; a list or other sequence of strings; or TIMEOUT and
EOF.
@@ -257,27 +399,79 @@ class SpawnBase(object):
Like :meth:`expect`, passing ``async_=True`` will make this return an
asyncio coroutine.
- """
- pass
+ '''
+ if timeout == -1:
+ timeout = self.timeout
+ if 'async' in kw:
+ async_ = kw.pop('async')
+ if kw:
+ raise TypeError("Unknown keyword arguments: {}".format(kw))
+
+ if (isinstance(pattern_list, self.allowed_string_types) or
+ pattern_list in (TIMEOUT, EOF)):
+ pattern_list = [pattern_list]
+
+ def prepare_pattern(pattern):
+ if pattern in (TIMEOUT, EOF):
+ return pattern
+ if isinstance(pattern, self.allowed_string_types):
+ return self._coerce_expect_string(pattern)
+ self._pattern_type_err(pattern)
+
+ try:
+ pattern_list = iter(pattern_list)
+ except TypeError:
+ self._pattern_type_err(pattern_list)
+ pattern_list = [prepare_pattern(p) for p in pattern_list]
+
+ exp = Expecter(self, searcher_string(pattern_list), searchwindowsize)
+ if async_:
+ from ._async import expect_async
+ return expect_async(exp, timeout)
+ else:
+ return exp.expect_loop(timeout)
def expect_loop(self, searcher, timeout=-1, searchwindowsize=-1):
- """This is the common loop used inside expect. The 'searcher' should be
+ '''This is the common loop used inside expect. The 'searcher' should be
an instance of searcher_re or searcher_string, which describes how and
what to search for in the input.
- See expect() for other arguments, return value and exceptions. """
- pass
+ See expect() for other arguments, return value and exceptions. '''
+
+ exp = Expecter(self, searcher, searchwindowsize)
+ return exp.expect_loop(timeout)
def read(self, size=-1):
- """This reads at most "size" bytes from the file (less if the read hits
+ '''This reads at most "size" bytes from the file (less if the read hits
EOF before obtaining size bytes). If the size argument is negative or
omitted, read all data until EOF is reached. The bytes are returned as
a string object. An empty string is returned when EOF is encountered
- immediately. """
- pass
+ immediately. '''
+
+ if size == 0:
+ return self.string_type()
+ if size < 0:
+ # delimiter default is EOF
+ self.expect(self.delimiter)
+ return self.before
+
+ # I could have done this more directly by not using expect(), but
+ # I deliberately decided to couple read() to expect() so that
+ # I would catch any bugs early and ensure consistent behavior.
+ # It's a little less efficient, but there is less for me to
+ # worry about if I have to later modify read() or expect().
+ # Note, it's OK if size==-1 in the regex. That just means it
+ # will never match anything in which case we stop only on EOF.
+ cre = re.compile(self._coerce_expect_string('.{%d}' % size), re.DOTALL)
+ # delimiter default is EOF
+ index = self.expect([cre, self.delimiter])
+ if index == 0:
+ ### FIXME self.before should be ''. Should I assert this?
+ return self.after
+ return self.before
def readline(self, size=-1):
- """This reads and returns one entire line. The newline at the end of
+ '''This reads and returns one entire line. The newline at the end of
line is returned as part of the string, unless the file ends without a
newline. An empty string is returned if EOF is encountered immediately.
This looks for a newline as a CR/LF pair (\\r\\n) even on UNIX because
@@ -286,39 +480,57 @@ class SpawnBase(object):
If the size argument is 0 then an empty string is returned. In all
other cases the size argument is ignored, which is not standard
- behavior for a file-like object. """
- pass
+ behavior for a file-like object. '''
+
+ if size == 0:
+ return self.string_type()
+ # delimiter default is EOF
+ index = self.expect([self.crlf, self.delimiter])
+ if index == 0:
+ return self.before + self.crlf
+ else:
+ return self.before
def __iter__(self):
- """This is to support iterators over a file-like object.
- """
+ '''This is to support iterators over a file-like object.
+ '''
return iter(self.readline, self.string_type())
def readlines(self, sizehint=-1):
- """This reads until EOF using readline() and returns a list containing
+ '''This reads until EOF using readline() and returns a list containing
the lines thus read. The optional 'sizehint' argument is ignored.
Remember, because this reads until EOF that means the child
process should have closed its stdout. If you run this method on
a child that is still running with its stdout open then this
- method will block until it timesout."""
- pass
+ method will block until it timesout.'''
+
+ lines = []
+ while True:
+ line = self.readline()
+ if not line:
+ break
+ lines.append(line)
+ return lines
def fileno(self):
- """Expose file descriptor for a file-like interface
- """
- pass
+ '''Expose file descriptor for a file-like interface
+ '''
+ return self.child_fd
def flush(self):
- """This does nothing. It is here to support the interface for a
- File-like object. """
+ '''This does nothing. It is here to support the interface for a
+ File-like object. '''
pass
def isatty(self):
"""Overridden in subclass using tty"""
- pass
+ return False
+ # For 'with spawn(...) as child:'
def __enter__(self):
return self
def __exit__(self, etype, evalue, tb):
+ # We rely on subclasses to implement close(). If they don't, it's not
+ # clear what a context manager should do.
self.close()
diff --git a/pexpect/utils.py b/pexpect/utils.py
index 960e622..f774519 100644
--- a/pexpect/utils.py
+++ b/pexpect/utils.py
@@ -4,14 +4,17 @@ import stat
import select
import time
import errno
+
try:
InterruptedError
except NameError:
+ # Alias Python2 exception to Python3
InterruptedError = select.error
+
if sys.version_info[0] >= 3:
- string_types = str,
+ string_types = (str,)
else:
- string_types = unicode, str
+ string_types = (unicode, str)
def is_executable_file(path):
@@ -19,33 +22,166 @@ def is_executable_file(path):
This is roughly ``os.path isfile(path) and os.access(path, os.X_OK)``.
"""
- pass
+ # follow symlinks,
+ fpath = os.path.realpath(path)
+
+ if not os.path.isfile(fpath):
+ # non-files (directories, fifo, etc.)
+ return False
+
+ mode = os.stat(fpath).st_mode
+
+ if (sys.platform.startswith('sunos')
+ and os.getuid() == 0):
+ # When root on Solaris, os.X_OK is True for *all* files, irregardless
+ # of their executability -- instead, any permission bit of any user,
+ # group, or other is fine enough.
+ #
+ # (This may be true for other "Unix98" OS's such as HP-UX and AIX)
+ return bool(mode & (stat.S_IXUSR |
+ stat.S_IXGRP |
+ stat.S_IXOTH))
+
+ return os.access(fpath, os.X_OK)
def which(filename, env=None):
- """This takes a given filename; tries to find it in the environment path;
+ '''This takes a given filename; tries to find it in the environment path;
then checks if it is executable. This returns the full path to the filename
- if found and executable. Otherwise this returns None."""
- pass
+ if found and executable. Otherwise this returns None.'''
+
+ # Special case where filename contains an explicit path.
+ if os.path.dirname(filename) != '' and is_executable_file(filename):
+ return filename
+ if env is None:
+ env = os.environ
+ p = env.get('PATH')
+ if not p:
+ p = os.defpath
+ pathlist = p.split(os.pathsep)
+ for path in pathlist:
+ ff = os.path.join(path, filename)
+ if is_executable_file(ff):
+ return ff
+ return None
def split_command_line(command_line):
- """This splits a command line into a list of arguments. It splits arguments
+
+ '''This splits a command line into a list of arguments. It splits arguments
on spaces, but handles embedded quotes, doublequotes, and escaped
characters. It's impossible to do this with a regular expression, so I
- wrote a little state machine to parse the command line. """
- pass
+ wrote a little state machine to parse the command line. '''
+
+ arg_list = []
+ arg = ''
+
+ # Constants to name the states we can be in.
+ state_basic = 0
+ state_esc = 1
+ state_singlequote = 2
+ state_doublequote = 3
+ # The state when consuming whitespace between commands.
+ state_whitespace = 4
+ state = state_basic
+
+ for c in command_line:
+ if state == state_basic or state == state_whitespace:
+ if c == '\\':
+ # Escape the next character
+ state = state_esc
+ elif c == r"'":
+ # Handle single quote
+ state = state_singlequote
+ elif c == r'"':
+ # Handle double quote
+ state = state_doublequote
+ elif c.isspace():
+ # Add arg to arg_list if we aren't in the middle of whitespace.
+ if state == state_whitespace:
+ # Do nothing.
+ None
+ else:
+ arg_list.append(arg)
+ arg = ''
+ state = state_whitespace
+ else:
+ arg = arg + c
+ state = state_basic
+ elif state == state_esc:
+ arg = arg + c
+ state = state_basic
+ elif state == state_singlequote:
+ if c == r"'":
+ state = state_basic
+ else:
+ arg = arg + c
+ elif state == state_doublequote:
+ if c == r'"':
+ state = state_basic
+ else:
+ arg = arg + c
+
+ if arg != '':
+ arg_list.append(arg)
+ return arg_list
def select_ignore_interrupts(iwtd, owtd, ewtd, timeout=None):
- """This is a wrapper around select.select() that ignores signals. If
+
+ '''This is a wrapper around select.select() that ignores signals. If
select.select raises a select.error exception and errno is an EINTR
error then it is ignored. Mainly this is used to ignore sigwinch
- (terminal resize). """
- pass
+ (terminal resize). '''
+
+ # if select() is interrupted by a signal (errno==EINTR) then
+ # we loop back and enter the select() again.
+ if timeout is not None:
+ end_time = time.time() + timeout
+ while True:
+ try:
+ return select.select(iwtd, owtd, ewtd, timeout)
+ except InterruptedError:
+ err = sys.exc_info()[1]
+ if err.args[0] == errno.EINTR:
+ # if we loop back we have to subtract the
+ # amount of time we already waited.
+ if timeout is not None:
+ timeout = end_time - time.time()
+ if timeout < 0:
+ return([], [], [])
+ else:
+ # something else caused the select.error, so
+ # this actually is an exception.
+ raise
def poll_ignore_interrupts(fds, timeout=None):
- """Simple wrapper around poll to register file descriptors and
- ignore signals."""
- pass
+ '''Simple wrapper around poll to register file descriptors and
+ ignore signals.'''
+
+ if timeout is not None:
+ end_time = time.time() + timeout
+
+ poller = select.poll()
+ for fd in fds:
+ poller.register(fd, select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR)
+
+ while True:
+ try:
+ timeout_ms = None if timeout is None else timeout * 1000
+ results = poller.poll(timeout_ms)
+ return [afd for afd, _ in results]
+ except InterruptedError:
+ err = sys.exc_info()[1]
+ if err.args[0] == errno.EINTR:
+ # if we loop back we have to subtract the
+ # amount of time we already waited.
+ if timeout is not None:
+ timeout = end_time - time.time()
+ if timeout < 0:
+ return []
+ else:
+ # something else caused the select.error, so
+ # this actually is an exception.
+ raise