plesty.lib.test.resource_allocation
===================================

.. py:module:: plesty.lib.test.resource_allocation

.. autoapi-nested-parse::

   Multiprocessing helpers for testing server/client resource allocation.

   Failures raise :class:`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
-------

.. autoapisummary::

   plesty.lib.test.resource_allocation.ClientProcessConfig


Functions
---------

.. autoapisummary::

   plesty.lib.test.resource_allocation._test_multiclient_resource_allocation
   plesty.lib.test.resource_allocation._run_server_process
   plesty.lib.test.resource_allocation._run_client_process
   plesty.lib.test.resource_allocation._is_alive
   plesty.lib.test.resource_allocation._stop_process
   plesty.lib.test.resource_allocation._stop_all_processes
   plesty.lib.test.resource_allocation._drain_server_error
   plesty.lib.test.resource_allocation._wait_for_client_results
   plesty.lib.test.resource_allocation._assert_connection_info_matches
   plesty.lib.test.resource_allocation.assert_resource_manager_client_allocation


Module Contents
---------------

.. py:function:: _test_multiclient_resource_allocation(device_cls, device_args: dict[str, Any], client_resources: list[list[str]])

.. py:class:: ClientProcessConfig

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


   .. py:attribute:: client_id
      :type:  str


   .. py:attribute:: resources
      :type:  list[str]


   .. py:attribute:: phase
      :type:  int
      :value: 0



   .. py:attribute:: hold_seconds
      :type:  float
      :value: 0



   .. py:attribute:: expected_connection_info
      :type:  dict[str, Any]


.. py:function:: _run_server_process(device_cls, device_args: dict[str, Any], server_address: str, result_queue, build_server_func: Callable = build_server)

.. py:function:: _run_client_process(client_id: str, resources: list[str], client_address: str | None, hold_seconds: float, result_queue, build_client_func: Callable = build_client)

.. py:function:: _is_alive(proc) -> bool

.. py:function:: _stop_process(proc, name: str)

.. py:function:: _stop_all_processes(processes: list[tuple[str, multiprocessing.Process]])

.. py:function:: _drain_server_error(result_queue)

.. py:function:: _wait_for_client_results(result_queue, expected_client_ids: list[str], timeout: float)

.. py:function:: _assert_connection_info_matches(client_id: str, actual: dict[str, Any], expected: dict[str, Any])

.. py:function:: 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.

   .. rubric:: 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