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

@@ -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.*

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

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

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."""