# Test Helpers

`plesty.lib.test` provides lightweight helpers for development-time smoke testing of device APIs.
None of these replace full integration or regression suites — they validate interface contracts
and surface obvious breakage quickly.

## Modules

| Module | What it does |
|---|---|
| `device_param_system` | Auto-test read/write paths for all registered config parameters |
| `device_func_system` | Auto-test all registered operations in the function system |
| `grouped_param_keys` | pytest fixtures that verify grouped parameter key resolution |
| `resource_allocation` | Multiprocessing helpers for testing multi-client resource contention |
| `schema_params` | Helpers for refreshing a JSON parameter schema from a real device |

---

## 1. `device_param_system` — Config Parameter Auto Test

**Location:** `plesty.lib.test.device_param_system`

Iterates every registered config key and exercises read-only, write-only, and read-write paths.

**What it does:**

1. Opens the device with a context manager.
2. Prints the identity string (failure is non-fatal).
3. Optionally binds a mock `_write_` / `_query_` pair that stores writes in memory and
   generates schema-compatible responses with `DataGenerator`.
4. For each key:
   - `read_only` → calls `query`.
   - `write_only` → skipped (logged as ignored).
   - read-write → calls `query` to get the current value, then writes it back.
5. Prints a `Passed / Failed` summary.

**Key parameters:**

| Parameter | Default | Description |
|---|---|---|
| `use_mock_solver` | `True` | Bind in-memory mock handlers instead of real hardware I/O |
| `ignore_keys` | `None` | List of keys to skip |
| `seed` | `123` | RNG seed for `DataGenerator` |
| `sleep_time` | `0.1` | Seconds between parameter tests |

**Example (mock mode):**

```python
from plesty.lib.test.device_param_system import auto_test
from my_device import MyDevice

auto_test(MyDevice, "device-id", use_mock_solver=True, seed=42)
```

**Example (real hardware):**

```python
from plesty.lib.test.device_param_system import auto_test
from my_device import MyDevice

auto_test(
    MyDevice,
    "USB0::0x1313::0x8078::P0000001::INSTR",
    use_mock_solver=False,
    ignore_keys=["DANGEROUS_PARAM"],
)
```

---

## 2. `device_func_system` — Function System Auto Test

**Location:** `plesty.lib.test.device_func_system`

Iterates every registered operation and validates that it accepts generated payloads and returns
a `dict`.

**What it does:**

1. Opens the device with a context manager.
2. Prints the identity string (failure is non-fatal).
3. Optionally binds a mock op solver that generates `FuncOutput`-shaped response dicts from
   `FuncParam` metadata.
4. For each operation:
   - Builds a payload from `FuncParam` constraints using `DataGenerator`.
   - Randomly omits optional fields (30 % chance) to exercise default handling.
   - Calls the operation.
   - Asserts the response is a `dict`.
5. Prints a `Passed / Failed` summary.

**Key parameters:**

| Parameter | Default | Description |
|---|---|---|
| `use_mock_solver` | `True` | Bind a mock op solver instead of the real backend |
| `ignore_ops` | `None` | List of operation names to skip |
| `seed` | `123` | RNG seed for `DataGenerator` |
| `sleep_time` | `0.1` | Seconds between operation calls |

**Example (mock solver):**

```python
from plesty.lib.test.device_func_system import auto_test
from my_device import MyDevice

auto_test(MyDevice, "device-id", use_mock_solver=True, seed=123)
```

**Example (real solver, skip dangerous ops):**

```python
from plesty.lib.test.device_func_system import auto_test
from my_device import MyDevice

auto_test(MyDevice, "device-id", use_mock_solver=False, ignore_ops=["reset", "calibrate"])
```

---

## 3. `grouped_param_keys` — Grouped Parameter Key pytest Tests

**Location:** `plesty.lib.test.grouped_param_keys`

Three `pytest` test functions that verify `ConfigSystem` key resolution rules for grouped
parameters. Import and call them from your own test suite.

**Tests provided:**

| Function | What it asserts |
|---|---|
| `test_default_param_uses_bare_key_only` | Default-group params are accessible by bare key only; `default.key` raises `KeyError` |
| `test_grouped_param_requires_full_key` | Grouped params require the `PREFIX.key` form; bare key raises `KeyError` |
| `test_get_config_list_returns_full_keys_for_grouped_params` | `get_config_list()` returns `PREFIX.key` for grouped params and bare keys for defaults |

**Usage in your test suite:**

```python
from plesty.lib.test.grouped_param_keys import (
    test_default_param_uses_bare_key_only,
    test_grouped_param_requires_full_key,
    test_get_config_list_returns_full_keys_for_grouped_params,
)
```

pytest discovers and runs them automatically when imported into a test file.

---

## 4. `resource_allocation` — Multi-Client Resource Contention Helpers

**Location:** `plesty.lib.test.resource_allocation`

Multiprocessing helpers for testing that the server-side `ResourceManager` correctly grants and
denies resource allocations across concurrent clients.

