Process Automation with Pexpect

In this tutorial, you will learn how to automate interactive command-line applications using the pexpect module. This powerful module allows you to spawn processes, send input, and wait for expected output patterns, making it ideal for automating tasks that require human-like interaction.

What You Will Learn

• Spawning and controlling child processes
• Pattern matching with expect() and expectExact()
• Handling timeouts and EOF conditions
• Sending input with send(), sendline(), and control characters
• Reading process output with various strategies
• Managing process lifecycle and cleanup
• Building real-world automation workflows

When to Use Pexpect

Use the pexpect module when you need to:

• Automate interactive CLI tools that require user input
• Handle password prompts and authentication flows
• Test interactive applications
• Script FTP, SSH, or other protocol clients
• Control REPLs (Python, Node, databases)

For simple command execution where you just need stdout/stderr, use the subprocess module instead.

Part 1: Basic Concepts

Step 1: Your First Spawn

The Spawn class creates and manages a child process:

import "pexpect" for Spawn

var child = Spawn.new("cat")
child.sendline("Hello, World!")
child.sendeof()
child.wait()
child.close()

Step 2: The Expect-Send Pattern

The core workflow with pexpect follows the expect-send pattern:

import "pexpect" for Spawn

var child = Spawn.new("python3")

child.expect(">>>")
child.sendline("print('Hello from Python!')")

child.expect(">>>")
System.print("Output: %(child.before)")

child.sendline("exit()")
child.wait()
child.close()

Step 3: Understanding Buffers

The Spawn class maintains three important buffers:

Step 4: Pexpect Constants

The Pexpect class provides constants for special conditions:

import "pexpect" for Pexpect, Spawn

System.print("EOF constant: %(Pexpect.EOF)")
System.print("TIMEOUT constant: %(Pexpect.TIMEOUT)")

Part 2: Pattern Matching

Step 5: Single Pattern Matching

import "pexpect" for Spawn

var child = Spawn.new("python3")

child.expect(">>>")
child.sendline("2 + 2")

child.expect(">>>")
System.print(child.before)

child.sendeof()
child.close()

Step 6: Multiple Pattern Matching

Pass a list to match multiple patterns. The return value indicates which pattern matched:

import "pexpect" for Pexpect, Spawn

var child = Spawn.new("bash -c 'echo success'")

var idx = child.expect(["success", "failure", "error"])

if (idx == 0) {
    System.print("Operation succeeded!")
} else if (idx == 1) {
    System.print("Operation failed")
} else if (idx == 2) {
    System.print("Error occurred")
}

child.close()

Step 7: Regex Pattern Matching

The expect() method supports regular expressions:

import "pexpect" for Spawn

var child = Spawn.new("python3")
child.expect(">>>")

child.sendline("import sys; print(sys.version)")

child.expect("\\d+\\.\\d+\\.\\d+")
System.print("Python version: %(child.after)")

child.sendeof()
child.close()

Step 8: Exact String Matching

Use expectExact() to match literal strings without regex interpretation:

import "pexpect" for Spawn

var child = Spawn.new("echo 'test[123]'")

child.expectExact("[123]")
System.print("Found literal brackets: %(child.after)")

child.close()

Step 9: Matching EOF and TIMEOUT

import "pexpect" for Pexpect, Spawn

var child = Spawn.new("echo 'done'")

var idx = child.expect(["done", Pexpect.EOF, Pexpect.TIMEOUT])

if (idx == 0) {
    System.print("Found 'done'")
} else if (idx == Pexpect.EOF) {
    System.print("Process ended")
} else if (idx == Pexpect.TIMEOUT) {
    System.print("Timed out waiting")
}

child.close()

Part 3: Timeout Handling

Step 10: Default Timeout Configuration

import "pexpect" for Spawn

var child = Spawn.new("python3")
child.timeout = 10

System.print("Current timeout: %(child.timeout) seconds")

child.expect(">>>")
child.sendeof()
child.close()

Step 11: Per-Operation Timeouts

import "pexpect" for Spawn

var child = Spawn.new("bash")
child.timeout = 30

child.expect("\\$", 5)
child.sendline("sleep 2 && echo done")

child.expect("done", 10)
System.print("Long operation completed")

child.sendeof()
child.close()

