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

* Add some helper methods to simplify USB integration testing

* Re-export `usb_device_from_port`
This commit is contained in:
puddly 2025-03-29 17:26:37 -04:00 committed by Franck Nijhof
parent eed075dbfa
commit 8ac8401b4e
No known key found for this signature in database
GPG Key ID: D62583BA8AB11CA3
4 changed files with 266 additions and 268 deletions

View File

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

View File

@ -2,6 +2,9 @@
from __future__ import annotations 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 serial.tools.list_ports_common import ListPortInfo
from .models import USBDevice from .models import USBDevice
@ -17,3 +20,12 @@ def usb_device_from_port(port: ListPortInfo) -> USBDevice:
manufacturer=port.manufacturer, manufacturer=port.manufacturer,
description=port.description, 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
]

View File

@ -1,44 +1,29 @@
"""Tests for the USB Discovery integration.""" """Tests for the USB Discovery integration."""
from homeassistant.components.usb.models import USBDevice from unittest.mock import patch
conbee_device = USBDevice( from aiousbwatcher import InotifyNotAvailableError
device="/dev/cu.usbmodemDE24338801", import pytest
vid="1CF1",
pid="0030", from homeassistant.components.usb import async_request_scan as usb_async_request_scan
serial_number="DE2433880", from homeassistant.core import HomeAssistant
manufacturer="dresden elektronik ingenieurtechnik GmbH",
description="ConBee II",
) @pytest.fixture(name="force_usb_polling_watcher")
slae_sh_device = USBDevice( def force_usb_polling_watcher():
device="/dev/cu.usbserial-110", """Patch the USB integration to not use inotify and fall back to polling."""
vid="10C4", with patch(
pid="EA60", "homeassistant.components.usb.AIOUSBWatcher.async_start",
serial_number="00_12_4B_00_22_98_88_7F", side_effect=InotifyNotAvailableError,
manufacturer="Silicon Labs", ):
description="slae.sh cc2652rb stick - slaesh's iot stuff", yield
)
electro_lama_device = USBDevice(
device="/dev/cu.usbserial-110", def patch_scanned_serial_ports(**kwargs) -> None:
vid="1A86", """Patch the USB integration's list of scanned serial ports."""
pid="7523", return patch("homeassistant.components.usb.scan_serial_ports", **kwargs)
serial_number=None,
manufacturer=None,
description="USB2.0-Serial", async def async_request_scan(hass: HomeAssistant) -> None:
) """Request a USB scan."""
skyconnect_macos_correct = USBDevice( return await usb_async_request_scan(hass)
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",
)

File diff suppressed because it is too large Load Diff