diff --git a/homeassistant/components/usb/__init__.py b/homeassistant/components/usb/__init__.py index c57516de9b7..6dc3a20cc7a 100644 --- a/homeassistant/components/usb/__init__.py +++ b/homeassistant/components/usb/__init__.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio from collections.abc import Callable, Coroutine, Sequence +import dataclasses from datetime import datetime, timedelta import logging import os @@ -167,6 +168,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: await usb_discovery.async_setup() hass.data[_USB_DATA] = usb_discovery websocket_api.async_register_command(hass, websocket_usb_scan) + websocket_api.async_register_command(hass, websocket_usb_list_serial_ports) return True @@ -481,3 +483,23 @@ async def websocket_usb_scan( """Scan for new usb devices.""" await async_request_scan(hass) connection.send_result(msg["id"]) + + +@websocket_api.require_admin +@websocket_api.websocket_command({vol.Required("type"): "usb/list_serial_ports"}) +@websocket_api.async_response +async def websocket_usb_list_serial_ports( + hass: HomeAssistant, + connection: ActiveConnection, + msg: dict[str, Any], +) -> None: + """List available serial ports.""" + try: + ports = await async_scan_serial_ports(hass) + except OSError as err: + connection.send_error(msg["id"], websocket_api.ERR_UNKNOWN_ERROR, str(err)) + return + connection.send_result( + msg["id"], + [dataclasses.asdict(port) for port in ports], + ) diff --git a/homeassistant/helpers/selector.py b/homeassistant/helpers/selector.py index 3194de03dc5..a306c264e4e 100644 --- a/homeassistant/helpers/selector.py +++ b/homeassistant/helpers/selector.py @@ -1771,6 +1771,28 @@ class SelectSelector(Selector[SelectSelectorConfig]): return [parent_schema(vol.Schema(str)(val)) for val in data] +class SerialSelectorConfig(BaseSelectorConfig): + """Class to represent a serial selector config.""" + + +@SELECTORS.register("serial") +class SerialSelector(Selector[SerialSelectorConfig]): + """Selector for a serial port.""" + + selector_type = "serial" + + CONFIG_SCHEMA = make_selector_config_schema() + + def __init__(self, config: SerialSelectorConfig | None = None) -> None: + """Instantiate a selector.""" + super().__init__(config) + + def __call__(self, data: Any) -> str: + """Validate the passed selection.""" + serial: str = vol.Schema(str)(data) + return serial + + class StateSelectorConfig(BaseSelectorConfig, total=False): """Class to represent an state selector config.""" diff --git a/tests/components/homeassistant_connect_zbt2/test_init.py b/tests/components/homeassistant_connect_zbt2/test_init.py index 09a89ab13fa..d8ea027db96 100644 --- a/tests/components/homeassistant_connect_zbt2/test_init.py +++ b/tests/components/homeassistant_connect_zbt2/test_init.py @@ -14,11 +14,8 @@ from homeassistant.setup import async_setup_component from homeassistant.util import dt as dt_util from tests.common import MockConfigEntry, async_fire_time_changed -from tests.components.usb import ( - async_request_scan, - force_usb_polling_watcher, # noqa: F401 - patch_scanned_serial_ports, -) +from tests.components.usb import async_request_scan, patch_scanned_serial_ports +from tests.components.usb.conftest import force_usb_polling_watcher # noqa: F401 async def test_setup_fails_on_missing_usb_port(hass: HomeAssistant) -> None: diff --git a/tests/components/homeassistant_sky_connect/test_init.py b/tests/components/homeassistant_sky_connect/test_init.py index 90c0cbbcc28..3809972e802 100644 --- a/tests/components/homeassistant_sky_connect/test_init.py +++ b/tests/components/homeassistant_sky_connect/test_init.py @@ -26,11 +26,8 @@ from homeassistant.setup import async_setup_component from homeassistant.util import dt as dt_util from tests.common import MockConfigEntry, async_fire_time_changed -from tests.components.usb import ( - async_request_scan, - force_usb_polling_watcher, # noqa: F401 - patch_scanned_serial_ports, -) +from tests.components.usb import async_request_scan, patch_scanned_serial_ports +from tests.components.usb.conftest import force_usb_polling_watcher # noqa: F401 async def test_config_entry_migration_v2(hass: HomeAssistant) -> None: diff --git a/tests/components/usb/__init__.py b/tests/components/usb/__init__.py index b849acdf0a9..28cd5358365 100644 --- a/tests/components/usb/__init__.py +++ b/tests/components/usb/__init__.py @@ -2,23 +2,10 @@ from unittest.mock import patch -from aiousbwatcher import InotifyNotAvailableError -import pytest - from homeassistant.components.usb import async_request_scan as usb_async_request_scan from homeassistant.core import HomeAssistant -@pytest.fixture(name="force_usb_polling_watcher") -def force_usb_polling_watcher(): - """Patch the USB integration to not use inotify and fall back to polling.""" - with patch( - "homeassistant.components.usb.AIOUSBWatcher.async_start", - side_effect=InotifyNotAvailableError, - ): - yield - - def patch_scanned_serial_ports(**kwargs) -> None: """Patch the USB integration's list of scanned serial ports.""" return patch("homeassistant.components.usb.utils.scan_serial_ports", **kwargs) diff --git a/tests/components/usb/conftest.py b/tests/components/usb/conftest.py new file mode 100644 index 00000000000..2aeba170ff7 --- /dev/null +++ b/tests/components/usb/conftest.py @@ -0,0 +1,36 @@ +"""Fixtures for USB Discovery integration tests.""" + +from unittest.mock import MagicMock, patch + +from aiousbwatcher import InotifyNotAvailableError +import pytest + +from homeassistant.components.usb import DOMAIN +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + +from . import patch_scanned_serial_ports + + +@pytest.fixture(name="force_usb_polling_watcher") +def force_usb_polling_watcher(): + """Patch the USB integration to not use inotify and fall back to polling.""" + with patch( + "homeassistant.components.usb.AIOUSBWatcher.async_start", + side_effect=InotifyNotAvailableError, + ): + yield + + +@pytest.fixture(name="setup_usb") +async def setup_usb_fixture( + hass: HomeAssistant, force_usb_polling_watcher: None +) -> MagicMock: + """Set up USB integration and return the scanned serial ports mock.""" + with ( + patch("homeassistant.components.usb.async_get_usb", return_value=[]), + patch_scanned_serial_ports(return_value=[]) as mock_serial_ports, + ): + assert await async_setup_component(hass, DOMAIN, {"usb": {}}) + await hass.async_block_till_done() + yield mock_serial_ports diff --git a/tests/components/usb/test_init.py b/tests/components/usb/test_init.py index b195ab88b34..698dc8f482d 100644 --- a/tests/components/usb/test_init.py +++ b/tests/components/usb/test_init.py @@ -22,13 +22,11 @@ from homeassistant.core import HomeAssistant from homeassistant.setup import async_setup_component from homeassistant.util import dt as dt_util -from . import ( - force_usb_polling_watcher, # noqa: F401 - patch_scanned_serial_ports, -) +from . import patch_scanned_serial_ports from tests.common import ( MockModule, + MockUser, async_fire_time_changed, mock_config_flow, mock_integration, @@ -1590,3 +1588,83 @@ async def test_removal_aborts_discovery_flows( final_flows = hass.config_entries.flow.async_progress() assert len(final_flows) == 1 assert final_flows[0]["handler"] == "test2" + + +async def test_list_serial_ports( + hass: HomeAssistant, + hass_ws_client: WebSocketGenerator, + setup_usb: MagicMock, +) -> None: + """Test listing serial ports via websocket.""" + setup_usb.return_value = [ + USBDevice( + device="/dev/ttyUSB0", + vid="10C4", + pid="EA60", + serial_number="001234", + manufacturer="Silicon Labs", + description="CP2102 USB to UART", + ), + SerialDevice( + device="/dev/ttyS0", + serial_number=None, + manufacturer=None, + description="ttyS0", + ), + ] + + ws_client = await hass_ws_client(hass) + await ws_client.send_json({"id": 1, "type": "usb/list_serial_ports"}) + response = await ws_client.receive_json() + + assert response["success"] + result = response["result"] + assert len(result) == 2 + + assert result[0]["device"] == "/dev/ttyUSB0" + assert result[0]["vid"] == "10C4" + assert result[0]["pid"] == "EA60" + assert result[0]["serial_number"] == "001234" + assert result[0]["manufacturer"] == "Silicon Labs" + assert result[0]["description"] == "CP2102 USB to UART" + + assert result[1]["device"] == "/dev/ttyS0" + assert result[1]["serial_number"] is None + assert result[1]["manufacturer"] is None + assert result[1]["description"] == "ttyS0" + assert "vid" not in result[1] + assert "pid" not in result[1] + + +async def test_list_serial_ports_require_admin( + hass: HomeAssistant, + hass_ws_client: WebSocketGenerator, + hass_admin_user: MockUser, + setup_usb: MagicMock, +) -> None: + """Test that listing serial ports requires admin.""" + hass_admin_user.groups = [] + + ws_client = await hass_ws_client(hass) + await ws_client.send_json({"id": 1, "type": "usb/list_serial_ports"}) + response = await ws_client.receive_json() + + assert not response["success"] + assert response["error"]["code"] == "unauthorized" + + +async def test_list_serial_ports_os_error( + hass: HomeAssistant, + hass_ws_client: WebSocketGenerator, + setup_usb: MagicMock, +) -> None: + """Test listing serial ports handles OSError.""" + setup_usb.side_effect = OSError("Permission denied") + + ws_client = await hass_ws_client(hass) + await ws_client.send_json({"id": 1, "type": "usb/list_serial_ports"}) + response = await ws_client.receive_json() + + assert not response["success"] + assert response["error"]["code"] == "unknown_error" + assert "Permission denied" in response["error"]["message"] diff --git a/tests/helpers/test_selector.py b/tests/helpers/test_selector.py index c6a6f7056bc..400dc49bf68 100644 --- a/tests/helpers/test_selector.py +++ b/tests/helpers/test_selector.py @@ -1122,6 +1122,22 @@ def test_state_selector_schema(schema, valid_selections, invalid_selections) -> _test_selector("state", schema, valid_selections, invalid_selections) +@pytest.mark.parametrize( + ("schema", "valid_selections", "invalid_selections"), + [ + (None, ("/dev/ttyUSB0", "/dev/ttyACM1", "COM3"), (None, 1, True)), + ({}, ("/dev/ttyUSB0",), (None,)), + ], +) +def test_serial_selector_schema( + schema: dict | None, + valid_selections: tuple[Any, ...], + invalid_selections: tuple[Any, ...], +) -> None: + """Test serial selector.""" + _test_selector("serial", schema, valid_selections, invalid_selections) + + @pytest.mark.parametrize( ("schema", "valid_selections", "invalid_selections"), [