Enable basic type checking in upnp (#66253)

Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2022-03-08 07:51:23 +01:00 committed by GitHub
parent 1793c29fac
commit 8260767e8f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 29 additions and 41 deletions

View File

@ -121,6 +121,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
cancel_discovered_callback() cancel_discovered_callback()
# Create device. # Create device.
assert discovery_info is not None
assert discovery_info.ssdp_location is not None
location = discovery_info.ssdp_location location = discovery_info.ssdp_location
try: try:
device = await Device.async_create_device(hass, location) device = await Device.async_create_device(hass, location)

View File

@ -43,6 +43,7 @@ async def async_setup_entry(
class UpnpStatusBinarySensor(UpnpEntity, BinarySensorEntity): class UpnpStatusBinarySensor(UpnpEntity, BinarySensorEntity):
"""Class for UPnP/IGD binary sensors.""" """Class for UPnP/IGD binary sensors."""
entity_description: UpnpBinarySensorEntityDescription
_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
def __init__( def __init__(

View File

@ -4,7 +4,7 @@ from __future__ import annotations
import asyncio import asyncio
from collections.abc import Mapping from collections.abc import Mapping
from datetime import timedelta from datetime import timedelta
from typing import Any from typing import Any, cast
import voluptuous as vol import voluptuous as vol
@ -31,16 +31,17 @@ from .const import (
def _friendly_name_from_discovery(discovery_info: ssdp.SsdpServiceInfo) -> str: def _friendly_name_from_discovery(discovery_info: ssdp.SsdpServiceInfo) -> str:
"""Extract user-friendly name from discovery.""" """Extract user-friendly name from discovery."""
return ( return cast(
str,
discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME) discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
or discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME) or discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME)
or discovery_info.ssdp_headers.get("_host", "") or discovery_info.ssdp_headers.get("_host", ""),
) )
def _is_complete_discovery(discovery_info: ssdp.SsdpServiceInfo) -> bool: def _is_complete_discovery(discovery_info: ssdp.SsdpServiceInfo) -> bool:
"""Test if discovery is complete and usable.""" """Test if discovery is complete and usable."""
return ( return bool(
ssdp.ATTR_UPNP_UDN in discovery_info.upnp ssdp.ATTR_UPNP_UDN in discovery_info.upnp
and discovery_info.ssdp_st and discovery_info.ssdp_st
and discovery_info.ssdp_location and discovery_info.ssdp_location
@ -114,14 +115,13 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize the UPnP/IGD config flow.""" """Initialize the UPnP/IGD config flow."""
self._discoveries: list[SsdpServiceInfo] | None = None self._discoveries: list[SsdpServiceInfo] | None = None
async def async_step_user( async def async_step_user(self, user_input: Mapping | None = None) -> FlowResult:
self, user_input: Mapping | None = None
) -> Mapping[str, Any]:
"""Handle a flow start.""" """Handle a flow start."""
LOGGER.debug("async_step_user: user_input: %s", user_input) LOGGER.debug("async_step_user: user_input: %s", user_input)
if user_input is not None: if user_input is not None:
# Ensure wanted device was discovered. # Ensure wanted device was discovered.
assert self._discoveries
matching_discoveries = [ matching_discoveries = [
discovery discovery
for discovery in self._discoveries for discovery in self._discoveries
@ -248,12 +248,13 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_ssdp_confirm( async def async_step_ssdp_confirm(
self, user_input: Mapping | None = None self, user_input: Mapping | None = None
) -> Mapping[str, Any]: ) -> FlowResult:
"""Confirm integration via SSDP.""" """Confirm integration via SSDP."""
LOGGER.debug("async_step_ssdp_confirm: user_input: %s", user_input) LOGGER.debug("async_step_ssdp_confirm: user_input: %s", user_input)
if user_input is None: if user_input is None:
return self.async_show_form(step_id="ssdp_confirm") return self.async_show_form(step_id="ssdp_confirm")
assert self._discoveries
discovery = self._discoveries[0] discovery = self._discoveries[0]
return await self._async_create_entry_from_discovery(discovery) return await self._async_create_entry_from_discovery(discovery)
@ -268,7 +269,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_create_entry_from_discovery( async def _async_create_entry_from_discovery(
self, self,
discovery: SsdpServiceInfo, discovery: SsdpServiceInfo,
) -> Mapping[str, Any]: ) -> FlowResult:
"""Create an entry from discovery.""" """Create an entry from discovery."""
LOGGER.debug( LOGGER.debug(
"_async_create_entry_from_discovery: discovery: %s", "_async_create_entry_from_discovery: discovery: %s",
@ -291,7 +292,7 @@ class UpnpOptionsFlowHandler(config_entries.OptionsFlow):
"""Initialize.""" """Initialize."""
self.config_entry = config_entry self.config_entry = config_entry
async def async_step_init(self, user_input: Mapping = None) -> None: async def async_step_init(self, user_input: Mapping = None) -> FlowResult:
"""Manage the options.""" """Manage the options."""
if user_input is not None: if user_input is not None:
coordinator = self.hass.data[DOMAIN][self.config_entry.entry_id] coordinator = self.hass.data[DOMAIN][self.config_entry.entry_id]

View File

@ -9,7 +9,7 @@ from urllib.parse import urlparse
from async_upnp_client import UpnpDevice, UpnpFactory from async_upnp_client import UpnpDevice, UpnpFactory
from async_upnp_client.aiohttp import AiohttpSessionRequester from async_upnp_client.aiohttp import AiohttpSessionRequester
from async_upnp_client.exceptions import UpnpError from async_upnp_client.exceptions import UpnpError
from async_upnp_client.profiles.igd import IgdDevice from async_upnp_client.profiles.igd import IgdDevice, StatusInfo
from homeassistant.components import ssdp from homeassistant.components import ssdp
from homeassistant.components.ssdp import SsdpChange, SsdpServiceInfo from homeassistant.components.ssdp import SsdpChange, SsdpServiceInfo
@ -49,7 +49,7 @@ class Device:
"""Initialize UPnP/IGD device.""" """Initialize UPnP/IGD device."""
self.hass = hass self.hass = hass
self._igd_device = igd_device self._igd_device = igd_device
self.coordinator: DataUpdateCoordinator = None self.coordinator: DataUpdateCoordinator | None = None
@classmethod @classmethod
async def async_create_device( async def async_create_device(
@ -129,7 +129,7 @@ class Device:
return self.usn return self.usn
@property @property
def hostname(self) -> str: def hostname(self) -> str | None:
"""Get the hostname.""" """Get the hostname."""
url = self._igd_device.device.device_url url = self._igd_device.device.device_url
parsed = urlparse(url) parsed = urlparse(url)
@ -177,7 +177,9 @@ class Device:
self._igd_device.async_get_external_ip_address(), self._igd_device.async_get_external_ip_address(),
return_exceptions=True, return_exceptions=True,
) )
result = [] status_info: StatusInfo | None = None
ip_address: str | None = None
for idx, value in enumerate(values): for idx, value in enumerate(values):
if isinstance(value, UpnpError): if isinstance(value, UpnpError):
# Not all routers support some of these items although based # Not all routers support some of these items although based
@ -188,16 +190,18 @@ class Device:
self, self,
str(value), str(value),
) )
result.append(None)
continue continue
if isinstance(value, Exception): if isinstance(value, Exception):
raise value raise value
result.append(value) if isinstance(value, StatusInfo):
status_info = value
elif isinstance(value, str):
ip_address = value
return { return {
WAN_STATUS: result[0][0] if result[0] is not None else None, WAN_STATUS: status_info[0] if status_info is not None else None,
ROUTER_UPTIME: result[0][2] if result[0] is not None else None, ROUTER_UPTIME: status_info[2] if status_info is not None else None,
ROUTER_IP: result[1], ROUTER_IP: ip_address,
} }

View File

@ -143,6 +143,8 @@ async def async_setup_entry(
class UpnpSensor(UpnpEntity, SensorEntity): class UpnpSensor(UpnpEntity, SensorEntity):
"""Base class for UPnP/IGD sensors.""" """Base class for UPnP/IGD sensors."""
entity_description: UpnpSensorEntityDescription
class RawUpnpSensor(UpnpSensor): class RawUpnpSensor(UpnpSensor):
"""Representation of a UPnP/IGD sensor.""" """Representation of a UPnP/IGD sensor."""
@ -159,8 +161,6 @@ class RawUpnpSensor(UpnpSensor):
class DerivedUpnpSensor(UpnpSensor): class DerivedUpnpSensor(UpnpSensor):
"""Representation of a UNIT Sent/Received per second sensor.""" """Representation of a UNIT Sent/Received per second sensor."""
entity_description: UpnpSensorEntityDescription
def __init__( def __init__(
self, self,
coordinator: UpnpDataUpdateCoordinator, coordinator: UpnpDataUpdateCoordinator,

View File

@ -2725,21 +2725,6 @@ ignore_errors = true
[mypy-homeassistant.components.unifi.unifi_entity_base] [mypy-homeassistant.components.unifi.unifi_entity_base]
ignore_errors = true ignore_errors = true
[mypy-homeassistant.components.upnp]
ignore_errors = true
[mypy-homeassistant.components.upnp.binary_sensor]
ignore_errors = true
[mypy-homeassistant.components.upnp.config_flow]
ignore_errors = true
[mypy-homeassistant.components.upnp.device]
ignore_errors = true
[mypy-homeassistant.components.upnp.sensor]
ignore_errors = true
[mypy-homeassistant.components.vizio.config_flow] [mypy-homeassistant.components.vizio.config_flow]
ignore_errors = true ignore_errors = true

View File

@ -150,11 +150,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.unifi.device_tracker", "homeassistant.components.unifi.device_tracker",
"homeassistant.components.unifi.diagnostics", "homeassistant.components.unifi.diagnostics",
"homeassistant.components.unifi.unifi_entity_base", "homeassistant.components.unifi.unifi_entity_base",
"homeassistant.components.upnp",
"homeassistant.components.upnp.binary_sensor",
"homeassistant.components.upnp.config_flow",
"homeassistant.components.upnp.device",
"homeassistant.components.upnp.sensor",
"homeassistant.components.vizio.config_flow", "homeassistant.components.vizio.config_flow",
"homeassistant.components.vizio.media_player", "homeassistant.components.vizio.media_player",
"homeassistant.components.withings", "homeassistant.components.withings",