jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate usb to use aiousbwatcher (#136676)

Browse Source 
* Migrate usb to use aiousbwatcher

aiousbwatcher uses inotify on /dev/bus/usb to look for devices
added and being removed which works on a lot more systems

* bump asyncinotify

* bump aiousbwatcher to 1.1.1

* tweaks

* tweaks

* tweaks

* fixes

* debugging

* Update homeassistant/components/usb/__init__.py

* Update homeassistant/components/usb/__init__.py

---------

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
J. Nick Koston 2025-01-28 04:57:11 -10:00 committed by GitHub
parent a05ac6255c
commit 3d7e3590d4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 175 additions and 210 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/keyboard_remote/manifest.json
View File

@ -7,5 +7,5 @@
"iot_class": "local_push",
"loggers": ["aionotify", "evdev"],
"quality_scale": "legacy",
"requirements": ["evdev==1.6.1", "asyncinotify==4.0.2"]
"requirements": ["evdev==1.6.1", "asyncinotify==4.2.0"]
}

124
homeassistant/components/usb/__init__.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine, Sequence
import dataclasses
from datetime import datetime, timedelta
@ -10,8 +11,9 @@ from functools import partial
import logging
import os
import sys
from typing import TYPE_CHECKING, Any, overload
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
@ -26,7 +28,7 @@ from homeassistant.core import (
HomeAssistant,
callback as hass_callback,
)
from homeassistant.helpers import config_validation as cv, discovery_flow, system_info
from homeassistant.helpers import config_validation as cv, discovery_flow
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.deprecation import (
DeprecatedConstant,
@ -43,15 +45,13 @@ from .const import DOMAIN
from .models import USBDevice
from .utils import usb_device_from_port
if TYPE_CHECKING:
from pyudev import Device, MonitorObserver
_LOGGER = logging.getLogger(__name__)
PORT_EVENT_CALLBACK_TYPE = Callable[[set[USBDevice], set[USBDevice]], None]
POLLING_MONITOR_SCAN_PERIOD = timedelta(seconds=5)
REQUEST_SCAN_COOLDOWN = 10 # 10 second cooldown
ADD_REMOVE_SCAN_COOLDOWN = 5 # 5 second cooldown to give devices a chance to register
__all__ = [
"USBCallbackMatcher",
@ -255,15 +255,17 @@ class USBDiscovery:
self.seen: set[tuple[str, ...]] = set()
self.observer_active = False
self._request_debouncer: Debouncer[Coroutine[Any, Any, None]] | None = None
self._add_remove_debouncer: Debouncer[Coroutine[Any, Any, None]] | None = None
self._request_callbacks: list[CALLBACK_TYPE] = []
self.initial_scan_done = False
self._initial_scan_callbacks: list[CALLBACK_TYPE] = []
self._port_event_callbacks: set[PORT_EVENT_CALLBACK_TYPE] = set()
self._last_processed_devices: set[USBDevice] = set()
self._scan_lock = asyncio.Lock()
async def async_setup(self) -> None:
"""Set up USB Discovery."""
if await self._async_supports_monitoring():
if self._async_supports_monitoring():
await self._async_start_monitor()
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STARTED, self.async_start)
@ -279,16 +281,19 @@ class USBDiscovery:
if self._request_debouncer:
self._request_debouncer.async_shutdown()
async def _async_supports_monitoring(self) -> bool:
info = await system_info.async_get_system_info(self.hass)
return not info.get("docker")
@hass_callback
def _async_supports_monitoring(self) -> bool:
return sys.platform == "linux"
async def _async_start_monitor(self) -> None:
"""Start monitoring hardware."""
if not await self._async_start_monitor_udev():
try:
await self._async_start_aiousbwatcher()
except InotifyNotAvailableError as ex:
_LOGGER.info(
"Falling back to periodic filesystem polling for development, libudev "
"is not present"
"Falling back to periodic filesystem polling for development, aiousbwatcher "
"is not available on this system: %s",
ex,
)
self._async_start_monitor_polling()
@ -309,70 +314,27 @@ class USBDiscovery:
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _stop_polling)
async def _async_start_monitor_udev(self) -> bool:
"""Start monitoring hardware with pyudev. Returns True if successful."""
if not sys.platform.startswith("linux"):
return False
async def _async_start_aiousbwatcher(self) -> None:
"""Start monitoring hardware with aiousbwatcher.
if not (
observer := await self.hass.async_add_executor_job(
self._get_monitor_observer
)
):
return False
def _stop_observer(event: Event) -> None:
observer.stop()
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _stop_observer)
self.observer_active = True
return True
def _get_monitor_observer(self) -> MonitorObserver | None:
"""Get the monitor observer.
This runs in the executor because the import
does blocking I/O.
Returns True if successful.
"""
from pyudev import ( # pylint: disable=import-outside-toplevel
Context,
Monitor,
MonitorObserver,
)
try:
context = Context()
except (ImportError, OSError):
return None
@hass_callback
def _usb_change_callback() -> None:
self._async_delayed_add_remove_scan()
monitor = Monitor.from_netlink(context)
try:
monitor.filter_by(subsystem="tty")
except ValueError as ex: # this fails on WSL
_LOGGER.debug(
"Unable to setup pyudev filtering; This is expected on WSL: %s", ex
)
return None
watcher = AIOUSBWatcher()
watcher.async_register_callback(_usb_change_callback)
cancel = watcher.async_start()
observer = MonitorObserver(
monitor, callback=self._device_event, name="usb-observer"
)
@hass_callback
def _async_stop_watcher(event: Event) -> None:
cancel()
observer.start()
return observer
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_stop_watcher)
def _device_event(self, device: Device) -> None:
"""Call when the observer receives a USB device event."""
if device.action not in ("add", "remove"):
return
_LOGGER.info(
"Received a udev device event %r for %s, triggering scan",
device.action,
device.device_node,
)
self.hass.create_task(self._async_scan())
self.observer_active = True
@hass_callback
def async_register_scan_request_callback(
@ -466,11 +428,13 @@ class USBDiscovery:
async def _async_process_ports(self, ports: Sequence[ListPortInfo]) -> 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
# `/dev/cu.SLAB_USBtoUART*`. The former does not work and we should ignore them.
@ -509,11 +473,27 @@ class USBDiscovery:
for usb_device in usb_devices:
await self._async_process_discovered_usb_device(usb_device)
@hass_callback
def _async_delayed_add_remove_scan(self) -> None:
"""Request a serial scan after a debouncer delay."""
if not self._add_remove_debouncer:
self._add_remove_debouncer = Debouncer(
self.hass,
_LOGGER,
cooldown=ADD_REMOVE_SCAN_COOLDOWN,
immediate=False,
function=self._async_scan,
background=True,
)
self._add_remove_debouncer.async_schedule_call()
async def _async_scan_serial(self) -> None:
"""Scan serial ports."""
await self._async_process_ports(
await self.hass.async_add_executor_job(comports)
)
_LOGGER.debug("Executing comports scan")
async with self._scan_lock:
await self._async_process_ports(
await self.hass.async_add_executor_job(comports)
)
if self.initial_scan_done:
return

2
homeassistant/components/usb/manifest.json
View File

@ -7,5 +7,5 @@
"integration_type": "system",
"iot_class": "local_push",
"quality_scale": "internal",
"requirements": ["pyudev==0.24.1", "pyserial==3.5"]
"requirements": ["aiousbwatcher==1.1.1", "pyserial==3.5"]
}

2
homeassistant/package_constraints.txt
View File

@ -8,6 +8,7 @@ aiohttp-asyncmdnsresolver==0.0.1
aiohttp-fast-zlib==0.2.0
aiohttp==3.11.11
aiohttp_cors==0.7.0
aiousbwatcher==1.1.1
aiozoneinfo==0.2.1
astral==2.2
async-interrupt==1.2.0
@ -57,7 +58,6 @@ pyserial==3.5
pyspeex-noise==1.0.2
python-slugify==8.0.4
PyTurboJPEG==1.7.5
pyudev==0.24.1
PyYAML==6.0.2
requests==2.32.3
securetar==2025.1.4

8
requirements_all.txt generated
View File

@ -403,6 +403,9 @@ aiotractive==0.6.0
# homeassistant.components.unifi
aiounifi==81
# homeassistant.components.usb
aiousbwatcher==1.1.1
# homeassistant.components.vlc_telnet
aiovlc==0.5.1
@ -511,7 +514,7 @@ async-upnp-client==0.43.0
asyncarve==0.1.1
# homeassistant.components.keyboard_remote
asyncinotify==4.0.2
asyncinotify==4.2.0
# homeassistant.components.supla
asyncpysupla==0.0.5
@ -2491,9 +2494,6 @@ pytrafikverket==1.1.1
# homeassistant.components.v2c
pytrydan==0.8.0
# homeassistant.components.usb
pyudev==0.24.1
# homeassistant.components.uptimerobot
pyuptimerobot==22.2.0

6
requirements_test_all.txt generated
View File

@ -385,6 +385,9 @@ aiotractive==0.6.0
# homeassistant.components.unifi
aiounifi==81
# homeassistant.components.usb
aiousbwatcher==1.1.1
# homeassistant.components.vlc_telnet
aiovlc==0.5.1
@ -2015,9 +2018,6 @@ pytrafikverket==1.1.1
# homeassistant.components.v2c
pytrydan==0.8.0
# homeassistant.components.usb
pyudev==0.24.1
# homeassistant.components.uptimerobot
pyuptimerobot==22.2.0

241
tests/components/usb/test_init.py
View File

@ -7,6 +7,7 @@ import os
from typing import Any
from unittest.mock import MagicMock, Mock, call, patch, sentinel
from aiousbwatcher import InotifyNotAvailableError
import pytest
from homeassistant.components import usb
@ -15,58 +16,29 @@ from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT
from homeassistant.core import HomeAssistant
from homeassistant.helpers.service_info.usb import UsbServiceInfo
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from . import conbee_device, slae_sh_device
from tests.common import import_and_test_deprecated_constant
from tests.common import async_fire_time_changed, import_and_test_deprecated_constant
from tests.typing import WebSocketGenerator
@pytest.fixture(name="operating_system")
def mock_operating_system():
"""Mock running Home Assistant Operating system."""
@pytest.fixture(name="aiousbwatcher_no_inotify")
def aiousbwatcher_no_inotify():
"""Patch AIOUSBWatcher to not use inotify."""
with patch(
"homeassistant.components.usb.system_info.async_get_system_info",
return_value={
"hassio": True,
"docker": True,
},
"homeassistant.components.usb.AIOUSBWatcher.async_start",
side_effect=InotifyNotAvailableError,
):
yield
@pytest.fixture(name="docker")
def mock_docker():
"""Mock running Home Assistant in docker container."""
with patch(
"homeassistant.components.usb.system_info.async_get_system_info",
return_value={
"hassio": False,
"docker": True,
},
):
yield
@pytest.fixture(name="venv")
def mock_venv():
"""Mock running Home Assistant in a venv container."""
with patch(
"homeassistant.components.usb.system_info.async_get_system_info",
return_value={
"hassio": False,
"docker": False,
"virtualenv": True,
},
):
yield
async def test_observer_discovery(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, venv
async def test_aiousbwatcher_discovery(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that observer can discover a device without raising an exception."""
new_usb = [{"domain": "test1", "vid": "3039"}]
"""Test that aiousbwatcher can discover a device without raising an exception."""
new_usb = [{"domain": "test1", "vid": "3039"}, {"domain": "test2", "vid": "0FA0"}]
mock_comports = [
MagicMock(
@ -78,26 +50,23 @@ async def test_observer_discovery(
description=slae_sh_device.description,
)
]
mock_observer = None
async def _mock_monitor_observer_callback(callback):
await hass.async_add_executor_job(
callback, MagicMock(action="add", device_path="/dev/new")
)
aiousbwatcher_callback = None
def _create_mock_monitor_observer(monitor, callback, name):
nonlocal mock_observer
hass.create_task(_mock_monitor_observer_callback(callback))
mock_observer = MagicMock()
return mock_observer
def async_register_callback(callback):
nonlocal aiousbwatcher_callback
aiousbwatcher_callback = callback
MockAIOUSBWatcher = MagicMock()
MockAIOUSBWatcher.async_register_callback = async_register_callback
with (
patch("sys.platform", "linux"),
patch("pyudev.Context"),
patch("pyudev.MonitorObserver", new=_create_mock_monitor_observer),
patch("pyudev.Monitor.filter_by"),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch(
"homeassistant.components.usb.AIOUSBWatcher", return_value=MockAIOUSBWatcher
),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
):
assert await async_setup_component(hass, "usb", {"usb": {}})
@ -105,18 +74,42 @@ async def test_observer_discovery(
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert len(mock_config_flow.mock_calls) == 1
assert mock_config_flow.mock_calls[0][1][0] == "test1"
assert aiousbwatcher_callback is not None
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
assert len(mock_config_flow.mock_calls) == 1
assert mock_config_flow.mock_calls[0][1][0] == "test1"
await hass.async_block_till_done()
assert len(mock_config_flow.mock_calls) == 1
# pylint:disable-next=unnecessary-dunder-call
assert mock_observer.mock_calls == [call.start(), call.__bool__(), call.stop()]
mock_comports.append(
MagicMock(
device=slae_sh_device.device,
vid=4000,
pid=4000,
serial_number=slae_sh_device.serial_number,
manufacturer=slae_sh_device.manufacturer,
description=slae_sh_device.description,
)
)
aiousbwatcher_callback()
await hass.async_block_till_done()
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=usb.ADD_REMOVE_SCAN_COOLDOWN)
)
await hass.async_block_till_done(wait_background_tasks=True)
assert len(mock_config_flow.mock_calls) == 2
assert mock_config_flow.mock_calls[1][1][0] == "test2"
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_polling_discovery(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, venv
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that polling can discover a device without raising an exception."""
new_usb = [{"domain": "test1", "vid": "3039"}]
@ -143,10 +136,6 @@ async def test_polling_discovery(
with (
patch("sys.platform", "linux"),
patch(
"homeassistant.components.usb.USBDiscovery._get_monitor_observer",
return_value=None,
),
patch(
"homeassistant.components.usb.POLLING_MONITOR_SCAN_PERIOD",
timedelta(seconds=0.01),
@ -174,19 +163,9 @@ async def test_polling_discovery(
await hass.async_block_till_done()
async def test_removal_by_observer_before_started(
hass: HomeAssistant, operating_system
) -> None:
"""Test a device is removed by the observer before started."""
async def _mock_monitor_observer_callback(callback):
await hass.async_add_executor_job(
callback, MagicMock(action="remove", device_path="/dev/new")
)
def _create_mock_monitor_observer(monitor, callback, name):
hass.async_create_task(_mock_monitor_observer_callback(callback))
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_removal_by_aiousbwatcher_before_started(hass: HomeAssistant) -> None:
"""Test a device is removed by the aiousbwatcher before started."""
new_usb = [{"domain": "test1", "vid": "3039", "pid": "3039"}]
mock_comports = [
@ -203,7 +182,6 @@ async def test_removal_by_observer_before_started(
with (
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch("pyudev.MonitorObserver", new=_create_mock_monitor_observer),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
):
assert await async_setup_component(hass, "usb", {"usb": {}})
@ -219,6 +197,7 @@ async def test_removal_by_observer_before_started(
await hass.async_block_till_done()
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -237,7 +216,6 @@ async def test_discovered_by_websocket_scan(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -256,6 +234,7 @@ async def test_discovered_by_websocket_scan(
assert mock_config_flow.mock_calls[0][1][0] == "test1"
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan_limited_by_description_matcher(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -276,7 +255,6 @@ async def test_discovered_by_websocket_scan_limited_by_description_matcher(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -295,6 +273,7 @@ async def test_discovered_by_websocket_scan_limited_by_description_matcher(
assert mock_config_flow.mock_calls[0][1][0] == "test1"
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_most_targeted_matcher_wins(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -316,7 +295,6 @@ async def test_most_targeted_matcher_wins(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -335,6 +313,7 @@ async def test_most_targeted_matcher_wins(
assert mock_config_flow.mock_calls[0][1][0] == "more"
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan_rejected_by_description_matcher(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -355,7 +334,6 @@ async def test_discovered_by_websocket_scan_rejected_by_description_matcher(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -373,6 +351,7 @@ async def test_discovered_by_websocket_scan_rejected_by_description_matcher(
assert len(mock_config_flow.mock_calls) == 0
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan_limited_by_serial_number_matcher(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -398,7 +377,6 @@ async def test_discovered_by_websocket_scan_limited_by_serial_number_matcher(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -417,6 +395,7 @@ async def test_discovered_by_websocket_scan_limited_by_serial_number_matcher(
assert mock_config_flow.mock_calls[0][1][0] == "test1"
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan_rejected_by_serial_number_matcher(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -437,7 +416,6 @@ async def test_discovered_by_websocket_scan_rejected_by_serial_number_matcher(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -455,6 +433,7 @@ async def test_discovered_by_websocket_scan_rejected_by_serial_number_matcher(
assert len(mock_config_flow.mock_calls) == 0
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan_limited_by_manufacturer_matcher(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -480,7 +459,6 @@ async def test_discovered_by_websocket_scan_limited_by_manufacturer_matcher(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -499,6 +477,7 @@ async def test_discovered_by_websocket_scan_limited_by_manufacturer_matcher(
assert mock_config_flow.mock_calls[0][1][0] == "test1"
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan_rejected_by_manufacturer_matcher(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -524,7 +503,6 @@ async def test_discovered_by_websocket_scan_rejected_by_manufacturer_matcher(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -542,6 +520,7 @@ async def test_discovered_by_websocket_scan_rejected_by_manufacturer_matcher(
assert len(mock_config_flow.mock_calls) == 0
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_rejected_with_empty_serial_number_only(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -562,7 +541,6 @@ async def test_discovered_by_websocket_rejected_with_empty_serial_number_only(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -580,6 +558,7 @@ async def test_discovered_by_websocket_rejected_with_empty_serial_number_only(
assert len(mock_config_flow.mock_calls) == 0
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan_match_vid_only(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -598,7 +577,6 @@ async def test_discovered_by_websocket_scan_match_vid_only(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -617,6 +595,7 @@ async def test_discovered_by_websocket_scan_match_vid_only(
assert mock_config_flow.mock_calls[0][1][0] == "test1"
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_scan_match_vid_wrong_pid(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -635,7 +614,6 @@ async def test_discovered_by_websocket_scan_match_vid_wrong_pid(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -653,6 +631,7 @@ async def test_discovered_by_websocket_scan_match_vid_wrong_pid(
assert len(mock_config_flow.mock_calls) == 0
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_discovered_by_websocket_no_vid_pid(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -671,7 +650,6 @@ async def test_discovered_by_websocket_no_vid_pid(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -689,9 +667,9 @@ async def test_discovered_by_websocket_no_vid_pid(
assert len(mock_config_flow.mock_calls) == 0
@pytest.mark.parametrize("exception_type", [ImportError, OSError])
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_non_matching_discovered_by_scanner_after_started(
hass: HomeAssistant, exception_type, hass_ws_client: WebSocketGenerator
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test a websocket scan that does not match."""
new_usb = [{"domain": "test1", "vid": "4444", "pid": "4444"}]
@ -708,7 +686,6 @@ async def test_non_matching_discovered_by_scanner_after_started(
]
with (
patch("pyudev.Context", side_effect=exception_type),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -726,10 +703,10 @@ async def test_non_matching_discovered_by_scanner_after_started(
assert len(mock_config_flow.mock_calls) == 0
async def test_observer_on_wsl_fallback_without_throwing_exception(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, venv
async def test_aiousbwatcher_on_wsl_fallback_without_throwing_exception(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that observer on WSL failure results in fallback to scanning without raising an exception."""
"""Test that aiousbwatcher on WSL failure results in fallback to scanning without raising an exception."""
new_usb = [{"domain": "test1", "vid": "3039"}]
mock_comports = [
@ -744,8 +721,6 @@ async def test_observer_on_wsl_fallback_without_throwing_exception(
]
with (
patch("pyudev.Context"),
patch("pyudev.Monitor.filter_by", side_effect=ValueError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -764,20 +739,8 @@ async def test_observer_on_wsl_fallback_without_throwing_exception(
assert mock_config_flow.mock_calls[0][1][0] == "test1"
async def test_not_discovered_by_observer_before_started_on_docker(
hass: HomeAssistant, docker
) -> None:
"""Test a device is not discovered since observer is not running on bare docker."""
async def _mock_monitor_observer_callback(callback):
await hass.async_add_executor_job(
callback, MagicMock(action="add", device_path="/dev/new")
)
def _create_mock_monitor_observer(monitor, callback, name):
hass.async_create_task(_mock_monitor_observer_callback(callback))
return MagicMock()
async def test_discovered_by_aiousbwatcher_before_started(hass: HomeAssistant) -> None:
"""Test a device is discovered since aiousbwatcher is now running."""
new_usb = [{"domain": "test1", "vid": "3039", "pid": "3039"}]
mock_comports = [
@ -790,23 +753,45 @@ async def test_not_discovered_by_observer_before_started_on_docker(
description=slae_sh_device.description,
)
]
initial_mock_comports = []
aiousbwatcher_callback = None
def async_register_callback(callback):
nonlocal aiousbwatcher_callback
aiousbwatcher_callback = callback
MockAIOUSBWatcher = MagicMock()
MockAIOUSBWatcher.async_register_callback = async_register_callback
with (
patch("sys.platform", "linux"),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch("pyudev.MonitorObserver", new=_create_mock_monitor_observer),
patch(
"homeassistant.components.usb.comports", return_value=initial_mock_comports
),
patch(
"homeassistant.components.usb.AIOUSBWatcher", return_value=MockAIOUSBWatcher
),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
):
assert await async_setup_component(hass, "usb", {"usb": {}})
await hass.async_block_till_done()
with (
patch("homeassistant.components.usb.comports", return_value=[]),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
):
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert len(mock_config_flow.mock_calls) == 0
assert len(mock_config_flow.mock_calls) == 0
initial_mock_comports.extend(mock_comports)
aiousbwatcher_callback()
await hass.async_block_till_done()
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=usb.ADD_REMOVE_SCAN_COOLDOWN)
)
await hass.async_block_till_done(wait_background_tasks=True)
assert len(mock_config_flow.mock_calls) == 1
def test_get_serial_by_id_no_dir() -> None:
@ -889,6 +874,7 @@ def test_human_readable_device_name() -> None:
assert "8A2A" in name
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_async_is_plugged_in(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -912,7 +898,6 @@ async def test_async_is_plugged_in(
}
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=[]),
patch.object(hass.config_entries.flow, "async_init"),
@ -935,6 +920,7 @@ async def test_async_is_plugged_in(
assert usb.async_is_plugged_in(hass, matcher)
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
@pytest.mark.parametrize(
"matcher",
[
@ -953,7 +939,6 @@ async def test_async_is_plugged_in_case_enforcement(
new_usb = [{"domain": "test1", "vid": "ABCD"}]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=[]),
patch.object(hass.config_entries.flow, "async_init"),
@ -967,6 +952,7 @@ async def test_async_is_plugged_in_case_enforcement(
usb.async_is_plugged_in(hass, matcher)
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_web_socket_triggers_discovery_request_callbacks(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -974,7 +960,6 @@ async def test_web_socket_triggers_discovery_request_callbacks(
mock_callback = Mock()
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=[]),
patch("homeassistant.components.usb.comports", return_value=[]),
patch.object(hass.config_entries.flow, "async_init"),
@ -1002,6 +987,7 @@ async def test_web_socket_triggers_discovery_request_callbacks(
assert len(mock_callback.mock_calls) == 1
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_initial_scan_callback(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -1010,7 +996,6 @@ async def test_initial_scan_callback(
mock_callback_2 = Mock()
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=[]),
patch("homeassistant.components.usb.comports", return_value=[]),
patch.object(hass.config_entries.flow, "async_init"),
@ -1038,6 +1023,7 @@ async def test_initial_scan_callback(
cancel_2()
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_cancel_initial_scan_callback(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -1045,7 +1031,6 @@ async def test_cancel_initial_scan_callback(
mock_callback = Mock()
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=[]),
patch("homeassistant.components.usb.comports", return_value=[]),
patch.object(hass.config_entries.flow, "async_init"),
@ -1064,6 +1049,7 @@ async def test_cancel_initial_scan_callback(
assert len(mock_callback.mock_calls) == 0
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
async def test_resolve_serial_by_id(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
@ -1082,7 +1068,6 @@ async def test_resolve_serial_by_id(
]
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=mock_comports),
patch(
@ -1106,6 +1091,7 @@ async def test_resolve_serial_by_id(
assert mock_config_flow.mock_calls[0][2]["data"].device == "/dev/serial/by-id/bla"
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
@pytest.mark.parametrize(
"ports",
[
@ -1190,7 +1176,6 @@ async def test_cp2102n_ordering_on_macos(
with (
patch("sys.platform", "darwin"),
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.async_get_usb", return_value=new_usb),
patch("homeassistant.components.usb.comports", return_value=ports),
patch.object(hass.config_entries.flow, "async_init") as mock_config_flow,
@ -1239,6 +1224,7 @@ def test_deprecated_constants(
)
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
@patch("homeassistant.components.usb.REQUEST_SCAN_COOLDOWN", 0)
async def test_register_port_event_callback(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
@ -1273,7 +1259,6 @@ async def test_register_port_event_callback(
# Start off with no ports
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.comports", return_value=[]),
):
assert await async_setup_component(hass, "usb", {"usb": {}})
@ -1335,6 +1320,7 @@ async def test_register_port_event_callback(
assert mock_callback2.mock_calls == []
@pytest.mark.usefixtures("aiousbwatcher_no_inotify")
@patch("homeassistant.components.usb.REQUEST_SCAN_COOLDOWN", 0)
async def test_register_port_event_callback_failure(
hass: HomeAssistant,
@ -1371,7 +1357,6 @@ async def test_register_port_event_callback_failure(
# Start off with no ports
with (
patch("pyudev.Context", side_effect=ImportError),
patch("homeassistant.components.usb.comports", return_value=[]),
):
assert await async_setup_component(hass, "usb", {"usb": {}})