ria-toolkit-oss/tests/agent/test_hardware.py

from ria_toolkit_oss.agent import hardware
from ria_toolkit_oss.sdr import detect_available


def test_detect_available_includes_mock():
    drivers = detect_available()
    assert "mock" in drivers
    from ria_toolkit_oss.sdr.mock import MockSDR

    assert drivers["mock"] is MockSDR


def test_available_devices_sorted_list():
    devices = hardware.available_devices()
    assert isinstance(devices, list)
    assert devices == sorted(devices)
    assert "mock" in devices


def _device_names(hardware_list):
    return {e["device"] for e in hardware_list}


def test_heartbeat_payload_shape():
    p = hardware.heartbeat_payload()
    assert p["type"] == "heartbeat"
    assert p["status"] == "idle"
    # hardware is now a list of rich dict entries.
    assert "mock" in _device_names(p["hardware"])
    assert "app_id" not in p
    # New fields, default shape
    assert p["capabilities"] == ["rx"]
    assert p["tx_enabled"] is False

    p2 = hardware.heartbeat_payload(status="streaming", app_id="abc")
    assert p2["status"] == "streaming"
    assert p2["app_id"] == "abc"


def test_detect_devices_entry_shape():
    devices = hardware.detect_devices(use_cache=False)
    assert isinstance(devices, list)
    for entry in devices:
        assert set(entry) >= {"device", "identifier", "label", "connected"}
        assert isinstance(entry["device"], str)
        # identifier round-trips through parse_ident: None or a string.
        assert entry["identifier"] is None or isinstance(entry["identifier"], str)
    mock = next(e for e in devices if e["device"] == "mock")
    assert mock["label"]  # has a human label


def test_detect_devices_cache():
    a = hardware.detect_devices(use_cache=False)
    b = hardware.detect_devices(use_cache=True)
    assert _device_names(a) == _device_names(b)


def test_heartbeat_payload_tx_capability_from_cfg():
    from ria_toolkit_oss.agent.config import AgentConfig

    p = hardware.heartbeat_payload(cfg=AgentConfig(tx_enabled=True))
    assert p["capabilities"] == ["rx", "tx"]
    assert p["tx_enabled"] is True


def test_heartbeat_payload_sessions_field():
    sessions = {"rx": {"app_id": "a", "state": "streaming"}}
    p = hardware.heartbeat_payload(status="streaming", app_id="a", sessions=sessions)
    assert p["sessions"] == sessions


def test_heartbeat_payload_surfaces_tx_caps_when_enabled():
    from ria_toolkit_oss.agent.config import AgentConfig

    cfg = AgentConfig(
        tx_enabled=True,
        tx_max_gain_db=-10.0,
        tx_max_duration_s=60.0,
        tx_allowed_freq_ranges=[[2.4e9, 2.5e9], [5.7e9, 5.8e9]],
    )
    p = hardware.heartbeat_payload(cfg=cfg)
    assert p["tx_max_gain_db"] == -10.0
    assert p["tx_max_duration_s"] == 60.0
    assert p["tx_allowed_freq_ranges"] == [[2.4e9, 2.5e9], [5.7e9, 5.8e9]]


def test_heartbeat_payload_omits_caps_when_tx_disabled():
    from ria_toolkit_oss.agent.config import AgentConfig

    # Caps set but tx_enabled=False — don't leak them; they're only meaningful
    # when the hub can attempt a tx_start.
    cfg = AgentConfig(tx_enabled=False, tx_max_gain_db=-10.0)
    p = hardware.heartbeat_payload(cfg=cfg)
    assert "tx_max_gain_db" not in p
    assert "tx_max_duration_s" not in p
    assert "tx_allowed_freq_ranges" not in p