plesty.lib.test.resource_allocation

Multiprocessing helpers for testing server/client resource allocation.

Failures raise AssertionError rather than calling pytest.fail: the module ships in plesty-lib, where pytest is a dev-group dependency, and the helper is callable from a plain script (#19).

Classes

ClientProcessConfig

Configuration for a single client process in a resource allocation test.

Functions

_test_multiclient_resource_allocation(device_cls, ...)

_run_server_process(device_cls, device_args, ...[, ...])

_run_client_process(client_id, resources, ...[, ...])

_is_alive(→ bool)

_stop_process(proc, name)

_stop_all_processes(processes)

_drain_server_error(result_queue)

_wait_for_client_results(result_queue, ...)

_assert_connection_info_matches(client_id, actual, ...)

assert_resource_manager_client_allocation(→ Any)

Generic multiprocessing helper for testing resource allocation.

Module Contents

plesty.lib.test.resource_allocation._test_multiclient_resource_allocation(device_cls, device_args: dict[str, Any], client_resources: list[list[str]])
Parameters:

  • device_args (dict[str, Any])

  • client_resources (list[list[str]])

class plesty.lib.test.resource_allocation.ClientProcessConfig

Configuration for a single client process in a resource allocation test.

client_id: str
resources: list[str]
phase: int = 0
hold_seconds: float = 0
expected_connection_info: dict[str, Any]
plesty.lib.test.resource_allocation._run_server_process(device_cls, device_args: dict[str, Any], server_address: str, result_queue, build_server_func: Callable = build_server)
Parameters:

  • device_args (dict[str, Any])

  • server_address (str)

  • build_server_func (Callable)

plesty.lib.test.resource_allocation._run_client_process(client_id: str, resources: list[str], client_address: str | None, hold_seconds: float, result_queue, build_client_func: Callable = build_client)
Parameters:

  • client_id (str)

  • resources (list[str])

  • client_address (str | None)

  • hold_seconds (float)

  • build_client_func (Callable)

plesty.lib.test.resource_allocation._is_alive(proc) → bool
Return type:

bool

plesty.lib.test.resource_allocation._stop_process(proc, name: str)
Parameters:

name (str)

plesty.lib.test.resource_allocation._stop_all_processes(processes: list[tuple[str, multiprocessing.Process]])
Parameters:

processes (list[tuple[str, multiprocessing.Process]])

plesty.lib.test.resource_allocation._drain_server_error(result_queue)
plesty.lib.test.resource_allocation._wait_for_client_results(result_queue, expected_client_ids: list[str], timeout: float)
Parameters:

  • expected_client_ids (list[str])

  • timeout (float)

plesty.lib.test.resource_allocation._assert_connection_info_matches(client_id: str, actual: dict[str, Any], expected: dict[str, Any])
Parameters:

  • client_id (str)

  • actual (dict[str, Any])

  • expected (dict[str, Any])

plesty.lib.test.resource_allocation.assert_resource_manager_client_allocation(*, device_cls: Any, device_args: dict[str, Any], client_configs: list[ClientProcessConfig], server_address: str = 'tcp://*:5555', client_address: str | None = 'tcp://localhost:5555', server_startup_seconds: float = 2, phase_timeout: float = 10, build_server_func: Callable = build_server, build_client_func: Callable = build_client) → Any

Generic multiprocessing helper for testing resource allocation.

It starts:

  • one server process

  • one process per client

It then:

  • starts clients phase by phase

  • collects each client’s connection_info

  • checks expected connection_info values

  • kills all clients

  • kills the server

Use phases when ordering matters.

Example

phase 0: client1 reserves Dev1/port0/line0 phase 1: client2 tries Dev1/port0/line0 and should fail phase 1: client3 tries Dev1/port0/line1 and should succeed

Parameters:

  • device_cls (Any)

  • device_args (dict[str, Any])

  • client_configs (list[ClientProcessConfig])

  • server_address (str)

  • client_address (str | None)

  • server_startup_seconds (float)

  • phase_timeout (float)

  • build_server_func (Callable)

  • build_client_func (Callable)

Return type:

Any