process_connection_manager

drunc.grpc_testing_tools.process_connection_manager

Abstract Process Connection Manager

This module provides abstraction over different process execution methods (multiprocessing, SSH, etc.).

Classes

ProcessConnectionManager(env_vars=None)

Bases: ABC

Abstract base class for managing process connections.

Defines the interface that all process connection implementations must follow, whether using local multiprocessing or remote SSH execution.

Initialise process connection manager.

Parameters:

Name	Type	Description	Default
env_vars	Dict[str, str]	Environment variables to set for all processes	None

Source code in drunc/grpc_testing_tools/process_connection_manager.py

def __init__(self, env_vars: Dict[str, str] = None):
    """
    Initialise process connection manager.

    Args:
        env_vars: Environment variables to set for all processes
    """
    self.env_vars = env_vars or {}
    self.process_handles: Dict[str, RunningGrpcServer] = {}

Functions

cleanup() abstractmethod

Clean up all resources and stop any remaining processes.

Source code in drunc/grpc_testing_tools/process_connection_manager.py

@abstractmethod
def cleanup(self) -> None:
    """Clean up all resources and stop any remaining processes."""
    pass

create_process(process_id, target_func, *args, **kwargs) abstractmethod

Create a new process handle for execution.

Parameters:

Name	Type	Description	Default
process_id	str	Unique identifier for the process	required
target_func	Any	Function to execute as the process	required
*args		Positional arguments for target function	()
**kwargs		Keyword arguments for target function	{}

Returns:

Name	Type	Description
RunningGrpcServer	RunningGrpcServer	Handle for managing the created process

Source code in drunc/grpc_testing_tools/process_connection_manager.py

@abstractmethod
def create_process(
    self, process_id: str, target_func: Any, *args, **kwargs
) -> RunningGrpcServer:
    """
    Create a new process handle for execution.

    Args:
        process_id: Unique identifier for the process
        target_func: Function to execute as the process
        *args: Positional arguments for target function
        **kwargs: Keyword arguments for target function

    Returns:
        RunningGrpcServer: Handle for managing the created process
    """
    pass

get_process_handle(process_id)

Retrieve a process handle by ID.

Parameters:

Name	Type	Description	Default
process_id	str	ID of the process to retrieve	required

Returns:

Type	Description
Optional[RunningGrpcServer]	RunningGrpcServer if found, None otherwise

Source code in drunc/grpc_testing_tools/process_connection_manager.py

def get_process_handle(self, process_id: str) -> Optional[RunningGrpcServer]:
    """
    Retrieve a process handle by ID.

    Args:
        process_id: ID of the process to retrieve

    Returns:
        RunningGrpcServer if found, None otherwise
    """
    return self.process_handles.get(process_id)

is_process_alive(handle) abstractmethod

Check if a process is currently running.

Parameters:

Name	Type	Description	Default
handle	RunningGrpcServer	RunningGrpcServer to check	required

Returns:

Type	Description
bool	True if process is alive, False otherwise

Source code in drunc/grpc_testing_tools/process_connection_manager.py

@abstractmethod
def is_process_alive(self, handle: RunningGrpcServer) -> bool:
    """
    Check if a process is currently running.

    Args:
        handle: RunningGrpcServer to check

    Returns:
        True if process is alive, False otherwise
    """
    pass

list_process_ids()

Get list of all managed process IDs.

Returns:

Type	Description
List[str]	List of process IDs

Source code in drunc/grpc_testing_tools/process_connection_manager.py

def list_process_ids(self) -> List[str]:
    """
    Get list of all managed process IDs.

    Returns:
        List of process IDs
    """
    return list(self.process_handles.keys())

start_process(handle) abstractmethod

Start execution of a process.

Parameters:

Name	Type	Description	Default
handle	RunningGrpcServer	RunningGrpcServer for the process to start	required

Raises:

Type	Description
RuntimeError	If process cannot be started

Source code in drunc/grpc_testing_tools/process_connection_manager.py

@abstractmethod
def start_process(self, handle: RunningGrpcServer) -> None:
    """
    Start execution of a process.

    Args:
        handle: RunningGrpcServer for the process to start

    Raises:
        RuntimeError: If process cannot be started
    """
    pass

stop_process(handle, timeout=10.0) abstractmethod

Stop a running process gracefully.

Parameters:

Name	Type	Description	Default
handle	RunningGrpcServer	RunningGrpcServer for the process to stop	required
timeout	float	Maximum time to wait for graceful shutdown	10.0

Raises:

Type	Description
RuntimeError	If process cannot be stopped

Source code in drunc/grpc_testing_tools/process_connection_manager.py

@abstractmethod
def stop_process(self, handle: RunningGrpcServer, timeout: float = 10.0) -> None:
    """
    Stop a running process gracefully.

    Args:
        handle: RunningGrpcServer for the process to stop
        timeout: Maximum time to wait for graceful shutdown

    Raises:
        RuntimeError: If process cannot be stopped
    """
    pass

wait_for_termination(handle, timeout=None) abstractmethod

Wait for a process to terminate.

Parameters:

Name	Type	Description	Default
handle	RunningGrpcServer	RunningGrpcServer to wait for	required
timeout	Optional[float]	Maximum time to wait (None for indefinite)	None

Source code in drunc/grpc_testing_tools/process_connection_manager.py

@abstractmethod
def wait_for_termination(
    self, handle: RunningGrpcServer, timeout: Optional[float] = None
) -> None:
    """
    Wait for a process to terminate.

    Args:
        handle: RunningGrpcServer to wait for
        timeout: Maximum time to wait (None for indefinite)
    """
    pass