tessl/pypi-tox

tox is a generic virtualenv management and test command line tool

Overview

Eval results

Files

Execution System

Name: tessl/pypi-tox
Author: tessl

Command execution infrastructure for running commands within tox environments. The execution system supports subprocess execution, output capture, execution status tracking, and provides abstractions for different execution backends.

Capabilities

Execution Request

Specification for command execution including command, environment, and execution options.

class ExecuteRequest:
    """Specification for command execution."""
    
    def __init__(
        self, 
        cmd: list[str], 
        cwd: Path | None = None,
        env: dict[str, str] | None = None,
        stdin: str | None = None,
        run_id: str | None = None
    ) -> None:
        """
        Initialize execution request.
        
        Args:
            cmd: Command and arguments to execute
            cwd: Working directory for execution
            env: Environment variables
            stdin: Standard input data
            run_id: Unique identifier for this execution
        """
    
    @property
    def cmd(self) -> list[str]:
        """Command and arguments."""
        
    @property
    def cwd(self) -> Path | None:
        """Working directory."""
        
    @property
    def env(self) -> dict[str, str] | None:
        """Environment variables."""
        
    @property
    def stdin(self) -> str | None:
        """Standard input data."""
        
    @property
    def run_id(self) -> str | None:
        """Execution run identifier."""

Usage example:

from tox.execute.request import ExecuteRequest

# Simple command execution
request = ExecuteRequest(['pytest', '--verbose'])

# Command with working directory
request = ExecuteRequest(
    cmd=['python', 'setup.py', 'test'],
    cwd=Path('/project/root')
)

# Command with environment variables
request = ExecuteRequest(
    cmd=['pytest'],
    env={'PYTHONPATH': '/src', 'DEBUG': '1'}
)

Execute Interface

Abstract interface for command execution backends.

