diff --git a/homeassistant/components/elgato/config_flow.py b/homeassistant/components/elgato/config_flow.py index 60cc08dc9b3..e9138afd86c 100644 --- a/homeassistant/components/elgato/config_flow.py +++ b/homeassistant/components/elgato/config_flow.py @@ -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 diff --git a/homeassistant/components/elgato/light.py b/homeassistant/components/elgato/light.py index 313b5600248..eea80e60b15 100644 --- a/homeassistant/components/elgato/light.py +++ b/homeassistant/components/elgato/light.py @@ -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] diff --git a/tests/components/elgato/__init__.py b/tests/components/elgato/__init__.py index 3b1942aee14..ea63bc0c4d0 100644 --- a/tests/components/elgato/__init__.py +++ b/tests/components/elgato/__init__.py @@ -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", }, diff --git a/tests/components/elgato/test_config_flow.py b/tests/components/elgato/test_config_flow.py index c1dfa697041..0f3ff032722 100644 --- a/tests/components/elgato/test_config_flow.py +++ b/tests/components/elgato/test_config_flow.py @@ -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" diff --git a/tests/components/elgato/test_init.py b/tests/components/elgato/test_init.py index 2f0e39e05a8..069e533c423 100644 --- a/tests/components/elgato/test_init.py +++ b/tests/components/elgato/test_init.py @@ -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) diff --git a/tests/components/elgato/test_light.py b/tests/components/elgato/test_light.py index 838608c0aac..aed569c18fe 100644 --- a/tests/components/elgato/test_light.py +++ b/tests/components/elgato/test_light.py @@ -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,