Step 12: Handling Timeout Returns

import "pexpect" for Pexpect, Spawn

var child = Spawn.new("bash")
child.timeout = 2

child.expect("\\$")
child.sendline("sleep 10")

var idx = child.expect(["complete", Pexpect.TIMEOUT])

if (idx == Pexpect.TIMEOUT) {
    System.print("Operation timed out, terminating process")
    child.terminate()
}

child.close()

Part 4: Input Methods

Step 13: send() vs sendline()

import "pexpect" for Spawn

var child = Spawn.new("cat")

child.send("Hello ")
child.send("World")
child.sendline("!")

child.sendeof()
child.wait()
child.close()

Step 14: Control Characters

import "pexpect" for Spawn

var child = Spawn.new("bash")

child.expect("\\$")
child.sendline("cat")

child.sendline("This is some text")
child.sendcontrol("c")

child.expect("\\$")
System.print("Interrupted cat command")

child.sendeof()
child.close()

Common control characters:

• sendcontrol("c") - Interrupt (SIGINT)
• sendcontrol("d") - End of file (EOF)
• sendcontrol("z") - Suspend (SIGTSTP)
• sendcontrol("l") - Clear screen

Step 15: Writing Multiple Lines

import "pexpect" for Spawn

var child = Spawn.new("cat")

child.writelines([
    "Line 1\n",
    "Line 2\n",
    "Line 3\n"
])

child.sendeof()
child.wait()
child.close()

Part 5: Reading Strategies

Step 16: Non-Blocking Reads

import "pexpect" for Spawn

var child = Spawn.new("bash -c 'echo line1; sleep 0.1; echo line2'")

var output = ""
while (child.isalive || child.buffer.count > 0) {
    var data = child.readNonblocking(1024, 0.1)
    if (data != null && data.count > 0) {
        output = output + data
        System.print("Read: %(data.trim())")
    }
}

child.close()

Step 17: Line-Based Reading

import "pexpect" for Spawn

var child = Spawn.new("bash -c 'echo one; echo two; echo three'")

var line1 = child.readline()
var line2 = child.readline()
var line3 = child.readline()

System.print("Line 1: %(line1)")
System.print("Line 2: %(line2)")
System.print("Line 3: %(line3)")

child.close()

Part 6: Process Lifecycle

Step 18: Checking Process State

import "pexpect" for Spawn

var child = Spawn.new("sleep 1")

System.print("Is alive: %(child.isalive)")
System.print("PID: %(child.pid)")

child.wait()

System.print("Is alive after wait: %(child.isalive)")
System.print("Exit status: %(child.exitstatus)")

child.close()

Step 19: Graceful Termination

import "pexpect" for Spawn

var child = Spawn.new("sleep 60")

System.print("Process started, PID: %(child.pid)")

child.terminate()
System.print("Terminate signal sent")

child.wait()
System.print("Terminated: %(child.terminated)")

child.close()

Step 20: Force Termination

import "pexpect" for Spawn

var child = Spawn.new("bash -c 'trap \"\" SIGTERM; sleep 60'")

System.print("Started process that ignores SIGTERM")

child.terminate(false)

if (child.isalive) {
    System.print("Process still alive, forcing kill")
    child.terminate(true)
}

child.wait()
child.close()

Step 21: Sending Signals

import "pexpect" for Spawn

var child = Spawn.new("bash -c 'trap \"echo caught\" SIGUSR1; sleep 10'")

child.readNonblocking(100, 0.5)

child.kill(10)

child.expect("caught")
System.print("Signal was caught!")

child.terminate()
child.close()

Part 7: Advanced Features

Step 22: Echo Control

import "pexpect" for Spawn

var child = Spawn.new("bash")

child.expect("\\$")

child.setecho(false)
child.sendline("echo 'This should not echo'")

child.setecho(true)
child.expect("\\$")

child.sendeof()
child.close()

Step 23: Interactive Mode

import "pexpect" for Spawn

var child = Spawn.new("bash")

child.expect("\\$")
System.print("Entering interactive mode. Press Ctrl+] to exit.")

child.interact(29)

child.close()

Step 24: Logging Output

import "pexpect" for Spawn

var child = Spawn.new("bash")