**Core function:**

```python
assert_resource_manager_client_allocation(
    *,
    device_cls,
    device_args: dict,
    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 = build_server,
    build_client_func = build_client,
)
```

It spins up one server process and one process per client, runs them phase by phase, and
asserts `connection_info` matches expectations.

**`ClientProcessConfig` fields:**

| Field | Type | Description |
|---|---|---|
| `client_id` | `str` | Unique identifier for this client |
| `resources` | `list[str]` | Resources to request on connection |
| `phase` | `int` | Execution phase; lower phases start first |
| `hold_seconds` | `float` | How long to keep the process alive after reporting (keeps resources reserved) |
| `expected_connection_info` | `dict` | Key/value pairs that must appear in `client.connection_info` |

**Phases** control ordering. A client in phase 0 connects and optionally holds resources while
phase 1 clients connect. This makes it possible to assert that a conflicting request fails.

**Example:**

```python
from plesty.lib.test.resource_allocation import (
    assert_resource_manager_client_allocation,
    ClientProcessConfig,
)
from my_device import MyDevice


def test_resource_contention():
    assert_resource_manager_client_allocation(
        device_cls=MyDevice,
        device_args={"port": 5555},
        client_configs=[
            ClientProcessConfig(
                client_id="client1",
                resources=["Dev1/port0/line0"],
                phase=0,
                hold_seconds=10,
                expected_connection_info={"client_id": "client1", "resources": ["Dev1/port0/line0"]},
            ),
            ClientProcessConfig(
                client_id="client2",
                resources=["Dev1/port0/line0"],   # same resource — should fail
                phase=1,
                expected_connection_info={},
            ),
            ClientProcessConfig(
                client_id="client3",
                resources=["Dev1/port0/line1"],   # different resource — should succeed
                phase=1,
                expected_connection_info={"client_id": "client3", "resources": ["Dev1/port0/line1"]},
            ),
        ],
    )
```

A client that fails allocation raises `RuntimeError("ResourceAllocationError: Requested resources
are not available")`. The helper asserts this automatically for clients whose `expected_connection_info`
is empty.

---

## 5. `schema_params` — Schema Refresh Helper

**Location:** `plesty.lib.test.schema_params`

Connects to a real device, queries every registered parameter, and writes an updated JSON schema
file with inferred types and current defaults.

**Function:**

```python
test_real_device_refreshes_schema_defaults_and_types(
    device_cls,
    param_schema=None,   # path to the existing schema JSON file
    *args,
    **kwargs,
)
```

**What it does:**

1. Connects to the device (`device.connect()`).
2. Asserts `device.is_operatable`.
3. For each key in `get_config_list()`:
   - Queries the current value from hardware.
   - Infers the Python type (`bool`, `int`, `float`, or `str`).
   - Updates `type` and `default` in the schema entry.
4. Writes the refreshed schema to `<original_name>_refreshed.json`.
5. Prints a warning for any keys that failed to query.

**Supported schema shapes:**

- Flat: `{"PARAM": {...}, ...}`
- Grouped: `{"groups": {"camera": {"command_prefix": "CS", "parameters": {...}}}}`

Dotted runtime keys (`CS.gain`) are resolved to the correct group entry automatically.

**Example:**

```python
from plesty.lib.test.schema_params import test_real_device_refreshes_schema_defaults_and_types
from my_device import MyDevice


def test_refresh_schema():
    test_real_device_refreshes_schema_defaults_and_types(
        MyDevice,
        param_schema="config/my_device_params.json",
        address="USB0::0x1313::0x8078::P0000001::INSTR",
    )
```

The function is marked `__test__ = False` so pytest does not auto-collect it.

---

## Recommended Workflow

1. Start with `device_param_system.auto_test(use_mock_solver=True)` after registering parameters —
   validates schema and parsing logic without hardware.
2. Switch to `use_mock_solver=False` to run against a real device.
3. Add `device_func_system.auto_test(use_mock_solver=True)` after registering operations —
   validates operation schemas and response shapes.
4. Use `schema_params.test_real_device_refreshes_schema_defaults_and_types` to refresh JSON
   schemas from a real device after hardware changes.
5. Use `resource_allocation.assert_resource_manager_client_allocation` for server-mode devices
   that share resources across clients.

## Using with pytest

```python
from plesty.lib.test.device_param_system import auto_test as auto_test_params
from plesty.lib.test.device_func_system import auto_test as auto_test_funcs
from my_device import MyDevice


def test_params_mock():
    auto_test_params(MyDevice, "device-id", use_mock_solver=True, seed=0)


def test_funcs_mock():
    auto_test_funcs(MyDevice, "device-id", use_mock_solver=True, seed=0)


def test_no_failed_operations(capsys):
    auto_test_funcs(MyDevice, "device-id", use_mock_solver=True, seed=0)
    captured = capsys.readouterr()
    assert "Failed: 0" in captured.out, captured.out
```