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	TargetFunc[P, object]	Function to execute as the process	required
args	tuple[object, ...]	Positional arguments for the target function	required
kwargs	dict[str, object]	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: TargetFunc[P, object],
    args: tuple[object, ...],
    kwargs: dict[str, object],
) -> None:
    """
    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: multiprocessing.Process | None = None
    self._started = False
    self.startup_error: Exception | None = None
    self.host: str | None = None
    self.server_id: str | None = None
    self.port: int | None = None
    self.server_type: str | ServerType | None = None
    self.ready_event: multiprocessing.synchronize.Event | None = None
    self.stop_event: multiprocessing.synchronize.Event | None = None

Attributes

process property

Get the underlying process object (implementation-specific).

started property

Check if process has been started.

Methods:

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: multiprocessing.Process) -> 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 | ServerType
) -> None:
    """Set the server information for this process."""
    self.server_id = server_id
    self.host = host
    self.port = port
    self.server_type = server_type