Enforce strict typing for OpenUV (#53409)

* Enforce strict typing for OpenUV

* Linting

* Fix tests
This commit is contained in:
Aaron Bach 2021-07-24 06:50:01 -06:00 committed by GitHub
parent 5c86cc502f
commit 54ace4cdd4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 92 additions and 66 deletions

View file

@ -68,6 +68,7 @@ homeassistant.components.notify.*
homeassistant.components.notion.* homeassistant.components.notion.*
homeassistant.components.number.* homeassistant.components.number.*
homeassistant.components.onewire.* homeassistant.components.onewire.*
homeassistant.components.openuv.*
homeassistant.components.persistent_notification.* homeassistant.components.persistent_notification.*
homeassistant.components.pi_hole.* homeassistant.components.pi_hole.*
homeassistant.components.proximity.* homeassistant.components.proximity.*

View file

@ -1,9 +1,14 @@
"""Support for UV data from openuv.io.""" """Support for UV data from openuv.io."""
from __future__ import annotations
import asyncio import asyncio
from collections.abc import MutableMapping
from typing import Any
from pyopenuv import Client from pyopenuv import Client
from pyopenuv.errors import OpenUvError from pyopenuv.errors import OpenUvError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
ATTR_ATTRIBUTION, ATTR_ATTRIBUTION,
CONF_API_KEY, CONF_API_KEY,
@ -13,7 +18,7 @@ from homeassistant.const import (
CONF_LONGITUDE, CONF_LONGITUDE,
CONF_SENSORS, CONF_SENSORS,
) )
from homeassistant.core import callback from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import aiohttp_client from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.dispatcher import ( from homeassistant.helpers.dispatcher import (
@ -42,14 +47,10 @@ TOPIC_UPDATE = f"{DOMAIN}_data_update"
PLATFORMS = ["binary_sensor", "sensor"] PLATFORMS = ["binary_sensor", "sensor"]
async def async_setup(hass, config): async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up the OpenUV component."""
hass.data[DOMAIN] = {DATA_CLIENT: {}, DATA_LISTENER: {}}
return True
async def async_setup_entry(hass, config_entry):
"""Set up OpenUV as config entry.""" """Set up OpenUV as config entry."""
hass.data.setdefault(DOMAIN, {DATA_CLIENT: {}, DATA_LISTENER: {}})
_verify_domain_control = verify_domain_control(hass, DOMAIN) _verify_domain_control = verify_domain_control(hass, DOMAIN)
try: try:
@ -72,21 +73,21 @@ async def async_setup_entry(hass, config_entry):
hass.config_entries.async_setup_platforms(config_entry, PLATFORMS) hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
@_verify_domain_control @_verify_domain_control
async def update_data(service): async def update_data(_: ServiceCall) -> None:
"""Refresh all OpenUV data.""" """Refresh all OpenUV data."""
LOGGER.debug("Refreshing all OpenUV data") LOGGER.debug("Refreshing all OpenUV data")
await openuv.async_update() await openuv.async_update()
async_dispatcher_send(hass, TOPIC_UPDATE) async_dispatcher_send(hass, TOPIC_UPDATE)
@_verify_domain_control @_verify_domain_control
async def update_uv_index_data(service): async def update_uv_index_data(_: ServiceCall) -> None:
"""Refresh OpenUV UV index data.""" """Refresh OpenUV UV index data."""
LOGGER.debug("Refreshing OpenUV UV index data") LOGGER.debug("Refreshing OpenUV UV index data")
await openuv.async_update_uv_index_data() await openuv.async_update_uv_index_data()
async_dispatcher_send(hass, TOPIC_UPDATE) async_dispatcher_send(hass, TOPIC_UPDATE)
@_verify_domain_control @_verify_domain_control
async def update_protection_data(service): async def update_protection_data(_: ServiceCall) -> None:
"""Refresh OpenUV protection window data.""" """Refresh OpenUV protection window data."""
LOGGER.debug("Refreshing OpenUV protection window data") LOGGER.debug("Refreshing OpenUV protection window data")
await openuv.async_update_protection_data() await openuv.async_update_protection_data()
@ -102,7 +103,7 @@ async def async_setup_entry(hass, config_entry):
return True return True
async def async_unload_entry(hass, config_entry): async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload an OpenUV config entry.""" """Unload an OpenUV config entry."""
unload_ok = await hass.config_entries.async_unload_platforms( unload_ok = await hass.config_entries.async_unload_platforms(
config_entry, PLATFORMS config_entry, PLATFORMS
@ -113,7 +114,7 @@ async def async_unload_entry(hass, config_entry):
return unload_ok return unload_ok
async def async_migrate_entry(hass, config_entry): async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate the config entry upon new versions.""" """Migrate the config entry upon new versions."""
version = config_entry.version version = config_entry.version
data = {**config_entry.data} data = {**config_entry.data}
@ -134,12 +135,12 @@ async def async_migrate_entry(hass, config_entry):
class OpenUV: class OpenUV:
"""Define a generic OpenUV object.""" """Define a generic OpenUV object."""
def __init__(self, client): def __init__(self, client: Client) -> None:
"""Initialize.""" """Initialize."""
self.client = client self.client = client
self.data = {} self.data: dict[str, Any] = {}
async def async_update_protection_data(self): async def async_update_protection_data(self) -> None:
"""Update binary sensor (protection window) data.""" """Update binary sensor (protection window) data."""
try: try:
resp = await self.client.uv_protection_window() resp = await self.client.uv_protection_window()
@ -148,7 +149,7 @@ class OpenUV:
LOGGER.error("Error during protection data update: %s", err) LOGGER.error("Error during protection data update: %s", err)
self.data[DATA_PROTECTION_WINDOW] = {} self.data[DATA_PROTECTION_WINDOW] = {}
async def async_update_uv_index_data(self): async def async_update_uv_index_data(self) -> None:
"""Update sensor (uv index, etc) data.""" """Update sensor (uv index, etc) data."""
try: try:
data = await self.client.uv_index() data = await self.client.uv_index()
@ -157,7 +158,7 @@ class OpenUV:
LOGGER.error("Error during uv index data update: %s", err) LOGGER.error("Error during uv index data update: %s", err)
self.data[DATA_UV] = {} self.data[DATA_UV] = {}
async def async_update(self): async def async_update(self) -> None:
"""Update sensor/binary sensor data.""" """Update sensor/binary sensor data."""
tasks = [self.async_update_protection_data(), self.async_update_uv_index_data()] tasks = [self.async_update_protection_data(), self.async_update_uv_index_data()]
await asyncio.gather(*tasks) await asyncio.gather(*tasks)
@ -166,9 +167,11 @@ class OpenUV:
class OpenUvEntity(Entity): class OpenUvEntity(Entity):
"""Define a generic OpenUV entity.""" """Define a generic OpenUV entity."""
def __init__(self, openuv, sensor_type): def __init__(self, openuv: OpenUV, sensor_type: str) -> None:
"""Initialize.""" """Initialize."""
self._attr_extra_state_attributes = {ATTR_ATTRIBUTION: DEFAULT_ATTRIBUTION} self._attr_extra_state_attributes: MutableMapping[str, Any] = {
ATTR_ATTRIBUTION: DEFAULT_ATTRIBUTION
}
self._attr_should_poll = False self._attr_should_poll = False
self._attr_unique_id = ( self._attr_unique_id = (
f"{openuv.client.latitude}_{openuv.client.longitude}_{sensor_type}" f"{openuv.client.latitude}_{openuv.client.longitude}_{sensor_type}"
@ -176,11 +179,11 @@ class OpenUvEntity(Entity):
self._sensor_type = sensor_type self._sensor_type = sensor_type
self.openuv = openuv self.openuv = openuv
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callbacks.""" """Register callbacks."""
@callback @callback
def update(): def update() -> None:
"""Update the state.""" """Update the state."""
self.update_from_latest_data() self.update_from_latest_data()
self.async_write_ha_state() self.async_write_ha_state()
@ -189,6 +192,6 @@ class OpenUvEntity(Entity):
self.update_from_latest_data() self.update_from_latest_data()
def update_from_latest_data(self): def update_from_latest_data(self) -> None:
"""Update the sensor using the latest data.""" """Update the sensor using the latest data."""
raise NotImplementedError raise NotImplementedError

View file

@ -1,9 +1,11 @@
"""Support for OpenUV binary sensors.""" """Support for OpenUV binary sensors."""
from homeassistant.components.binary_sensor import BinarySensorEntity from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.core import callback from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.dt import as_local, parse_datetime, utcnow from homeassistant.util.dt import as_local, parse_datetime, utcnow
from . import OpenUvEntity from . import OpenUV, OpenUvEntity
from .const import ( from .const import (
DATA_CLIENT, DATA_CLIENT,
DATA_PROTECTION_WINDOW, DATA_PROTECTION_WINDOW,
@ -20,7 +22,9 @@ ATTR_PROTECTION_WINDOW_STARTING_UV = "start_uv"
BINARY_SENSORS = {TYPE_PROTECTION_WINDOW: ("Protection Window", "mdi:sunglasses")} BINARY_SENSORS = {TYPE_PROTECTION_WINDOW: ("Protection Window", "mdi:sunglasses")}
async def async_setup_entry(hass, entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up an OpenUV sensor based on a config entry.""" """Set up an OpenUV sensor based on a config entry."""
openuv = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] openuv = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id]
@ -35,7 +39,7 @@ async def async_setup_entry(hass, entry, async_add_entities):
class OpenUvBinarySensor(OpenUvEntity, BinarySensorEntity): class OpenUvBinarySensor(OpenUvEntity, BinarySensorEntity):
"""Define a binary sensor for OpenUV.""" """Define a binary sensor for OpenUV."""
def __init__(self, openuv, sensor_type, name, icon): def __init__(self, openuv: OpenUV, sensor_type: str, name: str, icon: str) -> None:
"""Initialize the sensor.""" """Initialize the sensor."""
super().__init__(openuv, sensor_type) super().__init__(openuv, sensor_type)
@ -43,7 +47,7 @@ class OpenUvBinarySensor(OpenUvEntity, BinarySensorEntity):
self._attr_name = name self._attr_name = name
@callback @callback
def update_from_latest_data(self): def update_from_latest_data(self) -> None:
"""Update the state.""" """Update the state."""
data = self.openuv.data[DATA_PROTECTION_WINDOW] data = self.openuv.data[DATA_PROTECTION_WINDOW]
@ -59,20 +63,24 @@ class OpenUvBinarySensor(OpenUvEntity, BinarySensorEntity):
return return
if self._sensor_type == TYPE_PROTECTION_WINDOW: if self._sensor_type == TYPE_PROTECTION_WINDOW:
self._attr_is_on = ( from_dt = parse_datetime(data["from_time"])
parse_datetime(data["from_time"]) to_dt = parse_datetime(data["to_time"])
<= utcnow()
<= parse_datetime(data["to_time"]) if not from_dt or not to_dt:
) LOGGER.warning(
"Unable to parse protection window datetimes: %s, %s",
data["from_time"],
data["to_time"],
)
self._attr_is_on = False
return
self._attr_is_on = from_dt <= utcnow() <= to_dt
self._attr_extra_state_attributes.update( self._attr_extra_state_attributes.update(
{ {
ATTR_PROTECTION_WINDOW_ENDING_TIME: as_local( ATTR_PROTECTION_WINDOW_ENDING_TIME: as_local(to_dt),
parse_datetime(data["to_time"])
),
ATTR_PROTECTION_WINDOW_ENDING_UV: data["to_uv"], ATTR_PROTECTION_WINDOW_ENDING_UV: data["to_uv"],
ATTR_PROTECTION_WINDOW_STARTING_UV: data["from_uv"], ATTR_PROTECTION_WINDOW_STARTING_UV: data["from_uv"],
ATTR_PROTECTION_WINDOW_STARTING_TIME: as_local( ATTR_PROTECTION_WINDOW_STARTING_TIME: as_local(from_dt),
parse_datetime(data["from_time"])
),
} }
) )

View file

@ -1,4 +1,8 @@
"""Config flow to configure the OpenUV component.""" """Config flow to configure the OpenUV component."""
from __future__ import annotations
from typing import Any
from pyopenuv import Client from pyopenuv import Client
from pyopenuv.errors import OpenUvError from pyopenuv.errors import OpenUvError
import voluptuous as vol import voluptuous as vol
@ -10,6 +14,7 @@ from homeassistant.const import (
CONF_LATITUDE, CONF_LATITUDE,
CONF_LONGITUDE, CONF_LONGITUDE,
) )
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client, config_validation as cv from homeassistant.helpers import aiohttp_client, config_validation as cv
from .const import DOMAIN from .const import DOMAIN
@ -21,7 +26,7 @@ class OpenUvFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 2 VERSION = 2
@property @property
def config_schema(self): def config_schema(self) -> vol.Schema:
"""Return the config schema.""" """Return the config schema."""
return vol.Schema( return vol.Schema(
{ {
@ -38,7 +43,7 @@ class OpenUvFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
} }
) )
async def _show_form(self, errors=None): async def _show_form(self, errors: dict[str, Any] | None = None) -> FlowResult:
"""Show the form to the user.""" """Show the form to the user."""
return self.async_show_form( return self.async_show_form(
step_id="user", step_id="user",
@ -46,11 +51,13 @@ class OpenUvFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors if errors else {}, errors=errors if errors else {},
) )
async def async_step_import(self, import_config): async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
"""Import a config entry from configuration.yaml.""" """Import a config entry from configuration.yaml."""
return await self.async_step_user(import_config) return await self.async_step_user(import_config)
async def async_step_user(self, user_input=None): async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the start of the config flow.""" """Handle the start of the config flow."""
if not user_input: if not user_input:
return await self._show_form() return await self._show_form()

View file

@ -1,10 +1,14 @@
"""Support for OpenUV sensors.""" """Support for OpenUV sensors."""
from __future__ import annotations
from homeassistant.components.sensor import SensorEntity from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import TIME_MINUTES, UV_INDEX from homeassistant.const import TIME_MINUTES, UV_INDEX
from homeassistant.core import callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.dt import as_local, parse_datetime from homeassistant.util.dt import as_local, parse_datetime
from . import OpenUvEntity from . import OpenUV, OpenUvEntity
from .const import ( from .const import (
DATA_CLIENT, DATA_CLIENT,
DATA_UV, DATA_UV,
@ -76,7 +80,9 @@ SENSORS = {
} }
async def async_setup_entry(hass, entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a OpenUV sensor based on a config entry.""" """Set up a OpenUV sensor based on a config entry."""
openuv = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] openuv = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id]
@ -91,7 +97,9 @@ async def async_setup_entry(hass, entry, async_add_entities):
class OpenUvSensor(OpenUvEntity, SensorEntity): class OpenUvSensor(OpenUvEntity, SensorEntity):
"""Define a binary sensor for OpenUV.""" """Define a binary sensor for OpenUV."""
def __init__(self, openuv, sensor_type, name, icon, unit): def __init__(
self, openuv: OpenUV, sensor_type: str, name: str, icon: str, unit: str | None
) -> None:
"""Initialize the sensor.""" """Initialize the sensor."""
super().__init__(openuv, sensor_type) super().__init__(openuv, sensor_type)
@ -100,7 +108,7 @@ class OpenUvSensor(OpenUvEntity, SensorEntity):
self._attr_unit_of_measurement = unit self._attr_unit_of_measurement = unit
@callback @callback
def update_from_latest_data(self): def update_from_latest_data(self) -> None:
"""Update the state.""" """Update the state."""
data = self.openuv.data[DATA_UV].get("result") data = self.openuv.data[DATA_UV].get("result")
@ -127,9 +135,11 @@ class OpenUvSensor(OpenUvEntity, SensorEntity):
self._attr_state = UV_LEVEL_LOW self._attr_state = UV_LEVEL_LOW
elif self._sensor_type == TYPE_MAX_UV_INDEX: elif self._sensor_type == TYPE_MAX_UV_INDEX:
self._attr_state = data["uv_max"] self._attr_state = data["uv_max"]
self._attr_extra_state_attributes.update( uv_max_time = parse_datetime(data["uv_max_time"])
{ATTR_MAX_UV_TIME: as_local(parse_datetime(data["uv_max_time"]))} if uv_max_time:
) self._attr_extra_state_attributes.update(
{ATTR_MAX_UV_TIME: as_local(uv_max_time)}
)
elif self._sensor_type in ( elif self._sensor_type in (
TYPE_SAFE_EXPOSURE_TIME_1, TYPE_SAFE_EXPOSURE_TIME_1,
TYPE_SAFE_EXPOSURE_TIME_2, TYPE_SAFE_EXPOSURE_TIME_2,

View file

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

View file

@ -2,7 +2,6 @@
from unittest.mock import patch from unittest.mock import patch
from pyopenuv.errors import InvalidApiKeyError from pyopenuv.errors import InvalidApiKeyError
import pytest
from homeassistant import data_entry_flow from homeassistant import data_entry_flow
from homeassistant.components.openuv import DOMAIN from homeassistant.components.openuv import DOMAIN
@ -17,19 +16,6 @@ from homeassistant.const import (
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
@pytest.fixture(autouse=True)
def mock_setup():
"""Prevent setup."""
with patch(
"homeassistant.components.openuv.async_setup",
return_value=True,
), patch(
"homeassistant.components.openuv.async_setup_entry",
return_value=True,
):
yield
async def test_duplicate_error(hass): async def test_duplicate_error(hass):
"""Test that errors are shown when duplicates are added.""" """Test that errors are shown when duplicates are added."""
conf = { conf = {
@ -81,7 +67,7 @@ async def test_step_user(hass):
} }
with patch( with patch(
"homeassistant.components.airvisual.async_setup_entry", return_value=True "homeassistant.components.openuv.async_setup_entry", return_value=True
), patch("pyopenuv.client.Client.uv_index"): ), patch("pyopenuv.client.Client.uv_index"):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER} DOMAIN, context={"source": SOURCE_USER}