home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Enable strict typing to emulated_hue (#72676)

Browse Source 
* Add typing to emulated_hue part 2

* cleanups

* adjust targets in test
This commit is contained in:
J. Nick Koston
2022-05-29 20:49:37 -10:00
committed by GitHub
parent 6e355e1074
commit 1c334605b6
9 changed files with 165 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -83,6 +83,7 @@ homeassistant.components.dunehd.*
homeassistant.components.efergy.*
homeassistant.components.elgato.*
homeassistant.components.elkm1.*
homeassistant.components.emulated_hue.*
homeassistant.components.esphome.*
homeassistant.components.energy.*
homeassistant.components.evil_genius_labs.*

97
homeassistant/components/emulated_hue/__init__.py
View File

@ -13,7 +13,7 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_START,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import Event, HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
@ -48,11 +48,7 @@ from .hue_api import (
HueUnauthorizedUser,
HueUsernameView,
)
from .upnp import (
DescriptionXmlView,
UPNPResponderProtocol,
create_upnp_datagram_endpoint,
)
from .upnp import DescriptionXmlView, async_create_upnp_datagram_endpoint
_LOGGER = logging.getLogger(__name__)
@ -93,6 +89,40 @@ CONFIG_SCHEMA = vol.Schema(
)
async def start_emulated_hue_bridge(
hass: HomeAssistant, config: Config, app: web.Application
) -> None:
"""Start the emulated hue bridge."""
protocol = await async_create_upnp_datagram_endpoint(
config.host_ip_addr,
config.upnp_bind_multicast,
config.advertise_ip,
config.advertise_port or config.listen_port,
)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, config.host_ip_addr, config.listen_port)
try:
await site.start()
except OSError as error:
_LOGGER.error(
"Failed to create HTTP server at port %d: %s", config.listen_port, error
)
protocol.close()
return
async def stop_emulated_hue_bridge(event: Event) -> None:
"""Stop the emulated hue bridge."""
protocol.close()
await site.stop()
await runner.cleanup()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_emulated_hue_bridge)
async def async_setup(hass: HomeAssistant, yaml_config: ConfigType) -> bool:
"""Activate the emulated_hue component."""
local_ip = await async_get_source_ip(hass)
@ -108,9 +138,6 @@ async def async_setup(hass: HomeAssistant, yaml_config: ConfigType) -> bool:
app._on_startup.freeze()
await app.startup()
runner = None
site = None
DescriptionXmlView(config).register(app, app.router)
HueUsernameView().register(app, app.router)
HueConfigView(config).register(app, app.router)
@ -122,54 +149,10 @@ async def async_setup(hass: HomeAssistant, yaml_config: ConfigType) -> bool:
HueGroupView(config).register(app, app.router)
HueFullStateView(config).register(app, app.router)
listen = create_upnp_datagram_endpoint(
config.host_ip_addr,
config.upnp_bind_multicast,
config.advertise_ip,
config.advertise_port or config.listen_port,
)
protocol: UPNPResponderProtocol | None = None
async def _start(event: Event) -> None:
"""Start the bridge."""
await start_emulated_hue_bridge(hass, config, app)
async def stop_emulated_hue_bridge(event):
"""Stop the emulated hue bridge."""
nonlocal protocol
nonlocal site
nonlocal runner
if protocol:
protocol.close()
if site:
await site.stop()
if runner:
await runner.cleanup()
async def start_emulated_hue_bridge(event):
"""Start the emulated hue bridge."""
nonlocal protocol
nonlocal site
nonlocal runner
transport_protocol = await listen
protocol = transport_protocol[1]
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, config.host_ip_addr, config.listen_port)
try:
await site.start()
except OSError as error:
_LOGGER.error(
"Failed to create HTTP server at port %d: %s", config.listen_port, error
)
if protocol:
protocol.close()
else:
hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, stop_emulated_hue_bridge
)
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, start_emulated_hue_bridge)
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, _start)
return True

