From ad87d06d1f974e2f91ef27380d5b67e63de9b9ba Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 2 Mar 2022 20:59:31 -1000 Subject: [PATCH] Enable strict typing for usb (#67466) * Enable strict typing for usb * Enable strict typing for usb * Enable strict typing for usb * adjust * coverage * Update tests/components/usb/test_init.py --- .strict-typing | 1 + homeassistant/components/usb/__init__.py | 15 ++++-- mypy.ini | 11 +++++ tests/components/usb/test_init.py | 62 ++++++++++++++++++++++-- 4 files changed, 81 insertions(+), 8 deletions(-) diff --git a/.strict-typing b/.strict-typing index 652d5fe29e1..73975e2d1f7 100644 --- a/.strict-typing +++ b/.strict-typing @@ -206,6 +206,7 @@ homeassistant.components.unifiprotect.* homeassistant.components.upcloud.* homeassistant.components.uptime.* homeassistant.components.uptimerobot.* +homeassistant.components.usb.* homeassistant.components.vacuum.* homeassistant.components.vallox.* homeassistant.components.velbus.* diff --git a/homeassistant/components/usb/__init__.py b/homeassistant/components/usb/__init__.py index 38377aadd9f..5ac390ad168 100644 --- a/homeassistant/components/usb/__init__.py +++ b/homeassistant/components/usb/__init__.py @@ -6,7 +6,7 @@ import fnmatch import logging import os import sys -from typing import Any +from typing import TYPE_CHECKING, Any from serial.tools.list_ports import comports from serial.tools.list_ports_common import ListPortInfo @@ -28,6 +28,9 @@ from .const import DOMAIN from .models import USBDevice from .utils import usb_device_from_port +if TYPE_CHECKING: + from pyudev import Device + _LOGGER = logging.getLogger(__name__) REQUEST_SCAN_COOLDOWN = 60 # 1 minute cooldown @@ -163,12 +166,14 @@ class USBDiscovery: monitor, callback=self._device_discovered, name="usb-observer" ) observer.start() - self.hass.bus.async_listen_once( - EVENT_HOMEASSISTANT_STOP, lambda event: observer.stop() - ) + + def _stop_observer(event: Event) -> None: + observer.stop() + + self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _stop_observer) self.observer_active = True - def _device_discovered(self, device): + def _device_discovered(self, device: Device) -> None: """Call when the observer discovers a new usb tty device.""" if device.action != "add": return diff --git a/mypy.ini b/mypy.ini index 2d4005afae6..e35fb90f0ea 100644 --- a/mypy.ini +++ b/mypy.ini @@ -2067,6 +2067,17 @@ no_implicit_optional = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.usb.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +no_implicit_optional = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.vacuum.*] check_untyped_defs = true disallow_incomplete_defs = true diff --git a/tests/components/usb/test_init.py b/tests/components/usb/test_init.py index ce86965f093..f1cf67bfb78 100644 --- a/tests/components/usb/test_init.py +++ b/tests/components/usb/test_init.py @@ -1,12 +1,12 @@ """Tests for the USB Discovery integration.""" import os import sys -from unittest.mock import MagicMock, patch, sentinel +from unittest.mock import MagicMock, call, patch, sentinel import pytest from homeassistant.components import usb -from homeassistant.const import EVENT_HOMEASSISTANT_STARTED +from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP from homeassistant.setup import async_setup_component from . import conbee_device, slae_sh_device @@ -52,6 +52,60 @@ def mock_venv(): yield +@pytest.mark.skipif( + not sys.platform.startswith("linux"), + reason="Only works on linux", +) +async def test_observer_discovery(hass, hass_ws_client, venv): + """Test that observer can discover a device without raising an exception.""" + new_usb = [{"domain": "test1", "vid": "3039"}] + + mock_comports = [ + MagicMock( + device=slae_sh_device.device, + vid=12345, + pid=12345, + serial_number=slae_sh_device.serial_number, + manufacturer=slae_sh_device.manufacturer, + description=slae_sh_device.description, + ) + ] + mock_observer = None + + async def _mock_monitor_observer_callback(callback): + await hass.async_add_executor_job( + callback, MagicMock(action="create", device_path="/dev/new") + ) + + def _create_mock_monitor_observer(monitor, callback, name): + nonlocal mock_observer + hass.async_create_task(_mock_monitor_observer_callback(callback)) + mock_observer = MagicMock() + return mock_observer + + with 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.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() + 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" + + hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP) + await hass.async_block_till_done() + + assert mock_observer.mock_calls == [call.start(), call.stop()] + + @pytest.mark.skipif( not sys.platform.startswith("linux"), reason="Only works on linux", @@ -66,7 +120,6 @@ async def test_removal_by_observer_before_started(hass, operating_system): def _create_mock_monitor_observer(monitor, callback, name): hass.async_create_task(_mock_monitor_observer_callback(callback)) - return MagicMock() new_usb = [{"domain": "test1", "vid": "3039", "pid": "3039"}] @@ -99,6 +152,9 @@ async def test_removal_by_observer_before_started(hass, operating_system): assert len(mock_config_flow.mock_calls) == 0 + hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP) + await hass.async_block_till_done() + async def test_discovered_by_websocket_scan(hass, hass_ws_client): """Test a device is discovered from websocket scan."""