Fix some homekit_controller pylint warnings and (local only) test failures (#76122)

homeassistant/components/homekit_controller/config_flow.py

@@ -1,14 +1,17 @@
 """Config flow to configure homekit_controller."""
 from __future__ import annotations
 
-from collections.abc import Awaitable
 import logging
 import re
 from typing import TYPE_CHECKING, Any, cast
 
 import aiohomekit
 from aiohomekit import Controller, const as aiohomekit_const
-from aiohomekit.controller.abstract import AbstractDiscovery, AbstractPairing
+from aiohomekit.controller.abstract import (
+    AbstractDiscovery,
+    AbstractPairing,
+    FinishPairing,
+)
 from aiohomekit.exceptions import AuthenticationError
 from aiohomekit.model.categories import Categories
 from aiohomekit.model.status_flags import StatusFlags
@@ -17,7 +20,7 @@ import voluptuous as vol
 
 from homeassistant import config_entries
 from homeassistant.components import zeroconf
-from homeassistant.core import callback
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.data_entry_flow import AbortFlow, FlowResult
 from homeassistant.helpers import device_registry as dr
 
@@ -78,7 +81,9 @@ def formatted_category(category: Categories) -> str:
 
 
 @callback
-def find_existing_host(hass, serial: str) -> config_entries.ConfigEntry | None:
+def find_existing_host(
+    hass: HomeAssistant, serial: str
+) -> config_entries.ConfigEntry | None:
     """Return a set of the configured hosts."""
     for entry in hass.config_entries.async_entries(DOMAIN):
         if entry.data.get("AccessoryPairingID") == serial:
@@ -115,15 +120,17 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         self.category: Categories | None = None
         self.devices: dict[str, AbstractDiscovery] = {}
         self.controller: Controller | None = None
-        self.finish_pairing: Awaitable[AbstractPairing] | None = None
+        self.finish_pairing: FinishPairing | None = None
 
-    async def _async_setup_controller(self):
+    async def _async_setup_controller(self) -> None:
         """Create the controller."""
         self.controller = await async_get_controller(self.hass)
 
-    async def async_step_user(self, user_input=None):
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Handle a flow start."""
-        errors = {}
+        errors: dict[str, str] = {}
 
         if user_input is not None:
             key = user_input["device"]
@@ -142,6 +149,8 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         if self.controller is None:
             await self._async_setup_controller()
 
+        assert self.controller
+
         self.devices = {}
 
         async for discovery in self.controller.async_discover():
@@ -167,7 +176,7 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
             ),
         )
 
-    async def async_step_unignore(self, user_input):
+    async def async_step_unignore(self, user_input: dict[str, Any]) -> FlowResult:
         """Rediscover a previously ignored discover."""
         unique_id = user_input["unique_id"]
         await self.async_set_unique_id(unique_id)
@@ -175,19 +184,21 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         if self.controller is None:
             await self._async_setup_controller()
 
+        assert self.controller
+
         try:
             discovery = await self.controller.async_find(unique_id)
         except aiohomekit.AccessoryNotFoundError:
             return self.async_abort(reason="accessory_not_found_error")
 
         self.name = discovery.description.name
-        self.model = discovery.description.model
+        self.model = getattr(discovery.description, "model", BLE_DEFAULT_NAME)
         self.category = discovery.description.category
         self.hkid = discovery.description.id
 
         return self._async_step_pair_show_form()
 
-    async def _hkid_is_homekit(self, hkid):
+    async def _hkid_is_homekit(self, hkid: str) -> bool:
         """Determine if the device is a homekit bridge or accessory."""
         dev_reg = dr.async_get(self.hass)
         device = dev_reg.async_get_device(
@@ -410,7 +421,9 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
 
         return self._async_step_pair_show_form()
 
-    async def async_step_pair(self, pair_info=None):
+    async def async_step_pair(
+        self, pair_info: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Pair with a new HomeKit accessory."""
         # If async_step_pair is called with no pairing code then we do the M1
         # phase of pairing. If this is successful the device enters pairing
@@ -428,11 +441,16 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         # callable. We call the callable with the pin that the user has typed
         # in.
 
+        # Should never call this step without setting self.hkid
+        assert self.hkid
+
         errors = {}
 
         if self.controller is None:
             await self._async_setup_controller()
 
+        assert self.controller
+
         if pair_info and self.finish_pairing:
             self.context["pairing"] = True
             code = pair_info["pairing_code"]
@@ -507,21 +525,27 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
 
         return self._async_step_pair_show_form(errors)
 
-    async def async_step_busy_error(self, user_input=None):
+    async def async_step_busy_error(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Retry pairing after the accessory is busy."""
         if user_input is not None:
             return await self.async_step_pair()
 
         return self.async_show_form(step_id="busy_error")
 
-    async def async_step_max_tries_error(self, user_input=None):
+    async def async_step_max_tries_error(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Retry pairing after the accessory has reached max tries."""
         if user_input is not None:
             return await self.async_step_pair()
 
         return self.async_show_form(step_id="max_tries_error")
 
-    async def async_step_protocol_error(self, user_input=None):
+    async def async_step_protocol_error(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Retry pairing after the accessory has a protocol error."""
         if user_input is not None:
             return await self.async_step_pair()
@@ -529,7 +553,11 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         return self.async_show_form(step_id="protocol_error")
 
     @callback
-    def _async_step_pair_show_form(self, errors=None):
+    def _async_step_pair_show_form(
+        self, errors: dict[str, str] | None = None
+    ) -> FlowResult:
+        assert self.category
+
         placeholders = self.context["title_placeholders"] = {
             "name": self.name,
             "category": formatted_category(self.category),

homeassistant/components/homekit_controller/light.py

@@ -107,9 +107,9 @@ class HomeKitLight(HomeKitEntity, LightEntity):
         return ColorMode.ONOFF
 
     @property
-    def supported_color_modes(self) -> set[ColorMode | str] | None:
+    def supported_color_modes(self) -> set[ColorMode]:
         """Flag supported color modes."""
-        color_modes: set[ColorMode | str] = set()
+        color_modes: set[ColorMode] = set()
 
         if self.service.has(CharacteristicsTypes.HUE) or self.service.has(
             CharacteristicsTypes.SATURATION

tests/components/homekit_controller/test_sensor.py

@@ -92,10 +92,12 @@ async def test_temperature_sensor_not_added_twice(hass, utcnow):
         hass, create_temperature_sensor_service, suffix="temperature"
     )
 
+    created_sensors = set()
     for state in hass.states.async_all():
-        if state.entity_id.startswith("button"):
-            continue
-        assert state.entity_id == helper.entity_id
+        if state.attributes.get("device_class") == SensorDeviceClass.TEMPERATURE:
+            created_sensors.add(state.entity_id)
+
+    assert created_sensors == {helper.entity_id}
 
 
 async def test_humidity_sensor_read_state(hass, utcnow):