Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve Elgato code quality (#46505)

pull/46513/head
Franck Nijhof 2021-02-13 23:50:25 +01:00 committed by GitHub
parent eecf07d7df
commit 7148071be8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 153 additions and 234 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

128
homeassistant/components/elgato/config_flow.py
View File

@ -1,13 +1,15 @@
"""Config flow to configure the Elgato Key Light integration."""
from typing import Any, Dict, Optional
from __future__ import annotations
from elgato import Elgato, ElgatoError, Info
from typing import Any, Dict
from elgato import Elgato, ElgatoError
import voluptuous as vol
from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL, ConfigFlow
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import CONF_SERIAL_NUMBER, DOMAIN # pylint: disable=unused-import
@ -18,91 +20,54 @@ class ElgatoFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1
CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL
host: str
port: int
serial_number: str
async def async_step_user(
self, user_input: Optional[ConfigType] = None
self, user_input: Dict[str, Any] | None = None
) -> Dict[str, Any]:
"""Handle a flow initiated by the user."""
if user_input is None:
return self._show_setup_form()
return self._async_show_setup_form()
self.host = user_input[CONF_HOST]
self.port = user_input[CONF_PORT]
try:
info = await self._get_elgato_info(
user_input[CONF_HOST], user_input[CONF_PORT]
)
await self._get_elgato_serial_number(raise_on_progress=False)
except ElgatoError:
return self._show_setup_form({"base": "cannot_connect"})
return self._async_show_setup_form({"base": "cannot_connect"})
# Check if already configured
await self.async_set_unique_id(info.serial_number)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=info.serial_number,
data={
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
CONF_SERIAL_NUMBER: info.serial_number,
},
)
return self._async_create_entry()
async def async_step_zeroconf(
self, user_input: Optional[ConfigType] = None
self, discovery_info: Dict[str, Any]
) -> Dict[str, Any]:
"""Handle zeroconf discovery."""
if user_input is None:
return self.async_abort(reason="cannot_connect")
self.host = discovery_info[CONF_HOST]
self.port = discovery_info[CONF_PORT]
try:
info = await self._get_elgato_info(
user_input[CONF_HOST], user_input[CONF_PORT]
)
await self._get_elgato_serial_number()
except ElgatoError:
return self.async_abort(reason="cannot_connect")
# Check if already configured
await self.async_set_unique_id(info.serial_number)
self._abort_if_unique_id_configured(updates={CONF_HOST: user_input[CONF_HOST]})
self.context.update(
{
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
CONF_SERIAL_NUMBER: info.serial_number,
"title_placeholders": {"serial_number": info.serial_number},
}
return self.async_show_form(
step_id="zeroconf_confirm",
description_placeholders={"serial_number": self.serial_number},
)
# Prepare configuration flow
return self._show_confirm_dialog()
async def async_step_zeroconf_confirm(
self, user_input: ConfigType = None
self, _: Dict[str, Any] | None = None
) -> Dict[str, Any]:
"""Handle a flow initiated by zeroconf."""
if user_input is None:
return self._show_confirm_dialog()
return self._async_create_entry()
try:
info = await self._get_elgato_info(
self.context.get(CONF_HOST), self.context.get(CONF_PORT)
)
except ElgatoError:
return self.async_abort(reason="cannot_connect")
# Check if already configured
await self.async_set_unique_id(info.serial_number)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=self.context.get(CONF_SERIAL_NUMBER),
data={
CONF_HOST: self.context.get(CONF_HOST),
CONF_PORT: self.context.get(CONF_PORT),
CONF_SERIAL_NUMBER: self.context.get(CONF_SERIAL_NUMBER),
},
)
def _show_setup_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
@callback
def _async_show_setup_form(
self, errors: Dict[str, str] | None = None
) -> Dict[str, Any]:
"""Show the setup form to the user."""
return self.async_show_form(
step_id="user",
@ -115,20 +80,33 @@ class ElgatoFlowHandler(ConfigFlow, domain=DOMAIN):
errors=errors or {},
)
def _show_confirm_dialog(self) -> Dict[str, Any]:
"""Show the confirm dialog to the user."""
serial_number = self.context.get(CONF_SERIAL_NUMBER)
return self.async_show_form(
step_id="zeroconf_confirm",
description_placeholders={"serial_number": serial_number},
@callback
def _async_create_entry(self) -> Dict[str, Any]:
return self.async_create_entry(
title=self.serial_number,
data={
CONF_HOST: self.host,
CONF_PORT: self.port,
CONF_SERIAL_NUMBER: self.serial_number,
},
)
async def _get_elgato_info(self, host: str, port: int) -> Info:
async def _get_elgato_serial_number(self, raise_on_progress: bool = True) -> None:
"""Get device information from an Elgato Key Light device."""
session = async_get_clientsession(self.hass)
elgato = Elgato(
host,
port=port,
host=self.host,
port=self.port,
session=session,
)
return await elgato.info()
info = await elgato.info()
# Check if already configured
await self.async_set_unique_id(
info.serial_number, raise_on_progress=raise_on_progress
)
self._abort_if_unique_id_configured(
updates={CONF_HOST: self.host, CONF_PORT: self.port}
)
self.serial_number = info.serial_number

24
homeassistant/components/elgato/light.py
View File

@ -1,7 +1,9 @@
"""Support for LED lights."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable, Dict, List
from elgato import Elgato, ElgatoError, Info, State
@ -42,7 +44,7 @@ async def async_setup_entry(
"""Set up Elgato Key Light based on a config entry."""
elgato: Elgato = hass.data[DOMAIN][entry.entry_id][DATA_ELGATO_CLIENT]
info = await elgato.info()
async_add_entities([ElgatoLight(entry.entry_id, elgato, info)], True)
async_add_entities([ElgatoLight(elgato, info)], True)
class ElgatoLight(LightEntity):
@ -50,15 +52,14 @@ class ElgatoLight(LightEntity):
def __init__(
self,
entry_id: str,
elgato: Elgato,
info: Info,
):
"""Initialize Elgato Key Light."""
self._brightness: Optional[int] = None
self._brightness: int | None = None
self._info: Info = info
self._state: Optional[bool] = None
self._temperature: Optional[int] = None
self._state: bool | None = None
self._temperature: int | None = None
self._available = True
self.elgato = elgato
@ -81,22 +82,22 @@ class ElgatoLight(LightEntity):
return self._info.serial_number
@property
def brightness(self) -> Optional[int]:
def brightness(self) -> int | None:
"""Return the brightness of this light between 1..255."""
return self._brightness
@property
def color_temp(self):
def color_temp(self) -> int | None:
"""Return the CT color value in mireds."""
return self._temperature
@property
def min_mireds(self):
def min_mireds(self) -> int:
"""Return the coldest color_temp that this light supports."""
return 143
@property
def max_mireds(self):
def max_mireds(self) -> int:
"""Return the warmest color_temp that this light supports."""
return 344
@ -116,9 +117,8 @@ class ElgatoLight(LightEntity):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the light."""
data = {}
data: Dict[str, bool | int] = {ATTR_ON: True}
data[ATTR_ON] = True
if ATTR_ON in kwargs:
data[ATTR_ON] = kwargs[ATTR_ON]

11
tests/components/elgato/__init__.py
View File

@ -14,27 +14,26 @@ async def init_integration(
skip_setup: bool = False,
) -> MockConfigEntry:
"""Set up the Elgato Key Light integration in Home Assistant."""
aioclient_mock.get(
"http://1.2.3.4:9123/elgato/accessory-info",
"http://127.0.0.1:9123/elgato/accessory-info",
text=load_fixture("elgato/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
aioclient_mock.put(
"http://1.2.3.4:9123/elgato/lights",
"http://127.0.0.1:9123/elgato/lights",
text=load_fixture("elgato/state.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
"http://1.2.3.4:9123/elgato/lights",
"http://127.0.0.1:9123/elgato/lights",
text=load_fixture("elgato/state.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
"http://5.6.7.8:9123/elgato/accessory-info",
"http://127.0.0.2:9123/elgato/accessory-info",
text=load_fixture("elgato/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
@ -43,7 +42,7 @@ async def init_integration(
domain=DOMAIN,
unique_id="CN11A1A00001",
data={
CONF_HOST: "1.2.3.4",
CONF_HOST: "127.0.0.1",
CONF_PORT: 9123,
CONF_SERIAL_NUMBER: "CN11A1A00001",
},

219
tests/components/elgato/test_config_flow.py
View File

@ -2,10 +2,9 @@
import aiohttp
from homeassistant import data_entry_flow
from homeassistant.components.elgato import config_flow
from homeassistant.components.elgato.const import CONF_SERIAL_NUMBER
from homeassistant.components.elgato.const import CONF_SERIAL_NUMBER, DOMAIN
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
from homeassistant.const import CONF_HOST, CONF_PORT, CONTENT_TYPE_JSON
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SOURCE, CONTENT_TYPE_JSON
from homeassistant.core import HomeAssistant
from . import init_integration
@ -14,62 +13,97 @@ from tests.common import load_fixture
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_show_user_form(hass: HomeAssistant) -> None:
"""Test that the user set up form is served."""
async def test_full_user_flow_implementation(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the full manual user flow from start to finish."""
aioclient_mock.get(
"http://127.0.0.1:9123/elgato/accessory-info",
text=load_fixture("elgato/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
# Start a discovered configuration flow, to guarantee a user flow doesn't abort
await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_ZEROCONF},
data={
"host": "127.0.0.1",
"hostname": "example.local.",
"port": 9123,
"properties": {},
},
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": SOURCE_USER},
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "127.0.0.1", CONF_PORT: 9123}
)
async def test_show_zeroconf_confirm_form(hass: HomeAssistant) -> None:
"""Test that the zeroconf confirmation form is served."""
flow = config_flow.ElgatoFlowHandler()
flow.hass = hass
flow.context = {"source": SOURCE_ZEROCONF, CONF_SERIAL_NUMBER: "12345"}
result = await flow.async_step_zeroconf_confirm()
assert result["data"][CONF_HOST] == "127.0.0.1"
assert result["data"][CONF_PORT] == 9123
assert result["data"][CONF_SERIAL_NUMBER] == "CN11A1A00001"
assert result["title"] == "CN11A1A00001"
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["description_placeholders"] == {CONF_SERIAL_NUMBER: "12345"}
assert result["step_id"] == "zeroconf_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
entries = hass.config_entries.async_entries(DOMAIN)
assert entries[0].unique_id == "CN11A1A00001"
async def test_show_zerconf_form(
async def test_full_zeroconf_flow_implementation(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test that the zeroconf confirmation form is served."""
"""Test the zeroconf flow from start to finish."""
aioclient_mock.get(
"http://1.2.3.4:9123/elgato/accessory-info",
"http://127.0.0.1:9123/elgato/accessory-info",
text=load_fixture("elgato/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
flow = config_flow.ElgatoFlowHandler()
flow.hass = hass
flow.context = {"source": SOURCE_ZEROCONF}
result = await flow.async_step_zeroconf({"host": "1.2.3.4", "port": 9123})
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_ZEROCONF},
data={
"host": "127.0.0.1",
"hostname": "example.local.",
"port": 9123,
"properties": {},
},
)
assert flow.context[CONF_HOST] == "1.2.3.4"
assert flow.context[CONF_PORT] == 9123
assert flow.context[CONF_SERIAL_NUMBER] == "CN11A1A00001"
assert result["description_placeholders"] == {CONF_SERIAL_NUMBER: "CN11A1A00001"}
assert result["step_id"] == "zeroconf_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["data"][CONF_HOST] == "127.0.0.1"
assert result["data"][CONF_PORT] == 9123
assert result["data"][CONF_SERIAL_NUMBER] == "CN11A1A00001"
assert result["title"] == "CN11A1A00001"
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
async def test_connection_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we show user form on Elgato Key Light connection error."""
aioclient_mock.get("http://1.2.3.4/elgato/accessory-info", exc=aiohttp.ClientError)
aioclient_mock.get(
"http://127.0.0.1/elgato/accessory-info", exc=aiohttp.ClientError
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": SOURCE_USER},
data={CONF_HOST: "1.2.3.4", CONF_PORT: 9123},
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
data={CONF_HOST: "127.0.0.1", CONF_PORT: 9123},
)
assert result["errors"] == {"base": "cannot_connect"}
@ -81,51 +115,20 @@ async def test_zeroconf_connection_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort zeroconf flow on Elgato Key Light connection error."""
aioclient_mock.get("http://1.2.3.4/elgato/accessory-info", exc=aiohttp.ClientError)
aioclient_mock.get(
"http://127.0.0.1/elgato/accessory-info", exc=aiohttp.ClientError
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data={"host": "1.2.3.4", "port": 9123},
data={"host": "127.0.0.1", "port": 9123},
)
assert result["reason"] == "cannot_connect"
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
async def test_zeroconf_confirm_connection_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort zeroconf flow on Elgato Key Light connection error."""
aioclient_mock.get("http://1.2.3.4/elgato/accessory-info", exc=aiohttp.ClientError)
flow = config_flow.ElgatoFlowHandler()
flow.hass = hass
flow.context = {
"source": SOURCE_ZEROCONF,
CONF_HOST: "1.2.3.4",
CONF_PORT: 9123,
}
result = await flow.async_step_zeroconf_confirm(
user_input={CONF_HOST: "1.2.3.4", CONF_PORT: 9123}
)
assert result["reason"] == "cannot_connect"
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
async def test_zeroconf_no_data(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort if zeroconf provides no data."""
flow = config_flow.ElgatoFlowHandler()
flow.hass = hass
result = await flow.async_step_zeroconf()
assert result["reason"] == "cannot_connect"
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
async def test_user_device_exists_abort(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
@ -133,9 +136,9 @@ async def test_user_device_exists_abort(
await init_integration(hass, aioclient_mock)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": SOURCE_USER},
data={CONF_HOST: "1.2.3.4", CONF_PORT: 9123},
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
data={CONF_HOST: "127.0.0.1", CONF_PORT: 9123},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
@ -148,84 +151,22 @@ async def test_zeroconf_device_exists_abort(
await init_integration(hass, aioclient_mock)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": SOURCE_ZEROCONF},
data={"host": "1.2.3.4", "port": 9123},
DOMAIN,
context={CONF_SOURCE: SOURCE_ZEROCONF},
data={"host": "127.0.0.1", "port": 9123},
)
assert result["reason"] == "already_configured"
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": SOURCE_ZEROCONF, CONF_HOST: "1.2.3.4", "port": 9123},
data={"host": "5.6.7.8", "port": 9123},
DOMAIN,
context={CONF_SOURCE: SOURCE_ZEROCONF},
data={"host": "127.0.0.2", "port": 9123},
)
assert result["reason"] == "already_configured"
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
entries = hass.config_entries.async_entries(config_flow.DOMAIN)
assert entries[0].data[CONF_HOST] == "5.6.7.8"
async def test_full_user_flow_implementation(
hass: HomeAssistant, aioclient_mock
) -> None:
"""Test the full manual user flow from start to finish."""
aioclient_mock.get(
"http://1.2.3.4:9123/elgato/accessory-info",
text=load_fixture("elgato/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": SOURCE_USER},
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "1.2.3.4", CONF_PORT: 9123}
)
assert result["data"][CONF_HOST] == "1.2.3.4"
assert result["data"][CONF_PORT] == 9123
assert result["data"][CONF_SERIAL_NUMBER] == "CN11A1A00001"
assert result["title"] == "CN11A1A00001"
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
entries = hass.config_entries.async_entries(config_flow.DOMAIN)
assert entries[0].unique_id == "CN11A1A00001"
async def test_full_zeroconf_flow_implementation(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the full manual user flow from start to finish."""
aioclient_mock.get(
"http://1.2.3.4:9123/elgato/accessory-info",
text=load_fixture("elgato/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
flow = config_flow.ElgatoFlowHandler()
flow.hass = hass
flow.context = {"source": SOURCE_ZEROCONF}
result = await flow.async_step_zeroconf({"host": "1.2.3.4", "port": 9123})
assert flow.context[CONF_HOST] == "1.2.3.4"
assert flow.context[CONF_PORT] == 9123
assert flow.context[CONF_SERIAL_NUMBER] == "CN11A1A00001"
assert result["description_placeholders"] == {CONF_SERIAL_NUMBER: "CN11A1A00001"}
assert result["step_id"] == "zeroconf_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await flow.async_step_zeroconf_confirm(user_input={CONF_HOST: "1.2.3.4"})
assert result["data"][CONF_HOST] == "1.2.3.4"
assert result["data"][CONF_PORT] == 9123
assert result["data"][CONF_SERIAL_NUMBER] == "CN11A1A00001"
assert result["title"] == "CN11A1A00001"
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
entries = hass.config_entries.async_entries(DOMAIN)
assert entries[0].data[CONF_HOST] == "127.0.0.2"

2
tests/components/elgato/test_init.py
View File

@ -14,7 +14,7 @@ async def test_config_entry_not_ready(
) -> None:
"""Test the Elgato Key Light configuration entry not ready."""
aioclient_mock.get(
"http://1.2.3.4:9123/elgato/accessory-info", exc=aiohttp.ClientError
"http://127.0.0.1:9123/elgato/accessory-info", exc=aiohttp.ClientError
)
entry = await init_integration(hass, aioclient_mock)

3
tests/components/elgato/test_light.py
View File

@ -1,7 +1,8 @@
"""Tests for the Elgato Key Light light platform."""
from unittest.mock import patch
from homeassistant.components.elgato.light import ElgatoError
from elgato import ElgatoError
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP,