Subprocess Command Execution

def run(command: str, **kwargs):
    """
    Run command in a subprocess and return a Result object.
    
    This is a convenience wrapper around Context().run().
    
    Parameters:
    - command (str): Shell command to execute
    - **kwargs: Additional execution options (see Context.run)
    
    Returns:
    Result: Command execution result
    """

def sudo(command: str, **kwargs):
    """
    Run command in a sudo subprocess and return a Result object.
    
    This is a convenience wrapper around Context().sudo().
    
    Parameters:
    - command (str): Shell command to execute with sudo
    - **kwargs: Additional execution options (see Context.sudo)
    
    Returns:
    Result: Command execution result
    """

from invoke import run, sudo

# Basic command execution
result = run("ls -la")
print(result.stdout)

# Sudo execution
result = sudo("systemctl restart nginx", pty=True)

class Context:
    """
    Execution context providing command running and configuration management.
    
    Attributes:
    - config (Config): Configuration object
    - cwd (str): Current working directory
    """
    
    def __init__(self, config=None):
        """
        Initialize Context.
        
        Parameters:
        - config (Config, optional): Configuration object
        """
    
    def run(self, command, **kwargs):
        """
        Execute a shell command via the configured runner.
        
        Parameters:
        - command (str): Shell command to execute
        - asynchronous (bool): Execute in background and return Promise
        - disown (bool): Fully detach subprocess (implies asynchronous=True)
        - echo (bool): Print command before execution  
        - encoding (str): Text encoding for I/O streams
        - env (dict): Environment variables
        - err_stream (file-like): Stream for stderr output
        - hide (bool/str): Hide stdout/stderr ('stdout', 'stderr', 'both', True)
        - in_stream (file-like): Stream for stdin input
        - out_stream (file-like): Stream for stdout output
        - pty (bool): Use a pseudo-terminal
        - replace_env (bool): Replace entire environment vs updating
        - shell (str): Shell to use for execution
        - timeout (float): Command timeout in seconds
        - warn (bool): Don't raise exceptions on non-zero exit
        - watchers (list): Stream watchers for interaction
        
        Returns:
        Result or Promise: Command execution result
        """
    
    def sudo(self, command, **kwargs):
        """
        Execute a shell command via sudo.
        
        Parameters:
        - command (str): Command to execute with sudo
        - password (str): Sudo password (will prompt if not provided)
        - user (str): User to sudo as (default: root)
        - **kwargs: Additional options (same as run())
        
        Returns:
        Result or Promise: Command execution result
        """
    
    def cd(self, path):
        """
        Context manager for temporarily changing working directory.
        
        Parameters:
        - path (str): Directory path to change to
        
        Returns:
        context manager: Directory change context
        """
    
    def prefix(self, command):
        """
        Context manager for prefixing commands.
        
        Parameters:
        - command (str): Command prefix to apply
        
        Returns:
        context manager: Command prefix context
        """

class MockContext(Context):
    """
    Mock context for testing, allowing predefined command results.
    
    Attributes:
    - config (Config): Configuration object
    """
    
    def set_result_for(self, command, result):
        """
        Set expected result for a command.
        
        Parameters:
        - command (str): Command string to mock
        - result (Result): Result to return
        """

class Runner:
    """
    Abstract base class for command runners.
    
    Attributes:
    - context (Context): Associated context
    - shell (str): Shell to use for execution
    - echo (bool): Whether to echo commands
    - encoding (str): Text encoding for streams
    """
    
    def run(self, command, **kwargs):
        """
        Execute command and return result.
        
        Parameters:
        - command (str): Command to execute
        - **kwargs: Execution options
        
        Returns:
        Result: Execution result
        """

class Local(Runner):
    """
    Local subprocess command runner.
    
    Executes commands on the local machine using subprocess.Popen.
    """
    
    def start(self, command, shell, env):
        """
        Start subprocess for command execution.
        
        Parameters:
        - command (str): Command to execute
        - shell (str): Shell to use
        - env (dict): Environment variables
        
        Returns:
        subprocess.Popen: Started subprocess
        """
    
    def wait(self):
        """
        Wait for subprocess completion.
        
        Returns:
        int: Process return code
        """