child.logfile = "pexpect.log"
child.logfileRead = "pexpect_read.log"
child.logfileSend = "pexpect_send.log"

child.expect("\\$")
child.sendline("echo 'Logged output'")
child.expect("\\$")

child.sendeof()
child.close()

System.print("Logs written to pexpect*.log files")

Step 25: Performance Tuning

import "pexpect" for Spawn

var child = Spawn.new("bash")

child.maxread = 8192
child.searchwindowsize = 2000
child.delaybeforesend = 0.05
child.delayafterclose = 0.1

System.print("Max read: %(child.maxread)")
System.print("Search window: %(child.searchwindowsize)")

child.expect("\\$")
child.sendeof()
child.close()

Part 8: Real-World Examples

Example 1: Automating a Python REPL

import "pexpect" for Pexpect, Spawn

class PythonRepl {
    construct new() {
        _child = Spawn.new("python3")
        _child.timeout = 10
        _child.expect(">>>")
    }

    execute(code) {
        _child.sendline(code)
        var idx = _child.expect([">>>", "\\.\\.\\.", Pexpect.TIMEOUT])

        if (idx == Pexpect.TIMEOUT) {
            return {"error": "Execution timed out"}
        }

        return {
            "output": _child.before.trim(),
            "continued": idx == 1
        }
    }

    close() {
        _child.sendeof()
        _child.wait()
        _child.close()
    }
}

var repl = PythonRepl.new()

var result = repl.execute("2 + 2")
System.print("Result: %(result["output"])")

result = repl.execute("import math")
result = repl.execute("math.pi")
System.print("Pi: %(result["output"])")

repl.close()

Example 2: SSH-like Authentication Flow

import "pexpect" for Pexpect, Spawn

class SecureConnection {
    construct new(host, username, password) {
        _host = host
        _username = username
        _password = password
        _child = null
    }

    connect() {
        _child = Spawn.new("ssh %(_username)@%(_host)")
        _child.timeout = 30

        var idx = _child.expect([
            "[Pp]assword:",
            "yes/no",
            "Connection refused",
            Pexpect.TIMEOUT
        ])

        if (idx == 1) {
            _child.sendline("yes")
            _child.expect("[Pp]assword:")
            idx = 0
        }

        if (idx == 0) {
            _child.setecho(false)
            _child.sendline(_password)
            _child.setecho(true)

            idx = _child.expect(["\\$", "#", "Permission denied"])
            if (idx == 2) {
                return {"success": false, "error": "Authentication failed"}
            }
            return {"success": true}
        }

        if (idx == 2) {
            return {"success": false, "error": "Connection refused"}
        }

        return {"success": false, "error": "Connection timed out"}
    }

    execute(command) {
        _child.sendline(command)
        _child.expect(["\\$", "#"])
        return _child.before.trim()
    }

    close() {
        if (_child != null) {
            _child.sendline("exit")
            _child.wait()
            _child.close()
        }
    }
}

System.print("SSH automation example (conceptual)")

Example 3: FTP File Transfer Automation

import "pexpect" for Pexpect, Spawn

class FtpClient {
    construct new(host) {
        _child = Spawn.new("ftp %(host)")
        _child.timeout = 30
    }

    login(username, password) {
        _child.expect("Name")
        _child.sendline(username)

        _child.expect("[Pp]assword:")
        _child.sendline(password)

        var idx = _child.expect(["230", "530", Pexpect.TIMEOUT])

        if (idx == 0) {
            _child.expect("ftp>")
            return true
        }
        return false
    }

    cd(directory) {
        _child.sendline("cd %(directory)")
        var idx = _child.expect(["250", "550", "ftp>"])
        return idx == 0
    }

    ls() {
        _child.sendline("ls")
        _child.expect("ftp>")
        return _child.before
    }

    get(filename) {
        _child.sendline("get %(filename)")
        var idx = _child.expect(["226", "550", "ftp>"])
        return idx == 0
    }

    quit() {
        _child.sendline("quit")
        _child.wait()
        _child.close()
    }
}

System.print("FTP automation example (conceptual)")

Example 4: System Administration with sudo

import "pexpect" for Pexpect, Spawn

