Enable AsusWRT strict typing (#70396)

* Enable AsusWRT strict typing

* Fix myPi errors
This commit is contained in:
ollo69 2022-04-22 09:14:13 +02:00 committed by GitHub
parent d35fa28721
commit 74b50a8009
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 55 additions and 33 deletions

View File

@ -46,6 +46,7 @@ homeassistant.components.ambient_station.*
homeassistant.components.amcrest.* homeassistant.components.amcrest.*
homeassistant.components.ampio.* homeassistant.components.ampio.*
homeassistant.components.aseko_pool_live.* homeassistant.components.aseko_pool_live.*
homeassistant.components.asuswrt.*
homeassistant.components.automation.* homeassistant.components.automation.*
homeassistant.components.backup.* homeassistant.components.backup.*
homeassistant.components.binary_sensor.* homeassistant.components.binary_sensor.*

View File

@ -2,7 +2,7 @@
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import HomeAssistant from homeassistant.core import Event, HomeAssistant
from .const import DATA_ASUSWRT, DOMAIN from .const import DATA_ASUSWRT, DOMAIN
from .router import AsusWrtRouter from .router import AsusWrtRouter
@ -18,7 +18,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
router.async_on_close(entry.add_update_listener(update_listener)) router.async_on_close(entry.add_update_listener(update_listener))
async def async_close_connection(event): async def async_close_connection(event: Event) -> None:
"""Close AsusWrt connection on HA Stop.""" """Close AsusWrt connection on HA Stop."""
await router.close() await router.close()

View File

@ -1,15 +1,19 @@
"""Config flow to configure the AsusWrt integration.""" """Config flow to configure the AsusWrt integration."""
from __future__ import annotations
import logging import logging
import os import os
import socket import socket
from typing import Any
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.device_tracker.const import ( from homeassistant.components.device_tracker.const import (
CONF_CONSIDER_HOME, CONF_CONSIDER_HOME,
DEFAULT_CONSIDER_HOME, DEFAULT_CONSIDER_HOME,
) )
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import ( from homeassistant.const import (
CONF_HOST, CONF_HOST,
CONF_MODE, CONF_MODE,
@ -19,6 +23,7 @@ from homeassistant.const import (
CONF_USERNAME, CONF_USERNAME,
) )
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from .const import ( from .const import (
@ -45,13 +50,13 @@ RESULT_SUCCESS = "success"
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
def _is_file(value) -> bool: def _is_file(value: str) -> bool:
"""Validate that the value is an existing file.""" """Validate that the value is an existing file."""
file_in = os.path.expanduser(str(value)) file_in = os.path.expanduser(value)
return os.path.isfile(file_in) and os.access(file_in, os.R_OK) return os.path.isfile(file_in) and os.access(file_in, os.R_OK)
def _get_ip(host): def _get_ip(host: str) -> str | None:
"""Get the ip address from the host name.""" """Get the ip address from the host name."""
try: try:
return socket.gethostbyname(host) return socket.gethostbyname(host)
@ -59,17 +64,17 @@ def _get_ip(host):
return None return None
class AsusWrtFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): class AsusWrtFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow.""" """Handle a config flow."""
VERSION = 1 VERSION = 1
def __init__(self):
"""Initialize AsusWrt config flow."""
self._host = None
@callback @callback
def _show_setup_form(self, user_input=None, errors=None): def _show_setup_form(
self,
user_input: dict[str, Any] | None = None,
errors: dict[str, str] | None = None,
) -> FlowResult:
"""Show the setup form to the user.""" """Show the setup form to the user."""
if user_input is None: if user_input is None:
@ -100,25 +105,27 @@ class AsusWrtFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors or {}, errors=errors or {},
) )
async def _async_check_connection(self, user_input): @staticmethod
async def _async_check_connection(user_input: dict[str, Any]) -> str:
"""Attempt to connect the AsusWrt router.""" """Attempt to connect the AsusWrt router."""
host: str = user_input[CONF_HOST]
api = get_api(user_input) api = get_api(user_input)
try: try:
await api.connection.async_connect() await api.connection.async_connect()
except OSError: except OSError:
_LOGGER.error("Error connecting to the AsusWrt router at %s", self._host) _LOGGER.error("Error connecting to the AsusWrt router at %s", host)
return RESULT_CONN_ERROR return RESULT_CONN_ERROR
except Exception: # pylint: disable=broad-except except Exception: # pylint: disable=broad-except
_LOGGER.exception( _LOGGER.exception(
"Unknown error connecting with AsusWrt router at %s", self._host "Unknown error connecting with AsusWrt router at %s", host
) )
return RESULT_UNKNOWN return RESULT_UNKNOWN
if not api.is_connected: if not api.is_connected:
_LOGGER.error("Error connecting to the AsusWrt router at %s", self._host) _LOGGER.error("Error connecting to the AsusWrt router at %s", host)
return RESULT_CONN_ERROR return RESULT_CONN_ERROR
conf_protocol = user_input[CONF_PROTOCOL] conf_protocol = user_input[CONF_PROTOCOL]
@ -126,7 +133,9 @@ class AsusWrtFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
api.connection.disconnect() api.connection.disconnect()
return RESULT_SUCCESS return RESULT_SUCCESS
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 initiated by the user.""" """Handle a flow initiated by the user."""
if self._async_current_entries(): if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed") return self.async_abort(reason="single_instance_allowed")
@ -134,11 +143,11 @@ class AsusWrtFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is None: if user_input is None:
return self._show_setup_form(user_input) return self._show_setup_form(user_input)
errors = {} errors: dict[str, str] = {}
self._host = user_input[CONF_HOST] host: str = user_input[CONF_HOST]
pwd = user_input.get(CONF_PASSWORD) pwd: str | None = user_input.get(CONF_PASSWORD)
ssh = user_input.get(CONF_SSH_KEY) ssh: str | None = user_input.get(CONF_SSH_KEY)
if not (pwd or ssh): if not (pwd or ssh):
errors["base"] = "pwd_or_ssh" errors["base"] = "pwd_or_ssh"
@ -151,7 +160,7 @@ class AsusWrtFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors["base"] = "ssh_not_file" errors["base"] = "ssh_not_file"
if not errors: if not errors:
ip_address = await self.hass.async_add_executor_job(_get_ip, self._host) ip_address = await self.hass.async_add_executor_job(_get_ip, host)
if not ip_address: if not ip_address:
errors["base"] = "invalid_host" errors["base"] = "invalid_host"
@ -164,25 +173,27 @@ class AsusWrtFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self._show_setup_form(user_input, errors) return self._show_setup_form(user_input, errors)
return self.async_create_entry( return self.async_create_entry(
title=self._host, title=host,
data=user_input, data=user_input,
) )
@staticmethod @staticmethod
@callback @callback
def async_get_options_flow(config_entry): def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
"""Get the options flow for this handler.""" """Get the options flow for this handler."""
return OptionsFlowHandler(config_entry) return OptionsFlowHandler(config_entry)
class OptionsFlowHandler(config_entries.OptionsFlow): class OptionsFlowHandler(OptionsFlow):
"""Handle a option flow for AsusWrt.""" """Handle a option flow for AsusWrt."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None: def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize options flow.""" """Initialize options flow."""
self.config_entry = config_entry self.config_entry = config_entry
async def async_step_init(self, user_input=None): async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle options flow.""" """Handle options flow."""
if user_input is not None: if user_input is not None:
return self.async_create_entry(title="", data=user_input) return self.async_create_entry(title="", data=user_input)

View File

@ -22,7 +22,7 @@ async def async_setup_entry(
tracked: set = set() tracked: set = set()
@callback @callback
def update_router(): def update_router() -> None:
"""Update the values of the router.""" """Update the values of the router."""
add_entities(router, async_add_entities, tracked) add_entities(router, async_add_entities, tracked)

View File

@ -119,7 +119,7 @@ class AsusWrtSensorDataHandler:
async def _get_temperatures(self) -> dict[str, Any]: async def _get_temperatures(self) -> dict[str, Any]:
"""Fetch temperatures information from the router.""" """Fetch temperatures information from the router."""
try: try:
temperatures = await self._api.async_get_temperature() temperatures: dict[str, Any] = await self._api.async_get_temperature()
except (OSError, ValueError) as exc: except (OSError, ValueError) as exc:
raise UpdateFailed(exc) from exc raise UpdateFailed(exc) from exc

View File

@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from numbers import Real
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
SensorDeviceClass, SensorDeviceClass,
@ -197,10 +196,10 @@ class AsusWrtSensor(CoordinatorEntity, SensorEntity):
self._attr_extra_state_attributes = {"hostname": router.host} self._attr_extra_state_attributes = {"hostname": router.host}
@property @property
def native_value(self) -> float | str | None: def native_value(self) -> float | int | str | None:
"""Return current state.""" """Return current state."""
descr = self.entity_description descr = self.entity_description
state = self.coordinator.data.get(descr.key) state: float | int | str | None = self.coordinator.data.get(descr.key)
if state is not None and descr.factor and isinstance(state, Real): if state is not None and descr.factor and isinstance(state, (float, int)):
return round(state / descr.factor, descr.precision) return round(state / descr.factor, descr.precision)
return state return state

View File

@ -308,6 +308,17 @@ no_implicit_optional = true
warn_return_any = true warn_return_any = true
warn_unreachable = true warn_unreachable = true
[mypy-homeassistant.components.asuswrt.*]
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.automation.*] [mypy-homeassistant.components.automation.*]
check_untyped_defs = true check_untyped_defs = true
disallow_incomplete_defs = true disallow_incomplete_defs = true