back to Claude Sonnet 3.5 - Fill-in summary
Claude Sonnet 3.5 - Fill-in: pexpect
Failed to run pytests for test tests
Pytest collection failure.
Patch diff
diff --git a/pexpect/ANSI.py b/pexpect/ANSI.py
index df126d4..eb81388 100644
--- a/pexpect/ANSI.py
+++ b/pexpect/ANSI.py
@@ -114,27 +114,45 @@ class ANSI(term):
def process(self, c):
"""Process a single character. Called by :meth:`write`."""
- pass
+ self.state.process(c)
def write(self, s):
"""Process text, writing it to the virtual screen while handling
ANSI escape codes.
"""
- pass
+ for c in s:
+ self.process(c)
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
+ r, c = self.cur_r, self.cur_c
+ self.put_abs(r, c, ch)
+ self.cursor_forward()
def do_sgr(self, fsm):
"""Select Graphic Rendition, e.g. color. """
- pass
+ for code in fsm.memory:
+ if code == 0:
+ self.attr = 0
+ elif code in range(30, 38):
+ self.attr = (self.attr & 0xfff0) | (code - 30)
+ elif code in range(40, 48):
+ self.attr = (self.attr & 0xff0f) | ((code - 40) << 4)
+ elif code == 1:
+ self.attr = self.attr | 0x0100
+ elif code == 4:
+ self.attr = self.attr | 0x0200
+ elif code == 7:
+ self.attr = self.attr | 0x0400
def do_decsca(self, fsm):
"""Select character protection attribute. """
- pass
+ if fsm.memory and fsm.memory[0] == 1:
+ self.attr = self.attr | 0x0800
+ else:
+ self.attr = self.attr & ~0x0800
def do_modecrap(self, fsm):
"""Handler for [?<number>h and [?<number>l. If anyone
diff --git a/pexpect/FSM.py b/pexpect/FSM.py
index 5fc4095..26d5d05 100644
--- a/pexpect/FSM.py
+++ b/pexpect/FSM.py
@@ -116,7 +116,8 @@ class FSM:
"""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
+ self.current_state = self.initial_state
+ self.input_symbol = None
def add_transition(self, input_symbol, state, action=None, next_state=None
):
@@ -130,7 +131,9 @@ class FSM:
You can also set transitions for a list of symbols by using
add_transition_list(). """
- pass
+ 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):
@@ -142,7 +145,8 @@ 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
+ 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:
@@ -156,7 +160,9 @@ 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
+ 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
@@ -167,7 +173,7 @@ class FSM:
The default transition can be removed by setting the attribute
default_transition to None. """
- pass
+ self.default_transition = (action, next_state)
def get_transition(self, input_symbol, state):
"""This returns (action, next state) given an input_symbol and state.
@@ -190,7 +196,15 @@ class FSM:
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):
"""This is the main method that you call to process input. This may
@@ -200,12 +214,18 @@ class FSM:
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
+ 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. """
- pass
+ for s in input_symbols:
+ self.process(s)
import sys
@@ -217,7 +237,27 @@ def main():
"""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
+ # Example usage of the FSM class
+ fsm = FSM('INIT')
+ fsm.add_transition('a', 'INIT', None, 'A')
+ fsm.add_transition('b', 'A', None, 'B')
+ fsm.add_transition('c', 'B', None, 'C')
+ fsm.add_transition_any('C', None, 'INIT')
+
+ print("Initial state:", fsm.current_state)
+ fsm.process('a')
+ print("After processing 'a':", fsm.current_state)
+ fsm.process('b')
+ print("After processing 'b':", fsm.current_state)
+ fsm.process('c')
+ print("After processing 'c':", fsm.current_state)
+ fsm.process('d')
+ print("After processing 'd':", fsm.current_state)
+
+ print("\nProcessing a list of inputs:")
+ fsm.reset()
+ fsm.process_list('abcd')
+ print("Final state:", fsm.current_state)
if __name__ == '__main__':
diff --git a/pexpect/exceptions.py b/pexpect/exceptions.py
index c66fe77..36e31dc 100644
--- a/pexpect/exceptions.py
+++ b/pexpect/exceptions.py
@@ -18,7 +18,13 @@ class ExceptionPexpect(Exception):
"""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
+ tblist = traceback.extract_tb(sys.exc_info()[2])
+ tblist = [item for item in tblist if self.__filter_not_pexpect(item[0])]
+ return ''.join(traceback.format_list(tblist))
+
+ def __filter_not_pexpect(self, filename):
+ """Return True if the filename is not in the pexpect module"""
+ return 'pexpect' not in filename.lower()
class EOF(ExceptionPexpect):
diff --git a/pexpect/expect.py b/pexpect/expect.py
index dabbe22..82529e5 100644
--- a/pexpect/expect.py
+++ b/pexpect/expect.py
@@ -16,7 +16,24 @@ class Expecter(object):
def expect_loop(self, timeout=-1):
"""Blocking expect"""
- pass
+ if timeout is not None:
+ end_time = time.time() + timeout
+
+ while True:
+ idx = self.searcher.search(self.spawn.buffer, len(self.spawn.buffer))
+ if idx >= 0:
+ return idx
+
+ # No match
+ if timeout is not None and time.time() > end_time:
+ return self.searcher.timeout_index
+
+ # Read more data
+ incoming = self.spawn.read_nonblocking(self.spawn.maxread, timeout)
+ if incoming == b'':
+ return self.searcher.eof_index
+
+ self.spawn.buffer += incoming
class searcher_string(object):
@@ -80,7 +97,25 @@ class searcher_string(object):
If there is a match this returns the index of that string, and sets
'start', 'end' and 'match'. Otherwise, this returns -1. """
- pass
+ absend = len(buffer)
+ abstart = absend - freshlen
+
+ if searchwindowsize is None:
+ searchstart = abstart
+ searchend = absend
+ else:
+ searchstart = max(0, absend - searchwindowsize)
+ searchend = absend
+
+ for index, s in self._strings:
+ pos = buffer.find(s, searchstart, searchend)
+ if pos >= 0:
+ self.match = s
+ self.start = pos
+ self.end = pos + len(s)
+ return index
+
+ return -1
class searcher_re(object):
@@ -143,4 +178,22 @@ class searcher_re(object):
If there is a match this returns the index of that string, and sets
'start', 'end' and 'match'. Otherwise, returns -1."""
- pass
+ absend = len(buffer)
+ abstart = absend - freshlen
+
+ if searchwindowsize is None:
+ searchstart = abstart
+ searchend = absend
+ else:
+ searchstart = max(0, absend - searchwindowsize)
+ searchend = absend
+
+ for index, s in self._searches:
+ match = s.search(buffer, searchstart, searchend)
+ if match is not None:
+ self.match = match
+ self.start = match.start()
+ self.end = match.end()
+ return index
+
+ return -1
diff --git a/pexpect/fdpexpect.py b/pexpect/fdpexpect.py
index 4f4c4d9..124b5b2 100644
--- a/pexpect/fdpexpect.py
+++ b/pexpect/fdpexpect.py
@@ -69,32 +69,47 @@ class fdspawn(SpawnBase):
Calling this method a second time does nothing, but if the file
descriptor was closed elsewhere, :class:`OSError` will be raised.
"""
- pass
+ if not self.closed:
+ os.close(self.child_fd)
+ self.closed = True
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
+ if self.closed:
+ return False
+ try:
+ os.fstat(self.child_fd)
+ return True
+ except OSError:
+ return False
def terminate(self, force=False):
"""Deprecated and invalid. Just raises an exception."""
- pass
+ raise ExceptionPexpect('This method is not valid for file descriptors.')
def send(self, s):
"""Write to fd, return number of bytes written"""
- pass
+ 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
+ n = self.send(s)
+ n += self.send(self.linesep)
+ return n
def write(self, s):
"""Write to fd, return None"""
- pass
+ self.send(s)
def writelines(self, sequence):
"""Call self.write() for each item in sequence"""
- pass
+ for s in sequence:
+ self.write(s)
def read_nonblocking(self, size=1, timeout=-1):
"""
@@ -112,4 +127,34 @@ class fdspawn(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
+
+ if self.use_poll:
+ rfd = poll_ignore_interrupts([self.child_fd], timeout)
+ else:
+ rfd = select_ignore_interrupts([self.child_fd], [], [], timeout)[0]
+
+ if not rfd:
+ raise TIMEOUT('Timeout exceeded.')
+
+ if self.closed:
+ raise OSError('File descriptor %d is closed.' % self.child_fd)
+
+ 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).')
+ raise
+
+ if s == b'':
+ # BSD-style EOF
+ self.flag_eof = True
+ raise EOF('End Of File (EOF).')
+
+ s = self._decoder.decode(s, final=False)
+ self._log(s, 'read')
+ return s
diff --git a/pexpect/popen_spawn.py b/pexpect/popen_spawn.py
index 5f7d56a..540115b 100644
--- a/pexpect/popen_spawn.py
+++ b/pexpect/popen_spawn.py
@@ -50,12 +50,24 @@ class PopenSpawn(SpawnBase):
def _read_incoming(self):
"""Run in a thread to move output from a pipe to a queue."""
- pass
+ while True:
+ try:
+ data = os.read(self.proc.stdout.fileno(), 1024)
+ except OSError:
+ # This happens when the fd is closed
+ break
+ if data == b'':
+ self._read_reached_eof = True
+ break
+ self._read_queue.put(data)
def write(self, s):
"""This is similar to send() except that there is no return value.
"""
- pass
+ if not isinstance(s, bytes):
+ s = s.encode(self.encoding, errors=self.codec_errors)
+ self.proc.stdin.write(s)
+ self.proc.stdin.flush()
def writelines(self, sequence):
"""This calls write() for each element in the sequence.
@@ -64,34 +76,51 @@ class PopenSpawn(SpawnBase):
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):
"""Send data to the subprocess' stdin.
Returns the number of bytes written.
"""
- pass
+ if not isinstance(s, bytes):
+ s = s.encode(self.encoding, errors=self.codec_errors)
+ self.proc.stdin.write(s)
+ self.proc.stdin.flush()
+ return len(s)
def sendline(self, s=''):
"""Wraps send(), sending string ``s`` to child process, with os.linesep
automatically appended. Returns number of bytes written. """
- pass
+ n = self.send(s)
+ n += self.send(self.crlf)
+ return n
def wait(self):
"""Wait for the subprocess to finish.
Returns the exit code.
"""
- pass
+ return self.proc.wait()
def kill(self, sig):
"""Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
"""
- pass
+ if sys.platform != 'win32':
+ os.kill(self.proc.pid, sig)
+ else:
+ if sig == signal.SIGTERM:
+ self.proc.terminate()
+ elif sig == signal.CTRL_C_EVENT:
+ os.kill(self.proc.pid, signal.CTRL_C_EVENT)
+ elif sig == signal.CTRL_BREAK_EVENT:
+ os.kill(self.proc.pid, signal.CTRL_BREAK_EVENT)
+ else:
+ raise ValueError("Unsupported signal on Windows: {}".format(sig))
def sendeof(self):
"""Closes the stdin pipe from the writing end."""
- pass
+ self.proc.stdin.close()
diff --git a/pexpect/pty_spawn.py b/pexpect/pty_spawn.py
index 4fc030b..74de295 100644
--- a/pexpect/pty_spawn.py
+++ b/pexpect/pty_spawn.py
@@ -237,11 +237,43 @@ class spawn(SpawnBase):
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
+ if not isinstance(args, list):
+ args = [args]
+
+ if not args:
+ self.args = split_command_line(command)
+ self.command = self.args[0]
+ else:
+ self.args = [command] + args
+ self.command = command
+
+ command_with_path = which(self.command)
+ 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 should be None.'
+ assert self.command is not None, 'The command member should not be None.'
+
+ kwargs = {'echo': self.echo, 'preexec_fn': preexec_fn}
+ if dimensions is not None:
+ kwargs['dimensions'] = dimensions
+
+ 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
+ return ptyprocess.PtyProcess.spawn(args, **kwargs)
def close(self, force=True):
"""This closes the connection with the child application. Note that
@@ -249,7 +281,12 @@ class spawn(SpawnBase):
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
+ if not self.closed:
+ self.flush()
+ self.ptyproc.close(force=force)
+ self.isalive() # Update exit status
+ self.child_fd = -1
+ self.closed = True
def isatty(self):
"""This returns True if the file descriptor is open and connected to a
@@ -259,7 +296,7 @@ class spawn(SpawnBase):
the child pty may not appear as a terminal device. This means
methods such as setecho(), setwinsize(), getwinsize() may raise an
IOError. """
- pass
+ return os.isatty(self.child_fd)
def waitnoecho(self, timeout=-1):
"""This waits until the terminal ECHO flag is set False. This returns
@@ -277,7 +314,16 @@ 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 is not None and time.time() > end_time:
+ return False
+ time.sleep(0.1)
def getecho(self):
"""This returns the terminal echo mode. This returns True if echo is
@@ -285,7 +331,9 @@ class spawn(SpawnBase):
to enter a password often set ECHO False. See waitnoecho().
Not supported on platforms where ``isatty()`` returns False. """
- pass
+ if not self.isatty():
+ raise IOError('getecho() failed: device is not a tty')
+ return self.ptyproc.getecho()
def setecho(self, state):
"""This sets the terminal echo mode on or off. Note that anything the
@@ -319,7 +367,9 @@ class spawn(SpawnBase):
Not supported on platforms where ``isatty()`` returns False.
"""
- pass
+ if not self.isatty():
+ raise IOError('setecho() failed: device is not a tty')
+ self.ptyproc.setecho(state)
def read_nonblocking(self, size=1, timeout=-1):
"""This reads at most size characters from the child application. It
@@ -346,7 +396,26 @@ class spawn(SpawnBase):
This is a wrapper around os.read(). It uses select.select() or
select.poll() to implement the timeout. """
- pass
+ if timeout == -1:
+ timeout = self.timeout
+
+ if not self.isalive():
+ raise EOF('End Of File (EOF). Child process has exited.')
+
+ if self.use_poll:
+ poller = poll_ignore_interrupts()
+ poller.register(self.child_fd, POLLIN)
+ else:
+ poller = None
+
+ try:
+ s, _ = self._read_nonblocking_impl(size, timeout, poller)
+ except TIMEOUT:
+ self.flag_eof = True
+ raise
+
+ self._log(s, 'read')
+ return s
def write(self, s):
"""This is similar to send() except that there is no return value.
diff --git a/pexpect/pxssh.py b/pexpect/pxssh.py
index b1c2a86..cde1ca1 100644
--- a/pexpect/pxssh.py
+++ b/pexpect/pxssh.py
@@ -130,7 +130,20 @@ class pxssh(spawn):
def levenshtein_distance(self, a, b):
"""This calculates the Levenshtein distance between a and b.
"""
- pass
+ if len(a) < len(b):
+ return self.levenshtein_distance(b, a)
+ if len(b) == 0:
+ return len(a)
+ previous_row = range(len(b) + 1)
+ for i, column1 in enumerate(a):
+ current_row = [i + 1]
+ for j, column2 in enumerate(b):
+ insertions = previous_row[j + 1] + 1
+ deletions = current_row[j] + 1
+ substitutions = previous_row[j] + (column1 != column2)
+ current_row.append(min(insertions, deletions, substitutions))
+ previous_row = current_row
+ return previous_row[-1]
def try_read_prompt(self, timeout_multiplier):
"""This facilitates using communication timeouts to perform
@@ -139,7 +152,20 @@ class pxssh(spawn):
should be read almost immediately. Worst case performance for this
method is timeout_multiplier * 3 seconds.
"""
- pass
+ timeout = self.timeout
+ pause = 0.1
+ max_attempts = 30
+ self.timeout = 0.1
+ for _ in range(int(max_attempts * timeout_multiplier)):
+ try:
+ self.read_nonblocking(size=1024, timeout=pause)
+ return True
+ except TIMEOUT:
+ time.sleep(pause)
+ except EOF:
+ return False
+ self.timeout = timeout
+ return False
def sync_original_prompt(self, sync_multiplier=1.0):
"""This attempts to find the prompt. Basically, press enter and record
@@ -205,14 +231,69 @@ class pxssh(spawn):
namespaces. For example ```cmd="ip netns exec vlan2 ssh"``` to execute the ssh in
network namespace named ```vlan```.
"""
- pass
+ if not spawn_local_ssh:
+ raise NotImplementedError("Non-local SSH spawning is not implemented")
+
+ ssh_options = ''
+ if ssh_key:
+ if isinstance(ssh_key, str):
+ ssh_options += f' -i {ssh_key}'
+ elif ssh_key is True:
+ ssh_options += ' -A'
+
+ if ssh_config:
+ ssh_options += f' -F {ssh_config}'
+
+ if port is not None:
+ ssh_options += f' -p {port}'
+
+ if username is not None:
+ server = f'{username}@{server}'
+
+ cmd = f'{cmd}{ssh_options} {server}'
+
+ if self.debug_command_string:
+ return cmd
+
+ spawn.__init__(self, cmd, timeout=login_timeout)
+
+ if not self.sync_original_prompt(sync_multiplier):
+ self.close()
+ raise ExceptionPxssh('Could not synchronize with original prompt')
+
+ if auto_prompt_reset:
+ if not self.set_unique_prompt():
+ self.close()
+ raise ExceptionPxssh('Could not set shell prompt')
+
+ if password:
+ self.waitnoecho()
+ self.sendline(password)
+ try:
+ i = self.expect([original_prompt, password_regex, TIMEOUT], timeout=login_timeout)
+ if i == 1:
+ self.close()
+ raise ExceptionPxssh('Password refused')
+ elif i == 2:
+ self.close()
+ raise ExceptionPxssh('Login timed out')
+ except EOF:
+ self.close()
+ raise ExceptionPxssh('Unexpected EOF')
+
+ 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.
@@ -230,7 +311,13 @@ 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 == 0:
+ return True
+ else:
+ return False
def set_unique_prompt(self):
"""This sets the remote prompt to something more unique than ``#`` or ``$``.
@@ -247,4 +334,14 @@ class pxssh(spawn):
:attr:`PROMPT` attribute to a regular expression. After that, the
:meth:`prompt` method will try to match your prompt pattern.
"""
- pass
+ self.sendline(self.PROMPT_SET_SH) # sh style
+ i = self.expect([TIMEOUT, self.PROMPT], timeout=10)
+ if i == 0: # timeout
+ self.sendline(self.PROMPT_SET_CSH) # csh style
+ i = self.expect([TIMEOUT, self.PROMPT], timeout=10)
+ if i == 0: # timeout
+ self.sendline(self.PROMPT_SET_ZSH) # zsh style
+ i = self.expect([TIMEOUT, self.PROMPT], timeout=10)
+ if i == 0: # timeout
+ return False
+ return True
diff --git a/pexpect/replwrap.py b/pexpect/replwrap.py
index 07a3d64..feb9375 100644
--- a/pexpect/replwrap.py
+++ b/pexpect/replwrap.py
@@ -65,19 +65,45 @@ class REPLWrapper(object):
:mod:`asyncio` Future, which you can yield from to get the same
result that this method would normally give directly.
"""
- pass
+ if async_:
+ import asyncio
+ return asyncio.ensure_future(self._run_command_async(command, timeout))
+
+ self.child.sendline(command)
+ self._expect_prompt(timeout=timeout)
+
+ # Remove the echoed command and the final prompt
+ return self.child.before.strip()
+
+ async def _run_command_async(self, command, timeout):
+ self.child.sendline(command)
+ await self._expect_prompt_async(timeout=timeout)
+ return self.child.before.strip()
+
+ def _expect_prompt(self, timeout=-1):
+ return self.child.expect([self.prompt, self.continuation_prompt], timeout=timeout)
+
+ async def _expect_prompt_async(self, timeout=-1):
+ return await self.child.expect_async([self.prompt, self.continuation_prompt], timeout=timeout)
def python(command=sys.executable):
"""Start a Python shell and return a :class:`REPLWrapper` object."""
- pass
+ orig_prompt = '>>>'
+ prompt_change = 'import sys; sys.ps1={0!r}; sys.ps2={1!r}'
+ return REPLWrapper(command, orig_prompt, prompt_change)
def bash(command='bash'):
"""Start a bash shell and return a :class:`REPLWrapper` object."""
- pass
+ orig_prompt = r'[$#] '
+ prompt_change = "PS1='{0}'; PS2='{1}'"
+ return REPLWrapper(command, orig_prompt, prompt_change)
def zsh(command='zsh', args=('--no-rcs', '-V', '+Z')):
"""Start a zsh shell and return a :class:`REPLWrapper` object."""
- pass
+ orig_prompt = r'[%#] '
+ prompt_change = "PROMPT='{0}'; PROMPT2='{1}'"
+ cmd = [command] + list(args)
+ return REPLWrapper(' '.join(cmd), orig_prompt, prompt_change)
diff --git a/pexpect/run.py b/pexpect/run.py
index 8710741..c519007 100644
--- a/pexpect/run.py
+++ b/pexpect/run.py
@@ -91,11 +91,63 @@ def run(command, timeout=30, withexitstatus=False, events=None, extra_args=
instead of bytes. You can pass *codec_errors* to control how errors in
encoding and decoding are handled.
"""
- pass
+ child = spawn(command, timeout=timeout, maxread=2000, logfile=logfile, cwd=cwd, env=env, **kwargs)
+
+ if events is None:
+ events = {}
+ if extra_args is None:
+ extra_args = {}
+
+ output = []
+ event_count = 0
+
+ while True:
+ try:
+ index = child.expect(list(events.keys()) + [EOF, TIMEOUT], timeout=timeout)
+ except EOF:
+ output.append(child.before)
+ break
+ except TIMEOUT:
+ output.append(child.before)
+ break
+
+ output.append(child.before)
+ event_count += 1
+
+ if index < len(events):
+ event = list(events.values())[index]
+ if isinstance(event, str):
+ child.sendline(event)
+ elif callable(event):
+ callback_result = event({
+ 'child': child,
+ 'event_count': event_count,
+ 'extra_args': extra_args
+ })
+ if callback_result is True:
+ break
+ elif isinstance(callback_result, str):
+ child.sendline(callback_result)
+
+ child.close()
+ exitstatus = child.exitstatus
+
+ output = ''.join(output)
+
+ if withexitstatus:
+ return (output, exitstatus)
+ else:
+ return output
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
+ import warnings
+ warnings.warn("runu() is deprecated. Use run() with encoding='utf-8' instead.",
+ DeprecationWarning, stacklevel=2)
+
+ kwargs['encoding'] = kwargs.get('encoding', 'utf-8')
+ return run(command, timeout, withexitstatus, events, extra_args,
+ logfile, cwd, env, **kwargs)
diff --git a/pexpect/screen.py b/pexpect/screen.py
index c1cb2a3..c87bd63 100644
--- a/pexpect/screen.py
+++ b/pexpect/screen.py
@@ -54,7 +54,7 @@ if PY3:
def constrain(n, min, max):
"""This returns a number, n constrained to the min and max bounds. """
- pass
+ return min if n < min else max if n > max else n
class screen:
@@ -96,13 +96,15 @@ class screen:
def _decode(self, s):
"""This converts from the external coding system (as passed to
the constructor) to the internal one (unicode). """
- pass
+ if self.decoder is None:
+ return s
+ return self.decoder.decode(s)
def _unicode(self):
"""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
+ return '\n'.join([''.join(row) for row in self.w])
if PY3:
__str__ = _unicode
else:
@@ -118,137 +120,181 @@ class screen:
"""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
+ return ''.join([''.join(row) for row in self.w])
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
+ top_border = '+' + '-' * self.cols + '+\n'
+ bottom_border = '\n+' + '-' * self.cols + '+'
+ screen_content = '\n'.join(['|' + ''.join(row) + '|' for row in self.w])
+ return top_border + screen_content + bottom_border
def cr(self):
"""This moves the cursor to the beginning (col 1) of the current row.
"""
- pass
+ self.cur_c = 1
def lf(self):
"""This moves the cursor down with scrolling.
"""
- pass
+ if self.cur_r == self.scroll_row_end:
+ self.scroll_up()
+ else:
+ self.cur_r = constrain(self.cur_r + 1, 1, self.rows)
def crlf(self):
"""This advances the cursor with CRLF properties.
The cursor will line wrap and the screen may scroll.
"""
- pass
+ self.cr()
+ self.lf()
def newline(self):
"""This is an alias for crlf().
"""
- pass
+ self.crlf()
def put_abs(self, r, c, ch):
"""Screen array starts at 1 index."""
- pass
+ r = constrain(r, 1, self.rows)
+ c = constrain(c, 1, self.cols)
+ self.w[r-1][c-1] = ch
def put(self, ch):
"""This puts a characters at the current cursor position.
"""
- pass
+ self.put_abs(self.cur_r, self.cur_c, ch)
+ self.cur_c = constrain(self.cur_c + 1, 1, self.cols)
+ if self.cur_c == 1:
+ self.lf()
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
+ r = constrain(r, 1, self.rows)
+ c = constrain(c, 1, self.cols)
+ self.w[r-1] = self.w[r-1][:c-1] + [ch] + self.w[r-1][c-1:-1]
def get_region(self, rs, cs, re, ce):
"""This returns a list of lines representing the region.
"""
- 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)
+ return [''.join(row[cs-1:ce]) for row in self.w[rs-1:re]]
def cursor_constrain(self):
"""This keeps the cursor within the screen area.
"""
- pass
+ self.cur_r = constrain(self.cur_r, 1, self.rows)
+ self.cur_c = constrain(self.cur_c, 1, self.cols)
def cursor_force_position(self, r, c):
"""Identical to Cursor Home."""
- pass
+ self.cur_r = constrain(r, 1, self.rows)
+ self.cur_c = constrain(c, 1, self.cols)
def cursor_save(self):
"""Save current cursor position."""
- pass
+ self.cur_saved_r = self.cur_r
+ self.cur_saved_c = self.cur_c
def cursor_unsave(self):
"""Restores cursor position after a Save Cursor."""
- pass
+ self.cur_r = self.cur_saved_r
+ self.cur_c = self.cur_saved_c
def cursor_save_attrs(self):
"""Save current cursor position."""
- pass
+ self.cursor_save()
def cursor_restore_attrs(self):
"""Restores cursor position after a Save Cursor."""
- pass
+ self.cursor_unsave()
def scroll_constrain(self):
"""This keeps the scroll region within the screen region."""
- pass
+ self.scroll_row_start = constrain(self.scroll_row_start, 1, self.rows)
+ self.scroll_row_end = constrain(self.scroll_row_end, 1, self.rows)
+ if self.scroll_row_start > self.scroll_row_end:
+ self.scroll_row_start, self.scroll_row_end = self.scroll_row_end, self.scroll_row_start
def scroll_screen(self):
"""Enable scrolling for entire display."""
- pass
+ self.scroll_row_start = 1
+ self.scroll_row_end = self.rows
def scroll_screen_rows(self, rs, re):
"""Enable scrolling from row {start} to row {end}."""
- pass
+ self.scroll_row_start = constrain(rs, 1, self.rows)
+ self.scroll_row_end = constrain(re, 1, self.rows)
+ self.scroll_constrain()
def scroll_down(self):
"""Scroll display down one line."""
- pass
+ s = self.scroll_row_start - 1
+ e = self.scroll_row_end
+ self.w[s+1:e] = self.w[s:e-1]
+ self.w[s] = [SPACE] * self.cols
def scroll_up(self):
"""Scroll display up one line."""
- pass
+ s = self.scroll_row_start - 1
+ e = self.scroll_row_end
+ self.w[s:e-1] = self.w[s+1:e]
+ self.w[e-1] = [SPACE] * self.cols
def erase_end_of_line(self):
"""Erases from the current cursor position to the end of the current
line."""
- pass
+ self.w[self.cur_r-1][self.cur_c-1:] = [SPACE] * (self.cols - self.cur_c + 1)
def erase_start_of_line(self):
"""Erases from the current cursor position to the start of the current
line."""
- pass
+ self.w[self.cur_r-1][:self.cur_c] = [SPACE] * self.cur_c
def erase_line(self):
"""Erases the entire current line."""
- pass
+ self.w[self.cur_r-1] = [SPACE] * self.cols
def erase_down(self):
"""Erases the screen from the current line down to the bottom of the
screen."""
- pass
+ self.erase_end_of_line()
+ for r in range(self.cur_r, self.rows):
+ self.w[r] = [SPACE] * self.cols
def erase_up(self):
"""Erases the screen from the current line up to the top of the
screen."""
- pass
+ self.erase_start_of_line()
+ for r in range(self.cur_r-1):
+ self.w[r] = [SPACE] * self.cols
def erase_screen(self):
"""Erases the screen with the background color."""
- pass
+ self.w = [[SPACE] * self.cols for _ in range(self.rows)]
def set_tab(self):
"""Sets a tab at the current position."""
+ # This method is not implemented in the original code
+ # and would require additional state to track tab positions
pass
def clear_tab(self):
"""Clears tab at the current position."""
+ # This method is not implemented in the original code
+ # and would require additional state to track tab positions
pass
def clear_all_tabs(self):
"""Clears all tabs."""
+ # This method is not implemented in the original code
+ # and would require additional state to track tab positions
pass
diff --git a/pexpect/socket_pexpect.py b/pexpect/socket_pexpect.py
index 5d7ca43..2014681 100644
--- a/pexpect/socket_pexpect.py
+++ b/pexpect/socket_pexpect.py
@@ -51,27 +51,36 @@ 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 not self.closed:
+ self.socket.close()
+ self.closed = True
def isalive(self):
""" Alive if the fileno is valid """
- pass
+ return not self.closed and self.socket.fileno() != -1
def send(self, s) ->int:
"""Write to socket, return number of bytes written"""
- pass
+ if isinstance(s, str):
+ s = s.encode(self.encoding)
+ return self.socket.send(s)
def sendline(self, s) ->int:
"""Write to socket with trailing newline, return number of bytes written"""
- pass
+ if isinstance(s, str):
+ s = s.encode(self.encoding)
+ return self.socket.send(s + b'\n')
def write(self, s):
"""Write to socket, return None"""
- pass
+ if isinstance(s, str):
+ s = s.encode(self.encoding)
+ self.socket.sendall(s)
def writelines(self, sequence):
"""Call self.write() for each item in sequence"""
- pass
+ for item in sequence:
+ self.write(item)
def read_nonblocking(self, size=1, timeout=-1):
"""
@@ -89,4 +98,20 @@ class SocketSpawn(SpawnBase):
ready to read. When -1 (default), use self.timeout. When 0, poll.
:return: String containing the bytes read
"""
- pass
+ import select
+
+ if timeout == -1:
+ timeout = self.timeout
+
+ try:
+ ready, _, _ = select.select([self.socket], [], [], timeout)
+ if not ready:
+ raise TIMEOUT('Timeout exceeded')
+
+ data = self.socket.recv(size)
+ if not data:
+ raise EOF('End of file')
+
+ return self.decoder.decode(data, final=False)
+ except socket.error as e:
+ raise EOF('Connection closed: %s' % str(e))
diff --git a/pexpect/spawnbase.py b/pexpect/spawnbase.py
index abe78e6..1402d45 100644
--- a/pexpect/spawnbase.py
+++ b/pexpect/spawnbase.py
@@ -101,7 +101,12 @@ class SpawnBase(object):
The timeout parameter is ignored.
"""
- pass
+ try:
+ return os.read(self.child_fd, size)
+ except OSError as e:
+ if e.errno == errno.EAGAIN:
+ return None
+ raise
def compile_pattern_list(self, patterns):
"""This compiles a pattern-string or a list of pattern-strings.
@@ -126,7 +131,25 @@ class SpawnBase(object):
i = self.expect_list(cpl, timeout)
...
"""
- pass
+ if patterns is None:
+ return []
+ if not isinstance(patterns, list):
+ patterns = [patterns]
+
+ compiled_pattern_list = []
+ for p in patterns:
+ if isinstance(p, (str, bytes)):
+ compiled_pattern_list.append(re.compile(p, re.DOTALL))
+ elif p is EOF:
+ compiled_pattern_list.append(EOF)
+ elif p is TIMEOUT:
+ compiled_pattern_list.append(TIMEOUT)
+ elif isinstance(p, type(re.compile(''))):
+ compiled_pattern_list.append(p)
+ else:
+ raise TypeError('Unsupported pattern type: %s' % type(p))
+
+ return compiled_pattern_list
def expect(self, pattern, timeout=-1, searchwindowsize=-1, async_=False,
**kw):
@@ -223,7 +246,12 @@ class SpawnBase(object):
index = yield from p.expect(patterns, async_=True)
"""
- pass
+ compiled_pattern_list = self.compile_pattern_list(pattern)
+ return self.expect_list(compiled_pattern_list,
+ timeout=timeout,
+ searchwindowsize=searchwindowsize,
+ async_=async_,
+ **kw)
def expect_list(self, pattern_list, timeout=-1, searchwindowsize=-1,
async_=False, **kw):
@@ -239,7 +267,16 @@ class SpawnBase(object):
Like :meth:`expect`, passing ``async_=True`` will make this return an
asyncio coroutine.
"""
- pass
+ if timeout == -1:
+ timeout = self.timeout
+ if searchwindowsize == -1:
+ searchwindowsize = self.searchwindowsize
+
+ exp = Expecter(self, searcher_re(pattern_list), searchwindowsize)
+ if async_:
+ return exp.expect_async(timeout, **kw)
+ else:
+ return exp.expect_loop(timeout)
def expect_exact(self, pattern_list, timeout=-1, searchwindowsize=-1,
async_=False, **kw):
diff --git a/pexpect/utils.py b/pexpect/utils.py
index 960e622..4aad058 100644
--- a/pexpect/utils.py
+++ b/pexpect/utils.py
@@ -19,14 +19,22 @@ def is_executable_file(path):
This is roughly ``os.path isfile(path) and os.access(path, os.X_OK)``.
"""
- pass
+ return os.path.isfile(path) and os.access(path, os.X_OK)
def which(filename, env=None):
"""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 env is None:
+ env = os.environ
+
+ path = env.get('PATH', '')
+ for directory in path.split(os.pathsep):
+ full_path = os.path.join(directory, filename)
+ if is_executable_file(full_path):
+ return full_path
+ return None
def split_command_line(command_line):
@@ -34,7 +42,37 @@ def split_command_line(command_line):
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
+ args = []
+ current_arg = ''
+ state = 'normal'
+ quote_char = None
+
+ for char in command_line:
+ if state == 'normal':
+ if char.isspace():
+ if current_arg:
+ args.append(current_arg)
+ current_arg = ''
+ elif char in ('"', "'"):
+ state = 'in_quote'
+ quote_char = char
+ elif char == '\\':
+ state = 'escaped'
+ else:
+ current_arg += char
+ elif state == 'in_quote':
+ if char == quote_char:
+ state = 'normal'
+ else:
+ current_arg += char
+ elif state == 'escaped':
+ current_arg += char
+ state = 'normal'
+
+ if current_arg:
+ args.append(current_arg)
+
+ return args
def select_ignore_interrupts(iwtd, owtd, ewtd, timeout=None):
@@ -42,10 +80,24 @@ def select_ignore_interrupts(iwtd, owtd, ewtd, timeout=None):
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
+ while True:
+ try:
+ return select.select(iwtd, owtd, ewtd, timeout)
+ except (select.error, InterruptedError) as e:
+ if e.args[0] != errno.EINTR:
+ raise
def poll_ignore_interrupts(fds, timeout=None):
"""Simple wrapper around poll to register file descriptors and
ignore signals."""
- pass
+ p = select.poll()
+ for fd in fds:
+ p.register(fd, select.POLLIN)
+
+ while True:
+ try:
+ return p.poll(timeout)
+ except (select.error, InterruptedError) as e:
+ if e.args[0] != errno.EINTR:
+ raise