class SudoCommand {
    static run(command, password) {
        var child = Spawn.new("sudo -S %(command)")
        child.timeout = 30

        var idx = child.expect(["[Pp]assword", Pexpect.EOF, Pexpect.TIMEOUT])

        if (idx == 0) {
            child.sendline(password)
            child.expect([Pexpect.EOF, Pexpect.TIMEOUT])
        }

        var output = child.before
        child.wait()
        var exitCode = child.exitstatus
        child.close()

        return {
            "output": output,
            "exitCode": exitCode,
            "success": exitCode == 0
        }
    }
}

System.print("Sudo automation example (conceptual)")

Example 5: Testing Interactive CLI Tools

import "pexpect" for Pexpect, Spawn

class CliTester {
    construct new(command) {
        _child = Spawn.new(command)
        _child.timeout = 10
        _failures = []
    }

    expectPrompt(prompt) {
        var idx = _child.expect([prompt, Pexpect.TIMEOUT])
        if (idx == Pexpect.TIMEOUT) {
            _failures.add("Expected prompt '%(prompt)' not found")
            return false
        }
        return true
    }

    sendInput(input) {
        _child.sendline(input)
    }

    expectOutput(pattern) {
        var idx = _child.expect([pattern, Pexpect.TIMEOUT])
        if (idx == Pexpect.TIMEOUT) {
            _failures.add("Expected output '%(pattern)' not found")
            return false
        }
        return true
    }

    assertContains(text) {
        if (!_child.before.contains(text)) {
            _failures.add("Output does not contain '%(text)'")
            return false
        }
        return true
    }

    close() {
        _child.sendeof()
        _child.close()
    }

    report() {
        if (_failures.count == 0) {
            System.print("All tests passed!")
            return true
        }

        System.print("%(_failures.count) test(s) failed:")
        for (failure in _failures) {
            System.print("  - %(failure)")
        }
        return false
    }
}

var tester = CliTester.new("python3")

tester.expectPrompt(">>>")
tester.sendInput("1 + 1")
tester.expectPrompt(">>>")
tester.assertContains("2")

tester.close()
tester.report()

Part 9: Error Handling

Step 26: Pattern Matching with Error Conditions

import "pexpect" for Pexpect, Spawn

var child = Spawn.new("bash -c 'echo start; sleep 1; echo end'")

var patterns = [
    "start",
    "error",
    Pexpect.EOF,
    Pexpect.TIMEOUT
]

var idx = child.expect(patterns)

if (idx == 0) {
    System.print("Process started successfully")
} else if (idx == 1) {
    System.print("Error detected: %(child.after)")
} else if (idx == Pexpect.EOF) {
    System.print("Process ended unexpectedly")
} else if (idx == Pexpect.TIMEOUT) {
    System.print("Operation timed out")
    child.terminate()
}

child.close()

Step 27: Cleanup on Errors

import "pexpect" for Pexpect, Spawn

class SafeSpawn {
    construct new(command) {
        _child = Spawn.new(command)
        _child.timeout = 30
    }

    run(fn) {
        var fiber = Fiber.new { fn.call(_child) }
        var result = fiber.try()

        if (fiber.error) {
            System.print("Error occurred: %(fiber.error)")
            cleanup()
            return {"success": false, "error": fiber.error}
        }

        return {"success": true, "result": result}
    }

    cleanup() {
        if (_child.isalive) {
            _child.terminate()
        }
        _child.close()
    }
}

var safe = SafeSpawn.new("python3")

var result = safe.run(Fn.new { |child|
    child.expect(">>>")
    child.sendline("print('Hello')")
    child.expect(">>>")
    child.sendeof()
    return child.before
})

System.print("Result: %(result)")

Part 10: Quick Command Execution

Step 28: Using Pexpect.run()

import "pexpect" for Pexpect

var output = Pexpect.run("ls -la")
System.print(output)

Step 29: Getting Exit Status

import "pexpect" for Pexpect

var result = Pexpect.run("grep pattern file.txt", 10, true)

var output = result[0]
var exitCode = result[1]

System.print("Output: %(output)")
System.print("Exit code: %(exitCode)")

Spawn Class Reference

Constructor

Input Methods

Expect Methods

Read Methods

Process Control

Properties

Next Steps

• Learn about subprocess for simpler command execution
• Explore regex for advanced pattern matching
• Build a complete CLI tool