class Result:
    """
    Command execution result.
    
    Attributes:
    - return_code (int): Command exit code
    - stdout (str): Standard output
    - stderr (str): Standard error  
    - ok (bool): True if return_code == 0
    - failed (bool): True if return_code != 0
    - shell (str): Shell used for execution
    - command (str): Executed command
    - encoding (str): Text encoding used
    """
    
    def __init__(self, stdout="", stderr="", encoding=None, command="", shell="", return_code=0, pty=False):
        """
        Initialize Result.
        
        Parameters:
        - stdout (str): Standard output
        - stderr (str): Standard error
        - encoding (str): Text encoding
        - command (str): Executed command
        - shell (str): Shell used
        - return_code (int): Exit code
        - pty (bool): Whether PTY was used
        """
    
    def tail(self, stream, count=10):
        """
        Get last N lines from output stream.
        
        Parameters:
        - stream (str): Stream name ('stdout' or 'stderr')
        - count (int): Number of lines
        
        Returns:
        str: Last N lines
        """

class Promise(Result):
    """
    Asynchronous execution promise.
    
    Provides access to running subprocess while it executes.
    """
    
    def join(self, timeout=None):
        """
        Wait for subprocess completion.
        
        Parameters:
        - timeout (float, optional): Wait timeout
        
        Returns:
        Result: Final execution result
        """
    
    def __enter__(self):
        """Context manager entry."""
        return self
    
    def __exit__(self, *args):
        """Context manager exit."""
        self.join()

class Failure:
    """
    Command execution failure representation.
    
    Attributes:
    - result (Result): Associated result object
    - reason (str): Failure reason
    """

from invoke import Context, run

# Using convenience function
result = run("echo 'Hello World'")
print(f"Output: {result.stdout}")
print(f"Exit code: {result.return_code}")
print(f"Success: {result.ok}")

# Using context
ctx = Context()
result = ctx.run("ls -la")
if result.failed:
    print("Command failed!")

from invoke import Context

ctx = Context()

# Hide output
result = ctx.run("make build", hide=True)

# Use PTY for interactive commands  
result = ctx.run("ssh user@host", pty=True)

# Set timeout
try:
    result = ctx.run("long-running-command", timeout=30)
except CommandTimedOut:
    print("Command timed out!")

# Don't raise on failure
result = ctx.run("might-fail", warn=True)
if result.failed:
    print(f"Command failed with code {result.return_code}")

from invoke import Context

ctx = Context()

# Set environment variables
result = ctx.run("echo $MY_VAR", env={'MY_VAR': 'hello'})

# Change working directory
with ctx.cd('/tmp'):
    result = ctx.run("pwd")  # Shows /tmp

# Use command prefixes
with ctx.prefix('source venv/bin/activate'):
    result = ctx.run("python --version")

from invoke import Context

ctx = Context()

# Background execution
promise = ctx.run("long-task", asynchronous=True)
print("Task started, doing other work...")

# Check if complete
if promise.join(timeout=1):  # Wait 1 second
    print("Task completed:", promise.stdout)
else:
    print("Task still running...")

# Context manager usage
with ctx.run("background-task", asynchronous=True) as promise:
    # Do other work
    pass
# Automatically waits for completion on exit

from invoke import Context, UnexpectedExit

ctx = Context()

try:
    result = ctx.run("false")  # Command that returns 1
except UnexpectedExit as e:
    print(f"Command failed: {e.result.return_code}")
    print(f"Stderr: {e.result.stderr}")

# Or use warn=True to avoid exceptions
result = ctx.run("false", warn=True)
if result.failed:
    print("Command failed but didn't raise exception")

from invoke import MockContext, Result

# Create mock context
ctx = MockContext()

# Set up expected results
ctx.set_result_for("echo hello", Result(stdout="hello\n"))

# Use in tests
result = ctx.run("echo hello")
assert result.stdout == "hello\n"