Add helper methods to simplify USB integration testing (#141733)

* Add some helper methods to simplify USB integration testing

* Re-export `usb_device_from_port`

homeassistant/components/usb/__init__.py

@@ -14,8 +14,6 @@ import sys
 from typing import Any, overload
 
 from aiousbwatcher import AIOUSBWatcher, InotifyNotAvailableError
-from serial.tools.list_ports import comports
-from serial.tools.list_ports_common import ListPortInfo
 import voluptuous as vol
 
 from homeassistant import config_entries
@@ -43,7 +41,10 @@ from homeassistant.loader import USBMatcher, async_get_usb
 
 from .const import DOMAIN
 from .models import USBDevice
-from .utils import usb_device_from_port
+from .utils import (
+    scan_serial_ports,
+    usb_device_from_port,  # noqa: F401
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -241,6 +242,13 @@ def _is_matching(device: USBDevice, matcher: USBMatcher | USBCallbackMatcher) ->
     return True
 
 
+async def async_request_scan(hass: HomeAssistant) -> None:
+    """Request a USB scan."""
+    usb_discovery: USBDiscovery = hass.data[DOMAIN]
+    if not usb_discovery.observer_active:
+        await usb_discovery.async_request_scan()
+
+
 class USBDiscovery:
     """Manage USB Discovery."""
 
@@ -417,14 +425,8 @@ class USBDiscovery:
                 service_info,
             )
 
-    async def _async_process_ports(self, ports: Sequence[ListPortInfo]) -> None:
+    async def _async_process_ports(self, usb_devices: Sequence[USBDevice]) -> None:
         """Process each discovered port."""
-        _LOGGER.debug("Processing ports: %r", ports)
-        usb_devices = {
-            usb_device_from_port(port)
-            for port in ports
-            if port.vid is not None or port.pid is not None
-        }
         _LOGGER.debug("USB devices: %r", usb_devices)
 
         # CP2102N chips create *two* serial ports on macOS: `/dev/cu.usbserial-` and
@@ -436,7 +438,7 @@ class USBDiscovery:
                 if dev.device.startswith("/dev/cu.SLAB_USBtoUART")
             }
 
-            usb_devices = {
+            filtered_usb_devices = {
                 dev
                 for dev in usb_devices
                 if dev.serial_number not in silabs_serials
@@ -445,10 +447,12 @@ class USBDiscovery:
                     and dev.device.startswith("/dev/cu.SLAB_USBtoUART")
                 )
             }
+        else:
+            filtered_usb_devices = set(usb_devices)
 
-        added_devices = usb_devices - self._last_processed_devices
-        removed_devices = self._last_processed_devices - usb_devices
-        self._last_processed_devices = usb_devices
+        added_devices = filtered_usb_devices - self._last_processed_devices
+        removed_devices = self._last_processed_devices - filtered_usb_devices
+        self._last_processed_devices = filtered_usb_devices
 
         _LOGGER.debug(
             "Added devices: %r, removed devices: %r", added_devices, removed_devices
@@ -461,7 +465,7 @@ class USBDiscovery:
                 except Exception:
                     _LOGGER.exception("Error in USB port event callback")
 
-        for usb_device in usb_devices:
+        for usb_device in filtered_usb_devices:
             await self._async_process_discovered_usb_device(usb_device)
 
     @hass_callback
@@ -483,7 +487,7 @@ class USBDiscovery:
         _LOGGER.debug("Executing comports scan")
         async with self._scan_lock:
             await self._async_process_ports(
-                await self.hass.async_add_executor_job(comports)
+                await self.hass.async_add_executor_job(scan_serial_ports)
             )
         if self.initial_scan_done:
             return
@@ -521,9 +525,7 @@ async def websocket_usb_scan(
     msg: dict[str, Any],
 ) -> None:
     """Scan for new usb devices."""
-    usb_discovery: USBDiscovery = hass.data[DOMAIN]
-    if not usb_discovery.observer_active:
-        await usb_discovery.async_request_scan()
+    await async_request_scan(hass)
     connection.send_result(msg["id"])
 
 

homeassistant/components/usb/utils.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
+
+from serial.tools.list_ports import comports
 from serial.tools.list_ports_common import ListPortInfo
 
 from .models import USBDevice
@@ -17,3 +20,12 @@ def usb_device_from_port(port: ListPortInfo) -> USBDevice:
         manufacturer=port.manufacturer,
         description=port.description,
     )
+
+
+def scan_serial_ports() -> Sequence[USBDevice]:
+    """Scan serial ports for USB devices."""
+    return [
+        usb_device_from_port(port)
+        for port in comports()
+        if port.vid is not None or port.pid is not None
+    ]

tests/components/usb/__init__.py

@@ -1,44 +1,29 @@
 """Tests for the USB Discovery integration."""
 
-from homeassistant.components.usb.models import USBDevice
+from unittest.mock import patch
 
-conbee_device = USBDevice(
-    device="/dev/cu.usbmodemDE24338801",
-    vid="1CF1",
-    pid="0030",
-    serial_number="DE2433880",
-    manufacturer="dresden elektronik ingenieurtechnik GmbH",
-    description="ConBee II",
-)
-slae_sh_device = USBDevice(
-    device="/dev/cu.usbserial-110",
-    vid="10C4",
-    pid="EA60",
-    serial_number="00_12_4B_00_22_98_88_7F",
-    manufacturer="Silicon Labs",
-    description="slae.sh cc2652rb stick - slaesh's iot stuff",
-)
-electro_lama_device = USBDevice(
-    device="/dev/cu.usbserial-110",
-    vid="1A86",
-    pid="7523",
-    serial_number=None,
-    manufacturer=None,
-    description="USB2.0-Serial",
-)
-skyconnect_macos_correct = USBDevice(
-    device="/dev/cu.SLAB_USBtoUART",
-    vid="10C4",
-    pid="EA60",
-    serial_number="9ab1da1ea4b3ed11956f4eaca7669f5d",
-    manufacturer="Nabu Casa",
-    description="SkyConnect v1.0",
-)
-skyconnect_macos_incorrect = USBDevice(
-    device="/dev/cu.usbserial-2110",
-    vid="10C4",
-    pid="EA60",
-    serial_number="9ab1da1ea4b3ed11956f4eaca7669f5d",
-    manufacturer="Nabu Casa",
-    description="SkyConnect v1.0",
-)
+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.scan_serial_ports", **kwargs)
+
+
+async def async_request_scan(hass: HomeAssistant) -> None:
+    """Request a USB scan."""
+    return await usb_async_request_scan(hass)