21
homeassistant/components/emulated_hue/config.py
View File

@ -55,9 +55,7 @@ _LOGGER = logging.getLogger(__name__)
class Config:
"""Hold configuration variables for the emulated hue bridge."""
def __init__(
self, hass: HomeAssistant, conf: ConfigType, local_ip: str | None
) -> None:
def __init__(self, hass: HomeAssistant, conf: ConfigType, local_ip: str) -> None:
"""Initialize the instance."""
self.hass = hass
self.type = conf.get(CONF_TYPE)
@ -73,17 +71,10 @@ class Config:
)
# Get the IP address that will be passed to the Echo during discovery
self.host_ip_addr = conf.get(CONF_HOST_IP)
if self.host_ip_addr is None:
self.host_ip_addr = local_ip
self.host_ip_addr: str = conf.get(CONF_HOST_IP) or local_ip
# Get the port that the Hue bridge will listen on
self.listen_port = conf.get(CONF_LISTEN_PORT)
if not isinstance(self.listen_port, int):
self.listen_port = DEFAULT_LISTEN_PORT
_LOGGER.info(
"Listen port not specified, defaulting to %s", self.listen_port
)
self.listen_port: int = conf.get(CONF_LISTEN_PORT) or DEFAULT_LISTEN_PORT
# Get whether or not UPNP binds to multicast address (239.255.255.250)
# or to the unicast address (host_ip_addr)
@ -113,11 +104,11 @@ class Config:
)
# Calculated effective advertised IP and port for network isolation
self.advertise_ip = conf.get(CONF_ADVERTISE_IP) or self.host_ip_addr
self.advertise_ip: str = conf.get(CONF_ADVERTISE_IP) or self.host_ip_addr
self.advertise_port = conf.get(CONF_ADVERTISE_PORT) or self.listen_port
self.advertise_port: int = conf.get(CONF_ADVERTISE_PORT) or self.listen_port
self.entities = conf.get(CONF_ENTITIES, {})
self.entities: dict[str, dict[str, str]] = conf.get(CONF_ENTITIES, {})
self._entities_with_hidden_attr_in_config = {}
for entity_id in self.entities:

2
homeassistant/components/emulated_hue/hue_api.py
View File

