From d5695a2d8656d2f9cb4d549c80cad331c914af1f Mon Sep 17 00:00:00 2001 From: Jc2k Date: Thu, 4 Aug 2022 11:30:37 +0100 Subject: [PATCH] Fix some homekit_controller pylint warnings and (local only) test failures (#76122) --- .../homekit_controller/config_flow.py | 60 ++++++++++++++----- .../components/homekit_controller/light.py | 4 +- .../homekit_controller/test_sensor.py | 8 ++- 3 files changed, 51 insertions(+), 21 deletions(-) diff --git a/homeassistant/components/homekit_controller/config_flow.py b/homeassistant/components/homekit_controller/config_flow.py index 31677e37b20..d5ce8c37e7c 100644 --- a/homeassistant/components/homekit_controller/config_flow.py +++ b/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), diff --git a/homeassistant/components/homekit_controller/light.py b/homeassistant/components/homekit_controller/light.py index df691ac3f6f..d882f6790f7 100644 --- a/homeassistant/components/homekit_controller/light.py +++ b/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 diff --git a/tests/components/homekit_controller/test_sensor.py b/tests/components/homekit_controller/test_sensor.py index c2a466d3997..79ae29c7434 100644 --- a/tests/components/homekit_controller/test_sensor.py +++ b/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):