arian/home-assistant-core
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

move async_get_usb_ports to usb component

Browse Source
This commit is contained in:
Petar Petrov
2025-09-12 09:17:15 +03:00
parent 1cf49096ce
commit 69e5bda0bd
6 changed files with 254 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/otbr/config_flow.py
+2 -41
View File
@@ -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):
homeassistant/components/usb/__init__.py
+46 -4
View File
@@ -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(
homeassistant/components/zwave_js/config_flow.py
+1 -39
View File
@@ -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."""
tests/components/otbr/test_config_flow.py
-70
View File
@@ -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"}
tests/components/usb/test_ports.py
+204
View File
@@ -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()
tests/components/zwave_js/test_config_flow.py
+1 -121
View File
@@ -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,