Skip to content

fix: declare resolved audio route in server info #4025

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
morozsm merged 4 commits into from
Oct 9, 2026
Merged

fix: declare resolved audio route in server info #4025

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3dc054e
test: pin info audio route declarations from backend facts
morozsm Oct 8, 2026
095841f
test: accept receiver keyword in info radio fixture
morozsm Oct 8, 2026
24b0837
test: format info route cases with pinned Ruff
morozsm Oct 8, 2026
4221410
fix: declare resolved audio route in server info
morozsm Oct 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion src/rigplane/web/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,13 +34,14 @@
import time
import urllib.parse
from collections.abc import Awaitable, Callable, Collection, Coroutine, Mapping
from dataclasses import dataclass, field
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from inspect import getattr_static
from typing import TYPE_CHECKING, Any, Literal, Protocol, TextIO, cast

from .. import __version__
from .._bounded_queue import BoundedQueue
from ..audio.route import resolve_audio_route
from ..core.acquisition_scheduler import (
AcquisitionScheduler,
MeterObservationCoalescer,
Expand Down Expand Up @@ -3887,6 +3888,11 @@ async def _serve_info(
# New structured fields
"model": model,
"audioReceive": self._audio_broadcaster.rx_start_status(),
"audioRoute": (
asdict(resolve_audio_route(self._radio))
if self._radio is not None
else None
),
"capabilities": capabilities,
"connection": {
"rigConnected": connected,
Expand Down
204 changes: 204 additions & 0 deletions tests/test_web_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import socket
import struct
import time
from dataclasses import replace
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
Expand Down Expand Up @@ -699,6 +700,209 @@ def test_make_frame_binary_large(self) -> None:


class TestHttpEndpoints:
@pytest.mark.parametrize(
"backend,usb,count,lan_input,supports,connected,expected",
[
(
"icom_serial",
False,
1,
False,
False,
True,
{
"radio_transport": "serial",
"tx_audio_source": "usb",
"rx_audio_source": "usb",
"data_mode_policy": "data1_usb",
"bridge_required": True,
},
),
(
"yaesu_cat",
False,
1,
False,
False,
True,
{
"radio_transport": "serial",
"tx_audio_source": "usb",
"rx_audio_source": "usb",
"data_mode_policy": "data1_usb",
"bridge_required": True,
},
),
(
None,
True,
1,
False,
False,
True,
{
"radio_transport": "serial",
"tx_audio_source": "usb",
"rx_audio_source": "usb",
"data_mode_policy": "data1_usb",
"bridge_required": True,
},
),
(
"rigplane",
True,
3,
False,
False,
True,
{
"radio_transport": "lan",
"tx_audio_source": "lan",
"rx_audio_source": "lan",
"data_mode_policy": "data2_lan",
"bridge_required": False,
},
),
(
"rigplane",
False,
1,
True,
True,
True,
{
"radio_transport": "lan",
"tx_audio_source": "lan",
"rx_audio_source": "lan",
"data_mode_policy": "data1_lan",
"bridge_required": False,
},
),
(
"rigplane",
False,
1,
True,
False,
True,
{
"radio_transport": "lan",
"tx_audio_source": "lan",
"rx_audio_source": "lan",
"data_mode_policy": "legacy",
"bridge_required": False,
},
),
(
"rigplane",
False,
1,
False,
True,
True,
{
"radio_transport": "lan",
"tx_audio_source": "lan",
"rx_audio_source": "lan",
"data_mode_policy": "legacy",
"bridge_required": False,
},
),
(
None,
False,
3,
True,
True,
True,
{
"radio_transport": "unknown",
"tx_audio_source": "unavailable",
"rx_audio_source": "unavailable",
"data_mode_policy": "legacy",
"bridge_required": False,
},
),
(
"icom_serial",
False,
1,
False,
False,
False,
{
"radio_transport": "serial",
"tx_audio_source": "usb",
"rx_audio_source": "usb",
"data_mode_policy": "data1_usb",
"bridge_required": True,
},
),
],
ids=[
"icom",
"yaesu",
"usb",
"lan-multi",
"lan-single",
"lan-no-command",
"lan-no-input",
"unknown",
"disconnected",
],
)
async def test_info_declares_audio_route(
self,
server: WebServer,
backend: str | None,
usb: bool,
count: int,
lan_input: bool,
supports: bool,
connected: bool,
expected: dict[str, object],
) -> None:
radio = server._radio
radio.backend_id = backend
radio.has_usb_audio = usb
radio.profile = replace(
resolve_radio_profile(model="IC-7610"),
data_mode_count=count,
data_mode_inputs=((5, "LAN"),) if lan_input else (),
)
radio.supports_command = lambda command, *, receiver=None: (
supports and command == "set_data1_mod_input"
)
radio.connected = connected
radio.control_connected = connected
radio.radio_ready = connected
host, port = _addr(server)
status, _, body = await _http_get(host, port, "/api/v1/info")
data = json.loads(body)
assert status == 200
assert data["audioRoute"] == expected
assert data["server"] == "rigplane"
assert isinstance(data["version"], str)
assert data["proto"] == 1
assert data["radio"] == data["model"] == "IC-7610"
assert data["capabilities"]["dataModeCount"] == count
assert data["connection"]["rigConnected"] is connected
assert data["connection"]["controlConnected"] is connected
assert data["connection"]["radioReady"] is connected
assert "audioReceive" in data
assert "backend" not in data and "audio_route" not in data

async def test_info_audio_route_null_without_radio(self, server: WebServer) -> None:
server._radio = None
host, port = _addr(server)
status, _, body = await _http_get(host, port, "/api/v1/info")
data = json.loads(body)
assert status == 200
assert data["audioRoute"] is None
assert data["proto"] == 1
assert data["connection"]["rigConnected"] is False
assert data["connection"]["radioReady"] is False

async def test_info_endpoint_status(self, server: WebServer) -> None:
host, port = _addr(server)
status, _, _ = await _http_get(host, port, "/api/v1/info")
Expand Down
Loading