class Execute:
    """Command execution abstraction."""
    
    def __call__(self, request: ExecuteRequest) -> ExecuteStatus:
        """
        Execute command request.
        
        Args:
            request: Execution specification
            
        Returns:
            ExecuteStatus: Execution result
        """
    
    def __enter__(self):
        """Context manager entry."""
        
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit."""

Execution Status

Result of command execution including exit code, output, and metadata.

class ExecuteStatus:
    """Result of command execution."""
    
    @property
    def exit_code(self) -> int:
        """Process exit code."""
        
    @property
    def out(self) -> str:
        """Standard output."""
        
    @property
    def err(self) -> str:
        """Standard error output."""
        
    @property
    def duration(self) -> float:
        """Execution duration in seconds."""
        
    @property
    def success(self) -> bool:
        """Whether execution was successful (exit code 0)."""

class Outcome:
    """Execution outcome with additional metadata."""
    
    @property
    def status(self) -> ExecuteStatus:
        """Execution status."""
        
    @property
    def exception(self) -> Exception | None:
        """Exception if execution failed."""

Usage example:

# Execute command and check results
status = executor(ExecuteRequest(['pytest']))

print(f"Exit code: {status.exit_code}")
print(f"Success: {status.success}")
print(f"Duration: {status.duration:.2f}s")

if not status.success:
    print(f"Error output: {status.err}")

Local Subprocess Execution

Concrete implementation for local subprocess execution.

class LocalSubProcessExecute(Execute):
    """Local subprocess executor."""
    
    def __init__(self, colored: bool = True) -> None:
        """
        Initialize subprocess executor.
        
        Args:
            colored: Whether to preserve ANSI color codes
        """
    
    def __call__(self, request: ExecuteRequest) -> ExecuteStatus:
        """Execute command as local subprocess."""
        
    def interrupt(self) -> None:
        """Interrupt running subprocess."""
        
    def terminate(self) -> None:
        """Terminate running subprocess."""

Command Execution Patterns

Basic Execution

Simple command execution within an environment:

from tox.execute.request import ExecuteRequest
from tox.execute.local_sub_process import LocalSubProcessExecute

# Create executor
executor = LocalSubProcessExecute()

# Execute command
request = ExecuteRequest(['python', '--version'])
status = executor(request)

print(f"Python version output: {status.out}")

Environment Integration

Execution integrated with tox environments:

# Within a ToxEnv implementation
def execute(self, request: ExecuteRequest) -> ExecuteStatus:
    """Execute command in this environment."""
    
    # Prepare environment-specific execution context
    env_vars = dict(os.environ)
    env_vars.update(self.conf.get('setenv', {}))
    
    # Create request with environment context
    env_request = ExecuteRequest(
        cmd=request.cmd,
        cwd=self.conf.get('changedir', self.work_dir),
        env=env_vars,
        run_id=request.run_id
    )
    
    # Execute with environment's executor
    return self._executor(env_request)

Output Streaming

Handle real-time output during execution:

class StreamingExecutor(Execute):
    """Executor with streaming output."""
    
    def __call__(self, request: ExecuteRequest) -> ExecuteStatus:
        """Execute with streaming output."""
        process = subprocess.Popen(
            request.cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            cwd=request.cwd,
            env=request.env,
            text=True,
            bufsize=1,
            universal_newlines=True
        )
        
        # Stream output in real-time
        for line in process.stdout:
            print(line.rstrip())
            
        exit_code = process.wait()
        return ExecuteStatus(exit_code, "", "")

Process Management

Process Lifecycle

The execution system manages the complete process lifecycle:

Preparation: Setup working directory, environment variables
Execution: Start subprocess with specified parameters
Monitoring: Track process status and capture output
Completion: Collect exit code and output
Cleanup: Release resources and handle interrupts

Signal Handling

Handle interrupts and termination gracefully:

class ManagedExecutor(Execute):
    """Executor with signal handling."""
    
    def __init__(self):
        self._current_process = None
        
    def __call__(self, request):
        """Execute with signal handling."""
        try:
            self._current_process = subprocess.Popen(...)
            return self._wait_for_completion()
        except KeyboardInterrupt:
            self.interrupt()
            raise
            
    def interrupt(self):
        """Handle interrupt signal."""
        if self._current_process:
            self._current_process.terminate()
            # Wait for graceful termination
            try:
                self._current_process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self._current_process.kill()

Resource Management

Proper resource management for subprocess execution:

def execute_with_resources(self, request: ExecuteRequest) -> ExecuteStatus:
    """Execute with proper resource management."""
    
    with tempfile.TemporaryDirectory() as temp_dir:
        # Setup temporary resources
        log_file = Path(temp_dir) / "execution.log"
        
        try:
            # Execute command
            process = subprocess.Popen(...)
            
            # Monitor and log
            with open(log_file, 'w') as log:
                # Capture output
                pass
                
        finally:
            # Cleanup resources
            if process and process.poll() is None:
                process.terminate()

Execution Context

Environment Variables

Control execution environment through variables:

def create_execution_env(self, tox_env: ToxEnv) -> dict[str, str]:
    """Create execution environment."""
    
    # Start with system environment
    env = dict(os.environ)
    
    # Add tox-specific variables
    env.update({
        'TOX_ENV_NAME': tox_env.name,
        'TOX_ENV_DIR': str(tox_env.env_dir),
        'TOX_WORK_DIR': str(tox_env.work_dir),
    })
    
    # Add environment-specific variables
    env.update(tox_env.conf.get('setenv', {}))
    
    # Filter through passenv
    passenv = tox_env.conf.get('passenv', [])
    if passenv:
        filtered_env = {}
        for key in passenv:
            if key in env:
                filtered_env[key] = env[key]
        env = filtered_env
        
    return env

Working Directory

Manage working directory for command execution:

def determine_working_dir(self, tox_env: ToxEnv, request: ExecuteRequest) -> Path:
    """Determine working directory for execution."""
    
    if request.cwd:
        return request.cwd
        
    # Check environment configuration
    changedir = tox_env.conf.get('changedir')
    if changedir:
        return Path(changedir)
        
    # Default to environment directory
    return tox_env.env_dir

Error Handling

Execution Errors

Handle various execution error conditions:

class ExecuteError(Exception):
    """Base execution error."""

class CommandNotFoundError(ExecuteError):
    """Command not found in PATH."""
    
class TimeoutError(ExecuteError):
    """Command execution timeout."""
    
class InterruptError(ExecuteError):
    """Command execution interrupted."""

Error Recovery

Implement error recovery strategies:

def execute_with_retry(self, request: ExecuteRequest, max_retries: int = 3) -> ExecuteStatus:
    """Execute command with retry logic."""
    
    for attempt in range(max_retries):
        try:
            status = self._executor(request)
            
            # Return on success
            if status.success:
                return status
                
            # Check if retry is appropriate
            if self._should_retry(status):
                print(f"Retrying command (attempt {attempt + 1}/{max_retries})")
                continue
            else:
                return status
                
        except ExecuteError as e:
            if attempt == max_retries - 1:
                raise
            print(f"Execution failed, retrying: {e}")
            
    return status

Integration with Tox Environments

The execution system integrates closely with tox environments:

Command Preparation: Environments prepare commands with substitutions
Environment Setup: Execution context includes environment variables and paths
Output Handling: Environments control output formatting and logging
Error Propagation: Execution errors are handled by environment lifecycle
Resource Cleanup: Environments ensure proper cleanup after execution

This tight integration ensures that commands execute correctly within the isolated environment context while maintaining proper resource management and error handling.

Install with Tessl CLI