tessl/pypi-pexpect

Pure Python module for spawning child applications and controlling them automatically with expect-like functionality.

—

Pending

Overview

Eval results

Files

Process Spawning and Control

Name: tessl/pypi-pexpect
Author: tessl

Core functionality for spawning and controlling child processes using pseudo-terminals. This module provides the fundamental classes and functions for process automation in pexpect.

Capabilities

Main Spawn Class

The primary interface for spawning and controlling child processes with pseudo-terminal (PTY) support.

class spawn(SpawnBase):
    """
    Main class interface for Pexpect. Use this class to start and control child applications.
    
    Parameters:
    - command (str): Command to execute, can include arguments
    - args (list): Command arguments as separate list items
    - timeout (int): Default timeout in seconds for expect operations
    - maxread (int): Maximum number of bytes to read at once
    - searchwindowsize (int): Number of bytes to keep in search window
    - logfile (file-like): File object to log all I/O
    - cwd (str): Working directory for child process
    - env (dict): Environment variables for child process
    - ignore_sighup (bool): Ignore SIGHUP signal
    - echo (bool): Enable/disable terminal echo
    - preexec_fn (callable): Function to call before exec in child
    - encoding (str): Text encoding for I/O operations
    - codec_errors (str): How to handle encoding errors
    - dimensions (tuple): Terminal size as (rows, cols)
    - use_poll (bool): Use poll() instead of select()
    """
    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): ...

Process Lifecycle Methods

def close(self, force=True):
    """
    Close the connection with the child application.
    
    Parameters:
    - force (bool): Force termination if child doesn't close gracefully
    """

def terminate(self, force=False):
    """
    Terminate the child process.
    
    Parameters:
    - force (bool): Send SIGKILL instead of SIGTERM
    """

def kill(self, sig):
    """
    Send signal to the child process.
    
    Parameters:
    - sig (int): Signal number to send
    """

def wait(self):
    """
    Wait for the child process to exit.
    
    Returns:
    int: Exit status of child process
    """

def isalive(self):
    """
    Check if child process is still running.
    
    Returns:
    bool: True if child is alive, False otherwise
    """

Input/Output Methods

def send(self, s):
    """
    Send string to child process.
    
    Parameters:
    - s (str or bytes): Data to send
    
    Returns:
    int: Number of bytes written
    """

def sendline(self, s=''):
    """
    Send string followed by line terminator to child.
    
    Parameters:
    - s (str): String to send (default: empty string)
    
    Returns:
    int: Number of bytes written
    """

def sendcontrol(self, char):
    """
    Send control character to child.
    
    Parameters:
    - char (str): Control character (e.g., 'c' for Ctrl-C)
    
    Returns:
    int: Number of bytes written
    """

def sendeof(self):
    """
    Send EOF to child process.
    """

def sendintr(self):
    """
    Send interrupt signal (Ctrl-C) to child.
    """

def read_nonblocking(self, size=1, timeout=-1):
    """
    Read data from child without blocking.
    
    Parameters:
    - size (int): Maximum bytes to read
    - timeout (int): Timeout in seconds (-1 for default)
    
    Returns:
    bytes: Data read from child
    
    Raises:
    - TIMEOUT: If timeout exceeded
    - EOF: If child has terminated
    """

def write(self, s):
    """
    Write data to child (no return value).
    
    Parameters:
    - s (str or bytes): Data to write
    """

def writelines(self, sequence):
    """
    Write sequence of strings to child.
    
    Parameters:
    - sequence (iterable): Sequence of strings to write
    """

def read(self, size=-1):
    """
    Read and return up to size bytes from child.
    
    Parameters:
    - size (int): Maximum bytes to read (-1 for all available)
    
    Returns:
    bytes or str: Data read from child
    """

def readline(self, size=-1):
    """
    Read and return one line from child.
    
    Parameters:
    - size (int): Maximum bytes to read (-1 for unlimited)
    
    Returns:
    bytes or str: One line from child
    """

def readlines(self, sizehint=-1):
    """
    Read and return list of lines from child.
    
    Parameters:
    - sizehint (int): Hint for number of bytes to read
    
    Returns:
    list: List of lines from child
    """

def flush(self):
    """
    Flush the write buffers (if applicable).
    """

def fileno(self):
    """
    Return file descriptor number for the child.
    
    Returns:
    int: File descriptor number
    """

Terminal Control Methods

def setecho(self, state):
    """
    Enable or disable terminal echo.
    
    Parameters:
    - state (bool): True to enable echo, False to disable
    """

def getecho(self):
    """
    Get current terminal echo state.
    
    Returns:
    bool: Current echo state
    """

