grpc_running_server_data

drunc.grpc_testing_tools.grpc_running_server_data

Classes

RunningGrpcServer(process_id, target_func, args, kwargs)

Abstract representation of a Running gRPC Server Process The server could have been started via any supported method (multiprocessing, SSH, etc.)

Initialise process handle with execution parameters.

Parameters:

Name	Type	Description	Default
process_id	str	Unique identifier for this process	required
target_func	Any	Function to execute as the process	required
args	tuple	Positional arguments for the target function	required
kwargs	dict	Keyword arguments for the target function	required

Source code in drunc/grpc_testing_tools/grpc_running_server_data.py

def __init__(self, process_id: str, target_func: Any, args: tuple, kwargs: dict):
    """
    Initialise process handle with execution parameters.

    Args:
        process_id: Unique identifier for this process
        target_func: Function to execute as the process
        args: Positional arguments for the target function
        kwargs: Keyword arguments for the target function
    """
    self.process_id = process_id
    self.target_func = target_func
    self.args = args
    self.kwargs = kwargs
    self._process = None
    self._started = False
    self.startup_error = None
    self.host = None
    self.server_id = None
    self.port = None
    self.server_type = None
    self.ready_event = None
    self.stop_event = None

Attributes

process property

Get the underlying process object (implementation-specific).

started property

Check if process has been started.

Functions

is_valid()

Check if the process handle is valid

Source code in drunc/grpc_testing_tools/grpc_running_server_data.py

def is_valid(self) -> bool:
    """Check if the process handle is valid"""
    required_not_none = [
        self.process_id,
        self.host,
        self.server_id,
        self.port,
        self.server_type,
    ]
    return all(param is not None for param in required_not_none)

mark_started()

Mark this process as started.

Source code in drunc/grpc_testing_tools/grpc_running_server_data.py

def mark_started(self) -> None:
    """Mark this process as started."""
    self._started = True

set_process(process)

Set the underlying process object.

Source code in drunc/grpc_testing_tools/grpc_running_server_data.py

def set_process(self, process: Any) -> None:
    """Set the underlying process object."""
    self._process = process

set_server_info(server_id, host, port, server_type)

Set the server information for this process.

Source code in drunc/grpc_testing_tools/grpc_running_server_data.py

def set_server_info(
    self, server_id: str, host: str, port: int, server_type: str
) -> None:
    """Set the server information for this process."""
    self.server_id = server_id
    self.host = host
    self.port = port
    self.server_type = server_type