From d2fddf129daec4805c9cb0808d60dae4cf43b8ad Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Tue, 28 Apr 2026 14:50:31 -0400 Subject: [PATCH] Include matching integrations in scanned ports WS API (#169387) --- homeassistant/components/usb/__init__.py | 20 +++- homeassistant/components/zha/config_flow.py | 11 +- homeassistant/helpers/selector.py | 10 +- tests/components/usb/test_init.py | 120 ++++++++++++++------ tests/helpers/test_selector.py | 11 ++ 5 files changed, 131 insertions(+), 41 deletions(-) diff --git a/homeassistant/components/usb/__init__.py b/homeassistant/components/usb/__init__.py index 5076499ddae..25152a43382 100644 --- a/homeassistant/components/usb/__init__.py +++ b/homeassistant/components/usb/__init__.py @@ -555,7 +555,19 @@ async def websocket_usb_list_serial_ports( 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], - ) + + result = [] + for port in ports: + entry = dataclasses.asdict(port) + + if isinstance(port, USBDevice): + matchers = async_get_usb_matchers_for_device(hass, port) + entry["matching_integrations"] = list( + dict.fromkeys(matcher["domain"] for matcher in matchers) + ) + else: + entry["matching_integrations"] = [] + + result.append(entry) + + connection.send_result(msg["id"], result) diff --git a/homeassistant/components/zha/config_flow.py b/homeassistant/components/zha/config_flow.py index 4a58d56dfd4..12472fc990d 100644 --- a/homeassistant/components/zha/config_flow.py +++ b/homeassistant/components/zha/config_flow.py @@ -41,6 +41,7 @@ from homeassistant.helpers.selector import ( FileSelector, FileSelectorConfig, SerialPortSelector, + SerialPortSelectorConfig, ) from homeassistant.helpers.service_info.usb import UsbServiceInfo from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo @@ -209,7 +210,15 @@ class BaseZhaFlow(ConfigEntryBaseFlow): { vol.Required( CONF_DEVICE_PATH, default=default_path - ): SerialPortSelector(), + ): SerialPortSelector( + SerialPortSelectorConfig( + extra_recommended_domains=[ + "homeassistant_yellow", + "homeassistant_sky_connect", + "homeassistant_connect_zbt2", + ] + ) + ), } ) return self.async_show_form(step_id="choose_serial_port", data_schema=schema) diff --git a/homeassistant/helpers/selector.py b/homeassistant/helpers/selector.py index b885081f83b..51c79596a78 100644 --- a/homeassistant/helpers/selector.py +++ b/homeassistant/helpers/selector.py @@ -1771,9 +1771,11 @@ class SelectSelector(Selector[SelectSelectorConfig]): return [parent_schema(vol.Schema(str)(val)) for val in data] -class SerialPortSelectorConfig(BaseSelectorConfig): +class SerialPortSelectorConfig(BaseSelectorConfig, total=False): """Class to represent a serial port selector config.""" + extra_recommended_domains: list[str] + @SELECTORS.register("serial_port") class SerialPortSelector(Selector[SerialPortSelectorConfig]): @@ -1781,7 +1783,11 @@ class SerialPortSelector(Selector[SerialPortSelectorConfig]): selector_type = "serial_port" - CONFIG_SCHEMA = make_selector_config_schema() + CONFIG_SCHEMA = make_selector_config_schema( + { + vol.Optional("extra_recommended_domains"): [str], + } + ) def __init__(self, config: SerialPortSelectorConfig | None = None) -> None: """Instantiate a selector.""" diff --git a/tests/components/usb/test_init.py b/tests/components/usb/test_init.py index 32e6eb816f5..2a49e2a728f 100644 --- a/tests/components/usb/test_init.py +++ b/tests/components/usb/test_init.py @@ -1169,9 +1169,7 @@ async def test_register_port_event_callback( mock_callback2 = Mock() # Start off with no ports - with ( - patch_scanned_serial_ports(return_value=[]), - ): + with patch_scanned_serial_ports(return_value=[]): assert await async_setup_component(hass, DOMAIN, {"usb": {}}) _cancel1 = usb.async_register_port_event_callback(hass, mock_callback1) @@ -1264,9 +1262,7 @@ async def test_register_port_event_callback_failure( mock_callback2 = Mock(side_effect=RuntimeError("Failure 2")) # Start off with no ports - with ( - patch_scanned_serial_ports(return_value=[]), - ): + with patch_scanned_serial_ports(return_value=[]): assert await async_setup_component(hass, DOMAIN, {"usb": {}}) usb.async_register_port_event_callback(hass, mock_callback1) @@ -1679,13 +1675,22 @@ async def test_removal_aborts_discovery_flows( assert final_flows[0]["handler"] == "test2" +@pytest.mark.usefixtures("force_usb_polling_watcher") 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 = [ + matchers = [ + { + "description": "*cp2102*", + "domain": "homeassistant_sky_connect", + "pid": "EA60", + "vid": "10C4", + }, + {"domain": "custom_component", "vid": "DEAD", "pid": "BEEF"}, + ] + mock_ports = [ USBDevice( device="/dev/ttyUSB0", vid="10C4", @@ -1697,6 +1702,22 @@ async def test_list_serial_ports( interface_description="CP2102 USB to UART Bridge", interface_num=0, ), + USBDevice( + device="/dev/ttyUSB1", + vid="DEAD", + pid="BEEF", + serial_number=None, + manufacturer=None, + description="Unknown adapter", + ), + USBDevice( + device="/dev/ttyUSB2", + vid="0000", + pid="0000", + serial_number=None, + manufacturer=None, + description="No matchers", + ), SerialDevice( device="/dev/ttyS0", serial_number=None, @@ -1705,34 +1726,65 @@ async def test_list_serial_ports( ), ] - 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() + with ( + patch("homeassistant.components.usb.async_get_usb", return_value=matchers), + patch_scanned_serial_ports(return_value=mock_ports), + ): + assert await async_setup_component(hass, DOMAIN, {"usb": {}}) + await hass.async_block_till_done() + + 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", - "vid": "10C4", - "pid": "EA60", - "serial_number": "001234", - "manufacturer": "Silicon Labs", - "description": "CP2102 USB to UART", - "bcd_device": 257, - "interface_description": "CP2102 USB to UART Bridge", - "interface_num": 0, - } - - assert result[1] == { - "device": "/dev/ttyS0", - "serial_number": None, - "manufacturer": None, - "description": "ttyS0", - "interface_description": None, - "interface_num": None, - } + assert response["result"] == [ + { + "device": "/dev/ttyUSB0", + "vid": "10C4", + "pid": "EA60", + "serial_number": "001234", + "manufacturer": "Silicon Labs", + "description": "CP2102 USB to UART", + "bcd_device": 257, + "interface_description": "CP2102 USB to UART Bridge", + "interface_num": 0, + "matching_integrations": ["homeassistant_sky_connect"], + }, + { + "device": "/dev/ttyUSB1", + "vid": "DEAD", + "pid": "BEEF", + "serial_number": None, + "manufacturer": None, + "description": "Unknown adapter", + "bcd_device": None, + "interface_description": None, + "interface_num": None, + "matching_integrations": ["custom_component"], + }, + { + "device": "/dev/ttyUSB2", + "vid": "0000", + "pid": "0000", + "serial_number": None, + "manufacturer": None, + "description": "No matchers", + "bcd_device": None, + "interface_description": None, + "interface_num": None, + "matching_integrations": [], + }, + { + "device": "/dev/ttyS0", + "serial_number": None, + "manufacturer": None, + "description": "ttyS0", + "interface_description": None, + "interface_num": None, + "matching_integrations": [], + }, + ] async def test_list_serial_ports_require_admin( diff --git a/tests/helpers/test_selector.py b/tests/helpers/test_selector.py index 7641bd90f08..eed983b9b20 100644 --- a/tests/helpers/test_selector.py +++ b/tests/helpers/test_selector.py @@ -1127,6 +1127,17 @@ def test_state_selector_schema(schema, valid_selections, invalid_selections) -> [ (None, ("/dev/ttyUSB0", "/dev/ttyACM1", "COM3"), (None, 1, True)), ({}, ("/dev/ttyUSB0",), (None,)), + ( + { + "extra_recommended_domains": [ + "homeassistant_yellow", + "homeassistant_sky_connect", + ] + }, + ("/dev/ttyUSB0",), + (None,), + ), + ({"extra_recommended_domains": []}, ("/dev/ttyUSB0",), (None,)), ], ) def test_serial_port_selector_schema(