diff --git a/homeassistant/components/otbr/config_flow.py b/homeassistant/components/otbr/config_flow.py index ea4ffe2131c..43638102cf6 100644 --- a/homeassistant/components/otbr/config_flow.py +++ b/homeassistant/components/otbr/config_flow.py @@ -12,11 +12,9 @@ from aiohttp.client_exceptions import ClientConnectorError import python_otbr_api from python_otbr_api import tlv_parser from python_otbr_api.tlv_parser import MeshcopTLVType -from serial.tools import list_ports import voluptuous as vol import yarl -from homeassistant.components import usb from homeassistant.components.hassio import ( AddonError, AddonInfo, @@ -26,6 +24,7 @@ from homeassistant.components.hassio import ( from homeassistant.components.homeassistant_hardware.util import get_otbr_addon_manager from homeassistant.components.homeassistant_yellow import hardware as yellow_hardware from homeassistant.components.thread import async_get_preferred_dataset +from homeassistant.components.usb import async_get_usb_ports from homeassistant.config_entries import ( SOURCE_HASSIO, ConfigEntryState, @@ -97,44 +96,6 @@ async def _title(hass: HomeAssistant, discovery_info: HassioServiceInfo) -> str: return discovery_info.name -def get_usb_ports() -> dict[str, str]: - """Return a dict of USB ports and their friendly names.""" - ports = list_ports.comports() - port_descriptions = {} - for port in ports: - vid: str | None = None - pid: str | None = None - if port.vid is not None and port.pid is not None: - usb_device = usb.usb_device_from_port(port) - vid = usb_device.vid - pid = usb_device.pid - dev_path = usb.get_serial_by_id(port.device) - human_name = usb.human_readable_device_name( - dev_path, - port.serial_number, - port.manufacturer, - port.description, - vid, - pid, - ) - port_descriptions[dev_path] = human_name - - # Filter out "n/a" descriptions only if there are other ports available - non_na_ports = { - path: desc - for path, desc in port_descriptions.items() - if not desc.lower().startswith("n/a") - } - - # If we have non-"n/a" ports, return only those; otherwise return all ports as-is - return non_na_ports if non_na_ports else port_descriptions - - -async def async_get_usb_ports(hass: HomeAssistant) -> dict[str, str]: - """Return a dict of USB ports and their friendly names.""" - return await hass.async_add_executor_job(get_usb_ports) - - class OTBRConfigFlow(ConfigFlow, domain=DOMAIN): """Handle a config flow for Open Thread Border Router.""" @@ -211,7 +172,7 @@ class OTBRConfigFlow(ConfigFlow, domain=DOMAIN): Returns the router's border agent id. """ api = python_otbr_api.OTBR(otbr_url, async_get_clientsession(self.hass), 10) - border_agent_id = await api.get_border_agent_id() + border_agent_id: bytes = await api.get_border_agent_id() _LOGGER.debug("border agent id for url %s: %s", otbr_url, border_agent_id.hex()) if await self._is_border_agent_id_configured(border_agent_id): diff --git a/homeassistant/components/usb/__init__.py b/homeassistant/components/usb/__init__.py index 90433b0f728..7d535910a84 100644 --- a/homeassistant/components/usb/__init__.py +++ b/homeassistant/components/usb/__init__.py @@ -14,6 +14,7 @@ import sys from typing import Any, overload from aiousbwatcher import AIOUSBWatcher, InotifyNotAvailableError +from serial.tools import list_ports import voluptuous as vol from homeassistant import config_entries @@ -41,10 +42,7 @@ from homeassistant.loader import USBMatcher, async_get_usb from .const import DOMAIN from .models import USBDevice -from .utils import ( - scan_serial_ports, - usb_device_from_port, # noqa: F401 -) +from .utils import scan_serial_ports, usb_device_from_port _LOGGER = logging.getLogger(__name__) @@ -529,6 +527,50 @@ async def websocket_usb_scan( connection.send_result(msg["id"]) +def get_usb_ports() -> dict[str, str]: + """Return a dict of USB ports and their friendly names.""" + + ports = list_ports.comports() + port_descriptions = {} + for port in ports: + vid: str | None = None + pid: str | None = None + if port.vid is not None and port.pid is not None: + usb_device = usb_device_from_port(port) + vid = usb_device.vid + pid = usb_device.pid + + dev_path = get_serial_by_id(port.device) + human_name = human_readable_device_name( + dev_path, + port.serial_number, + port.manufacturer, + port.description, + vid, + pid, + ) + port_descriptions[dev_path] = human_name + + # Filter out "n/a" descriptions only if there are other ports available + non_na_ports = { + path: desc + for path, desc in port_descriptions.items() + if not desc.lower().startswith("n/a") + } + + # If we have non-"n/a" ports, return only those; otherwise return all ports as-is + return non_na_ports if non_na_ports else port_descriptions + + +async def async_get_usb_ports(hass: HomeAssistant) -> dict[str, str]: + """Return a dict of USB ports and their friendly names.""" + try: + return await hass.async_add_executor_job(get_usb_ports) + except OSError: + _LOGGER.warning("Failed to scan USB ports", exc_info=True) + return {} + + # These can be removed if no deprecated constant are in this module anymore __getattr__ = partial(check_if_deprecated_constant, module_globals=globals()) __dir__ = partial( diff --git a/homeassistant/components/zwave_js/config_flow.py b/homeassistant/components/zwave_js/config_flow.py index 92912a2cdb5..8d9d4dbe31f 100644 --- a/homeassistant/components/zwave_js/config_flow.py +++ b/homeassistant/components/zwave_js/config_flow.py @@ -11,7 +11,6 @@ from pathlib import Path from typing import Any from awesomeversion import AwesomeVersion -from serial.tools import list_ports import voluptuous as vol from zwave_js_server.client import Client from zwave_js_server.exceptions import FailedCommand @@ -25,6 +24,7 @@ from homeassistant.components.hassio import ( AddonManager, AddonState, ) +from homeassistant.components.usb import async_get_usb_ports from homeassistant.config_entries import ( SOURCE_USB, ConfigEntryState, @@ -145,44 +145,6 @@ async def validate_input(hass: HomeAssistant, user_input: dict) -> VersionInfo: raise InvalidInput("cannot_connect") from err -def get_usb_ports() -> dict[str, str]: - """Return a dict of USB ports and their friendly names.""" - ports = list_ports.comports() - port_descriptions = {} - for port in ports: - vid: str | None = None - pid: str | None = None - if port.vid is not None and port.pid is not None: - usb_device = usb.usb_device_from_port(port) - vid = usb_device.vid - pid = usb_device.pid - dev_path = usb.get_serial_by_id(port.device) - human_name = usb.human_readable_device_name( - dev_path, - port.serial_number, - port.manufacturer, - port.description, - vid, - pid, - ) - port_descriptions[dev_path] = human_name - - # Filter out "n/a" descriptions only if there are other ports available - non_na_ports = { - path: desc - for path, desc in port_descriptions.items() - if not desc.lower().startswith("n/a") - } - - # If we have non-"n/a" ports, return only those; otherwise return all ports as-is - return non_na_ports if non_na_ports else port_descriptions - - -async def async_get_usb_ports(hass: HomeAssistant) -> dict[str, str]: - """Return a dict of USB ports and their friendly names.""" - return await hass.async_add_executor_job(get_usb_ports) - - class ZWaveJSConfigFlow(ConfigFlow, domain=DOMAIN): """Handle a config flow for Z-Wave JS.""" diff --git a/tests/components/otbr/test_config_flow.py b/tests/components/otbr/test_config_flow.py index 9aa2abb0f3c..337a2dc9950 100644 --- a/tests/components/otbr/test_config_flow.py +++ b/tests/components/otbr/test_config_flow.py @@ -19,7 +19,6 @@ from homeassistant.components.homeassistant_hardware.util import ( FirmwareInfo, OwningAddon, ) -from homeassistant.components.otbr.config_flow import get_usb_ports from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType from homeassistant.exceptions import HomeAssistantError @@ -2107,72 +2106,3 @@ async def test_complete_recommended_flow_success( # Additional tests to improve coverage - - -async def test_get_usb_ports_with_vid_pid() -> None: - """Test get_usb_ports with VID/PID information.""" - mock_port = Mock() - mock_port.device = "/dev/ttyUSB0" - mock_port.serial_number = "12345" - mock_port.manufacturer = "Test" - mock_port.description = "Valid Device" - mock_port.vid = 0x1234 - mock_port.pid = 0x5678 - - mock_usb_device = Mock() - mock_usb_device.vid = "1234" - mock_usb_device.pid = "5678" - - with ( - patch("serial.tools.list_ports.comports", return_value=[mock_port]), - patch( - "homeassistant.components.otbr.config_flow.usb.get_serial_by_id", - return_value="/dev/ttyUSB0", - ), - patch( - "homeassistant.components.otbr.config_flow.usb.usb_device_from_port", - return_value=mock_usb_device, - ), - patch( - "homeassistant.components.otbr.config_flow.usb.human_readable_device_name", - return_value="Valid Device", - ), - ): - result = get_usb_ports() - assert result == {"/dev/ttyUSB0": "Valid Device"} - - -async def test_get_usb_ports_filtering_mixed_ports() -> None: - """Test get_usb_ports filtering with mixed valid and 'n/a' ports.""" - mock_port1 = Mock() - mock_port1.device = "/dev/ttyUSB0" - mock_port1.serial_number = "12345" - mock_port1.manufacturer = "Test" - mock_port1.description = "Valid Device" - mock_port1.vid = None - mock_port1.pid = None - - mock_port2 = Mock() - mock_port2.device = "/dev/ttyUSB1" - mock_port2.serial_number = "67890" - mock_port2.manufacturer = "Test" - mock_port2.description = "n/a" - mock_port2.vid = None - mock_port2.pid = None - - with ( - patch( - "serial.tools.list_ports.comports", return_value=[mock_port1, mock_port2] - ), - patch( - "homeassistant.components.otbr.config_flow.usb.get_serial_by_id", - side_effect=["/dev/ttyUSB0", "/dev/ttyUSB1"], - ), - patch( - "homeassistant.components.otbr.config_flow.usb.human_readable_device_name", - side_effect=["Valid Device", "n/a"], - ), - ): - result = get_usb_ports() - # Should filter out the "n/a" port and only return the valid one - assert result == {"/dev/ttyUSB0": "Valid Device"} diff --git a/tests/components/usb/test_ports.py b/tests/components/usb/test_ports.py new file mode 100644 index 00000000000..acb137aae10 --- /dev/null +++ b/tests/components/usb/test_ports.py @@ -0,0 +1,204 @@ +"""Test USB utils.""" + +from unittest.mock import Mock, patch + +from serial.tools.list_ports_common import ListPortInfo + +from homeassistant.components.usb import async_get_usb_ports, get_usb_ports +from homeassistant.core import HomeAssistant + + +async def test_get_usb_ports_with_vid_pid() -> None: + """Test get_usb_ports with VID/PID information.""" + mock_port = Mock() + mock_port.device = "/dev/ttyUSB0" + mock_port.serial_number = "12345" + mock_port.manufacturer = "Test" + mock_port.description = "Valid Device" + mock_port.vid = 0x1234 + mock_port.pid = 0x5678 + + mock_usb_device = Mock() + mock_usb_device.vid = "1234" + mock_usb_device.pid = "5678" + + with ( + patch("serial.tools.list_ports.comports", return_value=[mock_port]), + patch( + "homeassistant.components.usb.get_serial_by_id", + return_value="/dev/ttyUSB0", + ), + patch( + "homeassistant.components.usb.utils.usb_device_from_port", + return_value=mock_usb_device, + ), + patch( + "homeassistant.components.usb.human_readable_device_name", + return_value="Valid Device", + ), + ): + result = get_usb_ports() + assert result == {"/dev/ttyUSB0": "Valid Device"} + + +async def test_get_usb_ports_filtering_mixed_ports() -> None: + """Test get_usb_ports filtering with mixed valid and 'n/a' ports.""" + mock_port1 = Mock() + mock_port1.device = "/dev/ttyUSB0" + mock_port1.serial_number = "12345" + mock_port1.manufacturer = "Test" + mock_port1.description = "Valid Device" + mock_port1.vid = None + mock_port1.pid = None + + mock_port2 = Mock() + mock_port2.device = "/dev/ttyUSB1" + mock_port2.serial_number = "67890" + mock_port2.manufacturer = "Test" + mock_port2.description = "n/a" + mock_port2.vid = None + mock_port2.pid = None + + with ( + patch( + "serial.tools.list_ports.comports", return_value=[mock_port1, mock_port2] + ), + patch( + "homeassistant.components.usb.get_serial_by_id", + side_effect=["/dev/ttyUSB0", "/dev/ttyUSB1"], + ), + patch( + "homeassistant.components.usb.human_readable_device_name", + side_effect=["Valid Device", "n/a"], + ), + ): + result = get_usb_ports() + # Should filter out the "n/a" port and only return the valid one + assert result == {"/dev/ttyUSB0": "Valid Device"} + + +async def test_get_usb_ports_filtering() -> None: + """Test that get_usb_ports filters out 'n/a' descriptions when other ports are available.""" + + mock_ports = [ + ListPortInfo("/dev/ttyUSB0"), + ListPortInfo("/dev/ttyUSB1"), + ListPortInfo("/dev/ttyUSB2"), + ListPortInfo("/dev/ttyUSB3"), + ] + mock_ports[0].description = "n/a" + mock_ports[1].description = "Device A" + mock_ports[2].description = "N/A" + mock_ports[3].description = "Device B" + + with patch("serial.tools.list_ports.comports", return_value=mock_ports): + result = get_usb_ports() + + descriptions = list(result.values()) + + # Verify that only non-"n/a" descriptions are returned + assert descriptions == [ + "Device A - /dev/ttyUSB1, s/n: n/a", + "Device B - /dev/ttyUSB3, s/n: n/a", + ] + + +async def test_get_usb_ports_all_na() -> None: + """Test that get_usb_ports returns all ports as-is when only 'n/a' descriptions exist.""" + + mock_ports = [ + ListPortInfo("/dev/ttyUSB0"), + ListPortInfo("/dev/ttyUSB1"), + ListPortInfo("/dev/ttyUSB2"), + ] + mock_ports[0].description = "n/a" + mock_ports[1].description = "N/A" + mock_ports[2].description = "n/a" + + with patch("serial.tools.list_ports.comports", return_value=mock_ports): + result = get_usb_ports() + + descriptions = list(result.values()) + + # Verify that all ports are returned since they all have "n/a" descriptions + assert len(descriptions) == 3 + # Verify that all descriptions contain "n/a" (case-insensitive) + assert all("n/a" in desc.lower() for desc in descriptions) + # Verify that all expected device paths are present + device_paths = [desc.split(" - ")[1].split(",")[0] for desc in descriptions] + assert "/dev/ttyUSB0" in device_paths + assert "/dev/ttyUSB1" in device_paths + assert "/dev/ttyUSB2" in device_paths + + +async def test_get_usb_ports_mixed_case_filtering() -> None: + """Test that get_usb_ports filters out 'n/a' descriptions with different case variations.""" + + mock_ports = [ + ListPortInfo("/dev/ttyUSB0"), + ListPortInfo("/dev/ttyUSB1"), + ListPortInfo("/dev/ttyUSB2"), + ListPortInfo("/dev/ttyUSB3"), + ] + mock_ports[0].description = "n/a" + mock_ports[1].description = "Not Available" + mock_ports[2].description = "N/A" + mock_ports[3].description = "Device B" + + with patch("serial.tools.list_ports.comports", return_value=mock_ports): + result = get_usb_ports() + + descriptions = list(result.values()) + + # Verify that only non-"n/a" descriptions are returned + assert descriptions == [ + "Not Available - /dev/ttyUSB1, s/n: n/a", + "Device B - /dev/ttyUSB3, s/n: n/a", + ] + + +async def test_get_usb_ports_empty_list() -> None: + """Test that get_usb_ports handles empty port list.""" + with patch("serial.tools.list_ports.comports", return_value=[]): + result = get_usb_ports() + assert result == {} + + +async def test_get_usb_ports_single_na_port() -> None: + """Test that get_usb_ports returns single 'n/a' port when it's the only one available.""" + + mock_port = ListPortInfo("/dev/ttyUSB0") + mock_port.description = "n/a" + + with patch("serial.tools.list_ports.comports", return_value=[mock_port]): + result = get_usb_ports() + assert len(result) == 1 + assert "/dev/ttyUSB0" in result + assert "n/a" in result["/dev/ttyUSB0"].lower() + + +async def test_get_usb_ports_single_valid_port() -> None: + """Test that get_usb_ports returns single valid port.""" + + mock_port = ListPortInfo("/dev/ttyUSB0") + mock_port.description = "Valid Device" + + with patch("serial.tools.list_ports.comports", return_value=[mock_port]): + result = get_usb_ports() + assert len(result) == 1 + assert "/dev/ttyUSB0" in result + assert "Valid Device" in result["/dev/ttyUSB0"] + + +async def test_async_get_usb_ports_exception_handling(hass: HomeAssistant) -> None: + """Test async_get_usb_ports exception handling.""" + with ( + patch( + "homeassistant.components.usb.get_usb_ports", + side_effect=OSError("USB scan failed"), + ), + patch("homeassistant.components.usb._LOGGER.warning") as mock_logger, + ): + result = await async_get_usb_ports(hass) + assert result == {} + mock_logger.assert_called_once() diff --git a/tests/components/zwave_js/test_config_flow.py b/tests/components/zwave_js/test_config_flow.py index bab13666a29..cfb924f3807 100644 --- a/tests/components/zwave_js/test_config_flow.py +++ b/tests/components/zwave_js/test_config_flow.py @@ -19,7 +19,7 @@ from zwave_js_server.model.node import Node from zwave_js_server.version import VersionInfo from homeassistant import config_entries, data_entry_flow -from homeassistant.components.zwave_js.config_flow import TITLE, get_usb_ports +from homeassistant.components.zwave_js.config_flow import TITLE from homeassistant.components.zwave_js.const import ( ADDON_SLUG, CONF_ADDON_DEVICE, @@ -4290,126 +4290,6 @@ async def test_configure_addon_usb_ports_failure( assert result["reason"] == "usb_ports_failed" -async def test_get_usb_ports_filtering() -> None: - """Test that get_usb_ports filters out 'n/a' descriptions when other ports are available.""" - mock_ports = [ - ListPortInfo("/dev/ttyUSB0"), - ListPortInfo("/dev/ttyUSB1"), - ListPortInfo("/dev/ttyUSB2"), - ListPortInfo("/dev/ttyUSB3"), - ] - mock_ports[0].description = "n/a" - mock_ports[1].description = "Device A" - mock_ports[2].description = "N/A" - mock_ports[3].description = "Device B" - - with patch("serial.tools.list_ports.comports", return_value=mock_ports): - result = get_usb_ports() - - descriptions = list(result.values()) - - # Verify that only non-"n/a" descriptions are returned - assert descriptions == [ - "Device A - /dev/ttyUSB1, s/n: n/a", - "Device B - /dev/ttyUSB3, s/n: n/a", - ] - - -async def test_get_usb_ports_all_na() -> None: - """Test that get_usb_ports returns all ports as-is when only 'n/a' descriptions exist.""" - mock_ports = [ - ListPortInfo("/dev/ttyUSB0"), - ListPortInfo("/dev/ttyUSB1"), - ListPortInfo("/dev/ttyUSB2"), - ] - mock_ports[0].description = "n/a" - mock_ports[1].description = "N/A" - mock_ports[2].description = "n/a" - - with patch("serial.tools.list_ports.comports", return_value=mock_ports): - result = get_usb_ports() - - descriptions = list(result.values()) - - # Verify that all ports are returned since they all have "n/a" descriptions - assert len(descriptions) == 3 - # Verify that all descriptions contain "n/a" (case-insensitive) - assert all("n/a" in desc.lower() for desc in descriptions) - # Verify that all expected device paths are present - device_paths = [desc.split(" - ")[1].split(",")[0] for desc in descriptions] - assert "/dev/ttyUSB0" in device_paths - assert "/dev/ttyUSB1" in device_paths - assert "/dev/ttyUSB2" in device_paths - - -async def test_get_usb_ports_mixed_case_filtering() -> None: - """Test that get_usb_ports filters out 'n/a' descriptions with different case variations.""" - mock_ports = [ - ListPortInfo("/dev/ttyUSB0"), - ListPortInfo("/dev/ttyUSB1"), - ListPortInfo("/dev/ttyUSB2"), - ListPortInfo("/dev/ttyUSB3"), - ListPortInfo("/dev/ttyUSB4"), - ] - mock_ports[0].description = "n/a" - mock_ports[1].description = "Device A" - mock_ports[2].description = "N/A" - mock_ports[3].description = "n/A" - mock_ports[4].description = "Device B" - - with patch("serial.tools.list_ports.comports", return_value=mock_ports): - result = get_usb_ports() - - descriptions = list(result.values()) - - # Verify that only non-"n/a" descriptions are returned (case-insensitive filtering) - assert descriptions == [ - "Device A - /dev/ttyUSB1, s/n: n/a", - "Device B - /dev/ttyUSB4, s/n: n/a", - ] - - -async def test_get_usb_ports_empty_list() -> None: - """Test that get_usb_ports handles empty port list.""" - with patch("serial.tools.list_ports.comports", return_value=[]): - result = get_usb_ports() - - # Verify that empty dict is returned - assert result == {} - - -async def test_get_usb_ports_single_na_port() -> None: - """Test that get_usb_ports returns single 'n/a' port when it's the only one available.""" - mock_ports = [ListPortInfo("/dev/ttyUSB0")] - mock_ports[0].description = "n/a" - - with patch("serial.tools.list_ports.comports", return_value=mock_ports): - result = get_usb_ports() - - descriptions = list(result.values()) - - # Verify that the single "n/a" port is returned - assert descriptions == [ - "n/a - /dev/ttyUSB0, s/n: n/a", - ] - - -async def test_get_usb_ports_single_valid_port() -> None: - """Test that get_usb_ports returns single valid port.""" - mock_ports = [ListPortInfo("/dev/ttyUSB0")] - mock_ports[0].description = "Device A" - - with patch("serial.tools.list_ports.comports", return_value=mock_ports): - result = get_usb_ports() - - descriptions = list(result.values()) - - # Verify that the single valid port is returned - assert descriptions == [ - "Device A - /dev/ttyUSB0, s/n: n/a", - ] - - @pytest.mark.usefixtures("supervisor", "addon_not_installed", "addon_info") async def test_intent_recommended_user( hass: HomeAssistant,