def waitnoecho(self, timeout=-1):
    """
    Wait for terminal echo to be disabled.
    
    Parameters:
    - timeout (int): Timeout in seconds
    
    Returns:
    bool: True if echo disabled, False on timeout
    """

def getwinsize(self):
    """
    Get terminal window size.
    
    Returns:
    tuple: (rows, cols) terminal dimensions
    """

def setwinsize(self, rows, cols):
    """
    Set terminal window size.
    
    Parameters:
    - rows (int): Number of rows
    - cols (int): Number of columns
    """

def isatty(self):
    """
    Check if connected to a TTY.
    
    Returns:
    bool: True if connected to TTY
    """

def interact(self, escape_character=chr(29), input_filter=None, output_filter=None):
    """
    Give control of child to user for interactive session.
    
    Parameters:
    - escape_character (str): Character to exit interaction (default: Ctrl-])
    - input_filter (callable): Filter function for user input
    - output_filter (callable): Filter function for child output
    """

def eof(self):
    """
    Check if EOF exception was ever raised.
    
    Returns:
    bool: True if EOF has been encountered
    """

@property
def flag_eof(self):
    """
    Boolean flag indicating if EOF has been encountered.
    
    Returns:
    bool: Current EOF flag state
    """

High-Level Run Function

Simple function interface for executing commands and capturing output.

def run(command, timeout=30, withexitstatus=False, events=None,
        extra_args=None, logfile=None, cwd=None, env=None, **kwargs):
    """
    Execute command and return output.
    
    Parameters:
    - command (str): Command to execute
    - timeout (int): Timeout in seconds
    - withexitstatus (bool): Return (output, exitstatus) tuple
    - events (dict/list): Pattern-response mappings for automation
    - extra_args (list): Additional arguments for spawn
    - logfile (file-like): File object for logging
    - cwd (str): Working directory
    - env (dict): Environment variables
    - **kwargs: Additional arguments passed to spawn
    
    Returns:
    str or tuple: Command output, or (output, exitstatus) if withexitstatus=True
    """

Legacy Unicode Wrappers

def spawnu(command, args=[], timeout=30, maxread=2000, searchwindowsize=None,
           logfile=None, cwd=None, env=None, ignore_sighup=False, echo=True,
           preexec_fn=None, encoding='utf-8', codec_errors='strict',
           dimensions=None, use_poll=False):
    """
    Deprecated: Use spawn with encoding parameter instead.
    Unicode wrapper for spawn class.
    """

def runu(command, timeout=30, withexitstatus=False, events=None,
         extra_args=None, logfile=None, cwd=None, env=None, **kwargs):
    """
    Deprecated: Use run with encoding parameter instead.
    Unicode wrapper for run function.
    """

Key Attributes

All spawn instances provide these important attributes:

# Process information
pid: int  # Child process ID
child_fd: int  # File descriptor for child
exitstatus: int  # Exit code (None if running)
signalstatus: int  # Signal that killed process (None if not killed by signal)

# Pattern matching results
before: str  # Text before matched pattern
after: str  # Text that matched pattern  
match: Match  # Regex match object (for regex patterns)
match_index: int  # Index of matched pattern in pattern list

# Configuration
timeout: int  # Current default timeout
logfile: file  # Main log file
logfile_read: file  # Log file for read operations
logfile_send: file  # Log file for send operations

Usage Examples

Basic Process Control

import pexpect

# Spawn a process
child = pexpect.spawn('ssh user@server')

# Check if process is running
if child.isalive():
    print(f"Process {child.pid} is running")

# Send commands
child.sendline('ls -la')
child.expect('$ ')

# Read output
output = child.before
print(output.decode())

# Clean shutdown
child.sendline('exit')
child.close()

Using run() for Simple Commands

import pexpect

# Simple command execution
output = pexpect.run('ls -la /tmp')
print(output.decode())

# Get exit status
output, exitstatus = pexpect.run('ls /nonexistent', withexitstatus=True)
print(f"Exit status: {exitstatus}")

# Automated interaction
events = {
    'Password:': 'mypassword\n',
    'Are you sure': 'yes\n'
}
output = pexpect.run('ssh user@server', events=events)

Context Manager Usage

import pexpect

# Automatic cleanup with context manager
with pexpect.spawn('ftp ftp.example.com') as child:
    child.expect('Name:')
    child.sendline('anonymous')
    child.expect('Password:')
    child.sendline('user@example.com')
    child.expect('ftp>')
    child.sendline('quit')
# Child is automatically closed

Install with Tessl CLI