@ -858,7 +858,7 @@ async def wait_for_state_change_or_timeout(
ev = asyncio.Event()
@core.callback
def _async_event_changed(_):
def _async_event_changed(event: core.Event) -> None:
ev.set()
unsub = async_track_state_change_event(hass, [entity_id], _async_event_changed)

175
homeassistant/components/emulated_hue/upnp.py
View File

@ -1,13 +1,17 @@
"""Support UPNP discovery method that mimics Hue hubs."""
from __future__ import annotations
import asyncio
import logging
import socket
from typing import cast
from aiohttp import web
from homeassistant import core
from homeassistant.components.http import HomeAssistantView
from .config import Config
from .const import HUE_SERIAL_NUMBER, HUE_UUID
_LOGGER = logging.getLogger(__name__)
@ -23,12 +27,12 @@ class DescriptionXmlView(HomeAssistantView):
name = "description:xml"
requires_auth = False
def __init__(self, config):
def __init__(self, config: Config) -> None:
"""Initialize the instance of the view."""
self.config = config
@core.callback
def get(self, request):
def get(self, request: web.Request) -> web.Response:
"""Handle a GET request."""
resp_text = f"""<?xml version="1.0" encoding="UTF-8" ?>
<root xmlns="urn:schemas-upnp-org:device-1-0">
@ -55,13 +59,91 @@ class DescriptionXmlView(HomeAssistantView):
return web.Response(text=resp_text, content_type="text/xml")
@core.callback
def create_upnp_datagram_endpoint(
host_ip_addr,
upnp_bind_multicast,
advertise_ip,
advertise_port,
):
class UPNPResponderProtocol(asyncio.Protocol):
"""Handle responding to UPNP/SSDP discovery requests."""
def __init__(
self,
loop: asyncio.AbstractEventLoop,
ssdp_socket: socket.socket,
advertise_ip: str,
advertise_port: int,
) -> None:
"""Initialize the class."""
self.transport: asyncio.DatagramTransport | None = None
self._loop = loop
self._sock = ssdp_socket
self.advertise_ip = advertise_ip
self.advertise_port = advertise_port
self._upnp_root_response = self._prepare_response(
"upnp:rootdevice", f"uuid:{HUE_UUID}::upnp:rootdevice"
)
self._upnp_device_response = self._prepare_response(
"urn:schemas-upnp-org:device:basic:1", f"uuid:{HUE_UUID}"
)
def connection_made(self, transport: asyncio.BaseTransport) -> None:
"""Set the transport."""
self.transport = cast(asyncio.DatagramTransport, transport)
def connection_lost(self, exc: Exception | None) -> None:
"""Handle connection lost."""
def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None:
"""Respond to msearch packets."""
decoded_data = data.decode("utf-8", errors="ignore")
if "M-SEARCH" not in decoded_data:
return
_LOGGER.debug("UPNP Responder M-SEARCH method received: %s", data)
# SSDP M-SEARCH method received, respond to it with our info
response = self._handle_request(decoded_data)
_LOGGER.debug("UPNP Responder responding with: %s", response)
assert self.transport is not None
self.transport.sendto(response, addr)
def error_received(self, exc: Exception) -> None:
"""Log UPNP errors."""
_LOGGER.error("UPNP Error received: %s", exc)
def close(self) -> None:
"""Stop the server."""
_LOGGER.info("UPNP responder shutting down")
if self.transport:
self.transport.close()
self._loop.remove_writer(self._sock.fileno())
self._loop.remove_reader(self._sock.fileno())
self._sock.close()
def _handle_request(self, decoded_data: str) -> bytes:
if "upnp:rootdevice" in decoded_data:
return self._upnp_root_response
return self._upnp_device_response
def _prepare_response(self, search_target: str, unique_service_name: str) -> bytes:
# Note that the double newline at the end of
# this string is required per the SSDP spec
response = f"""HTTP/1.1 200 OK
CACHE-CONTROL: max-age=60
EXT:
LOCATION: http://{self.advertise_ip}:{self.advertise_port}/description.xml
SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
hue-bridgeid: {HUE_SERIAL_NUMBER}
ST: {search_target}
USN: {unique_service_name}
"""
return response.replace("\n", "\r\n").encode("utf-8")
async def async_create_upnp_datagram_endpoint(
host_ip_addr: str,
upnp_bind_multicast: bool,
advertise_ip: str,
advertise_port: int,
) -> UPNPResponderProtocol:
"""Create the UPNP socket and protocol."""
# Listen for UDP port 1900 packets sent to SSDP multicast address
ssdp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@ -84,79 +166,8 @@ def create_upnp_datagram_endpoint(
loop = asyncio.get_event_loop()
return loop.create_datagram_endpoint(
transport_protocol = await loop.create_datagram_endpoint(
lambda: UPNPResponderProtocol(loop, ssdp_socket, advertise_ip, advertise_port),
sock=ssdp_socket,
)
class UPNPResponderProtocol:
"""Handle responding to UPNP/SSDP discovery requests."""
def __init__(self, loop, ssdp_socket, advertise_ip, advertise_port):
"""Initialize the class."""
self.transport = None
self._loop = loop
self._sock = ssdp_socket
self.advertise_ip = advertise_ip
self.advertise_port = advertise_port
self._upnp_root_response = self._prepare_response(
"upnp:rootdevice", f"uuid:{HUE_UUID}::upnp:rootdevice"
)
self._upnp_device_response = self._prepare_response(
"urn:schemas-upnp-org:device:basic:1", f"uuid:{HUE_UUID}"
)
def connection_made(self, transport):
"""Set the transport."""
self.transport = transport
def connection_lost(self, exc):
"""Handle connection lost."""
def datagram_received(self, data, addr):
"""Respond to msearch packets."""
decoded_data = data.decode("utf-8", errors="ignore")
if "M-SEARCH" not in decoded_data:
return
_LOGGER.debug("UPNP Responder M-SEARCH method received: %s", data)
# SSDP M-SEARCH method received, respond to it with our info
response = self._handle_request(decoded_data)
_LOGGER.debug("UPNP Responder responding with: %s", response)
self.transport.sendto(response, addr)
def error_received(self, exc):
"""Log UPNP errors."""
_LOGGER.error("UPNP Error received: %s", exc)
def close(self):
"""Stop the server."""
_LOGGER.info("UPNP responder shutting down")
if self.transport:
self.transport.close()
self._loop.remove_writer(self._sock.fileno())
self._loop.remove_reader(self._sock.fileno())
self._sock.close()
def _handle_request(self, decoded_data):
if "upnp:rootdevice" in decoded_data:
return self._upnp_root_response
return self._upnp_device_response
def _prepare_response(self, search_target, unique_service_name):
# Note that the double newline at the end of
# this string is required per the SSDP spec
response = f"""HTTP/1.1 200 OK
CACHE-CONTROL: max-age=60
EXT:
LOCATION: http://{self.advertise_ip}:{self.advertise_port}/description.xml
SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
hue-bridgeid: {HUE_SERIAL_NUMBER}
ST: {search_target}
USN: {unique_service_name}
"""
return response.replace("\n", "\r\n").encode("utf-8")
return transport_protocol[1]

11
mypy.ini
View File

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

8
tests/components/emulated_hue/test_hue_api.py
View File

@ -108,7 +108,9 @@ def hass_hue(loop, hass):
)
)
with patch("homeassistant.components.emulated_hue.create_upnp_datagram_endpoint"):
with patch(
"homeassistant.components.emulated_hue.async_create_upnp_datagram_endpoint"
):
loop.run_until_complete(
setup.async_setup_component(
hass,
@ -314,7 +316,9 @@ async def test_lights_all_dimmable(hass, hass_client_no_auth):
emulated_hue.CONF_EXPOSE_BY_DEFAULT: True,
emulated_hue.CONF_LIGHTS_ALL_DIMMABLE: True,
}
with patch("homeassistant.components.emulated_hue.create_upnp_datagram_endpoint"):
with patch(
"homeassistant.components.emulated_hue.async_create_upnp_datagram_endpoint"
):
await setup.async_setup_component(
hass,
emulated_hue.DOMAIN,

8
tests/components/emulated_hue/test_init.py
View File

@ -122,13 +122,13 @@ async def test_setup_works(hass):
"""Test setup works."""
hass.config.components.add("network")
with patch(
"homeassistant.components.emulated_hue.create_upnp_datagram_endpoint",
"homeassistant.components.emulated_hue.async_create_upnp_datagram_endpoint",
AsyncMock(),
) as mock_create_upnp_datagram_endpoint, patch(
"homeassistant.components.emulated_hue.async_get_source_ip"
):
assert await async_setup_component(hass, "emulated_hue", {})
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
assert len(mock_create_upnp_datagram_endpoint.mock_calls) == 2
assert len(mock_create_upnp_datagram_endpoint.mock_calls) == 1

4
tests/components/emulated_hue/test_upnp.py
View File

@ -53,7 +53,9 @@ def hue_client(aiohttp_client):
async def setup_hue(hass):
"""Set up the emulated_hue integration."""
with patch("homeassistant.components.emulated_hue.create_upnp_datagram_endpoint"):
with patch(
"homeassistant.components.emulated_hue.async_create_upnp_datagram_endpoint"
):
assert await setup.async_setup_component(
hass,
emulated_hue.DOMAIN,