Fix imports and add missing file

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
@@ -163,6 +164,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
 
@@ -477,3 +479,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],
+    )

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."""
 

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:

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:

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)

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

tests/components/usb/test_init.py

@@ -21,13 +21,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,
@@ -1646,3 +1644,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"]

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"),
     [