Add Elgato Key Light integration (#29592)

* Add Elgato Key Light integration

* Remove passing in of hass loop

* Tweaks a comment

* Tweaks a function name

* Ensure domain namespace in data exists in entry setup

CODEOWNERS

@@ -85,6 +85,7 @@ homeassistant/components/ecobee/* @marthoc
 homeassistant/components/ecovacs/* @OverloadUT
 homeassistant/components/egardia/* @jeroenterheerdt
 homeassistant/components/eight_sleep/* @mezz64
+homeassistant/components/elgato/* @frenck
 homeassistant/components/elv/* @majuss
 homeassistant/components/emby/* @mezz64
 homeassistant/components/emulated_hue/* @NobleKangaroo

homeassistant/components/elgato/.translations/en.json

@@ -0,0 +1,27 @@
+{
+  "config": {
+    "title": "Elgato Key Light",
+    "flow_title": "Elgato Key Light: {serial_number}",
+    "step": {
+      "user": {
+        "title": "Link your Elgato Key Light",
+        "description": "Set up your Elgato Key Light to integrate with Home Assistant.",
+        "data": {
+          "host": "Host or IP address",
+          "port": "Port number"
+        }
+      },
+      "zeroconf_confirm": {
+        "description": "Do you want to add the Elgato Key Light with serial number `{serial_number}` to Home Assistant?",
+        "title": "Discovered Elgato Key Light device"
+      }
+    },
+    "error": {
+      "connection_error": "Failed to connect to Elgato Key Light device."
+    },
+    "abort": {
+      "already_configured": "This Elgato Key Light device is already configured.",
+      "connection_error": "Failed to connect to Elgato Key Light device."
+    }
+  }
+}

homeassistant/components/elgato/__init__.py

@@ -0,0 +1,55 @@
+"""Support for Elgato Key Lights."""
+import logging
+
+from elgato import Elgato, ElgatoConnectionError
+
+from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.typing import ConfigType
+
+from .const import DATA_ELGATO_CLIENT, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the Elgato Key Light components."""
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Elgato Key Light from a config entry."""
+    session = async_get_clientsession(hass)
+    elgato = Elgato(entry.data[CONF_HOST], port=entry.data[CONF_PORT], session=session,)
+
+    # Ensure we can connect to it
+    try:
+        await elgato.info()
+    except ElgatoConnectionError as exception:
+        raise ConfigEntryNotReady from exception
+
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = {DATA_ELGATO_CLIENT: elgato}
+
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(entry, LIGHT_DOMAIN)
+    )
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload Elgato Key Light config entry."""
+    # Unload entities for this entry/device.
+    await hass.config_entries.async_forward_entry_unload(entry, LIGHT_DOMAIN)
+
+    # Cleanup
+    del hass.data[DOMAIN][entry.entry_id]
+    if not hass.data[DOMAIN]:
+        del hass.data[DOMAIN]
+
+    return True

homeassistant/components/elgato/config_flow.py

@@ -0,0 +1,146 @@
+"""Config flow to configure the Elgato Key Light integration."""
+import logging
+from typing import Any, Dict, Optional
+
+from elgato import Elgato, ElgatoError, Info
+import voluptuous as vol
+
+from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL, ConfigFlow
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.helpers import ConfigType
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import CONF_SERIAL_NUMBER, DOMAIN  # pylint: disable=unused-import
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ElgatoFlowHandler(ConfigFlow, domain=DOMAIN):
+    """Handle a Elgato Key Light config flow."""
+
+    VERSION = 1
+    CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL
+
+    async def async_step_user(
+        self, user_input: Optional[ConfigType] = None
+    ) -> Dict[str, Any]:
+        """Handle a flow initiated by the user."""
+        if user_input is None:
+            return self._show_setup_form()
+
+        try:
+            info = await self._get_elgato_info(
+                user_input[CONF_HOST], user_input[CONF_PORT]
+            )
+        except ElgatoError:
+            return self._show_setup_form({"base": "connection_error"})
+
+        # Check if already configured
+        if await self._device_already_configured(info):
+            # This serial number is already configured
+            return self.async_abort(reason="already_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,
+            },
+        )
+
+    async def async_step_zeroconf(
+        self, user_input: Optional[ConfigType] = None
+    ) -> Dict[str, Any]:
+        """Handle zeroconf discovery."""
+        if user_input is None:
+            return self.async_abort(reason="connection_error")
+
+        # Hostname is format: my-ke.local.
+        host = user_input["hostname"].rstrip(".")
+        try:
+            info = await self._get_elgato_info(host, user_input[CONF_PORT])
+        except ElgatoError:
+            return self.async_abort(reason="connection_error")
+
+        # Check if already configured
+        if await self._device_already_configured(info):
+            # This serial number is already configured
+            return self.async_abort(reason="already_configured")
+
+        # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+        self.context.update(
+            {
+                CONF_HOST: host,
+                CONF_PORT: user_input[CONF_PORT],
+                CONF_SERIAL_NUMBER: info.serial_number,
+                "title_placeholders": {"serial_number": info.serial_number},
+            }
+        )
+
+        # Prepare configuration flow
+        return self._show_confirm_dialog()
+
+    # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+    async def async_step_zeroconf_confirm(
+        self, user_input: ConfigType = None
+    ) -> Dict[str, Any]:
+        """Handle a flow initiated by zeroconf."""
+        if user_input is None:
+            return self._show_confirm_dialog()
+
+        try:
+            info = await self._get_elgato_info(
+                self.context.get(CONF_HOST), self.context.get(CONF_PORT)
+            )
+        except ElgatoError:
+            return self.async_abort(reason="connection_error")
+
+        # Check if already configured
+        if await self._device_already_configured(info):
+            # This serial number is already configured
+            return self.async_abort(reason="already_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]:
+        """Show the setup form to the user."""
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_HOST): str,
+                    vol.Optional(CONF_PORT, default=9123): int,
+                }
+            ),
+            errors=errors or {},
+        )
+
+    def _show_confirm_dialog(self) -> Dict[str, Any]:
+        """Show the confirm dialog to the user."""
+        # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+        serial_number = self.context.get(CONF_SERIAL_NUMBER)
+        return self.async_show_form(
+            step_id="zeroconf_confirm",
+            description_placeholders={"serial_number": serial_number},
+        )
+
+    async def _get_elgato_info(self, host: str, port: int) -> Info:
+        """Get device information from an Elgato Key Light device."""
+        session = async_get_clientsession(self.hass)
+        elgato = Elgato(host, port=port, session=session,)
+        return await elgato.info()
+
+    async def _device_already_configured(self, info: Info) -> bool:
+        """Return if a Elgato Key Light is already configured."""
+        for entry in self._async_current_entries():
+            if entry.data[CONF_SERIAL_NUMBER] == info.serial_number:
+                return True
+        return False

homeassistant/components/elgato/const.py

@@ -0,0 +1,17 @@
+"""Constants for the Elgato Key Light integration."""
+
+# Integration domain
+DOMAIN = "elgato"
+
+# Hass data keys
+DATA_ELGATO_CLIENT = "elgato_client"
+
+# Attributes
+ATTR_IDENTIFIERS = "identifiers"
+ATTR_MANUFACTURER = "manufacturer"
+ATTR_MODEL = "model"
+ATTR_ON = "on"
+ATTR_SOFTWARE_VERSION = "sw_version"
+ATTR_TEMPERATURE = "temperature"
+
+CONF_SERIAL_NUMBER = "serial_number"

homeassistant/components/elgato/light.py

@@ -0,0 +1,158 @@
+"""Support for LED lights."""
+from datetime import timedelta
+import logging
+from typing import Any, Callable, Dict, List, Optional
+
+from elgato import Elgato, ElgatoError, Info, State
+
+from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
+    ATTR_COLOR_TEMP,
+    SUPPORT_BRIGHTNESS,
+    SUPPORT_COLOR_TEMP,
+    Light,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import ATTR_NAME
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.typing import HomeAssistantType
+
+from .const import (
+    ATTR_IDENTIFIERS,
+    ATTR_MANUFACTURER,
+    ATTR_MODEL,
+    ATTR_ON,
+    ATTR_SOFTWARE_VERSION,
+    ATTR_TEMPERATURE,
+    DATA_ELGATO_CLIENT,
+    DOMAIN,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+PARALLEL_UPDATES = 1
+SCAN_INTERVAL = timedelta(seconds=10)
+
+
+async def async_setup_entry(
+    hass: HomeAssistantType,
+    entry: ConfigEntry,
+    async_add_entities: Callable[[List[Entity], bool], None],
+) -> None:
+    """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)
+
+
+class ElgatoLight(Light):
+    """Defines a Elgato Key Light."""
+
+    def __init__(
+        self, entry_id: str, elgato: Elgato, info: Info,
+    ):
+        """Initialize Elgato Key Light."""
+        self._brightness: Optional[int] = None
+        self._info: Info = info
+        self._state: Optional[bool] = None
+        self._temperature: Optional[int] = None
+        self._available = True
+        self.elgato = elgato
+
+    @property
+    def name(self) -> str:
+        """Return the name of the entity."""
+        # Return the product name, if display name is not set
+        if not self._info.display_name:
+            return self._info.product_name
+        return self._info.display_name
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        return self._available
+
+    @property
+    def unique_id(self) -> str:
+        """Return the unique ID for this sensor."""
+        return self._info.serial_number
+
+    @property
+    def brightness(self) -> Optional[int]:
+        """Return the brightness of this light between 1..255."""
+        return self._brightness
+
+    @property
+    def color_temp(self):
+        """Return the CT color value in mireds."""
+        return self._temperature
+
+    @property
+    def min_mireds(self):
+        """Return the coldest color_temp that this light supports."""
+        return 143
+
+    @property
+    def max_mireds(self):
+        """Return the warmest color_temp that this light supports."""
+        return 344
+
+    @property
+    def supported_features(self) -> int:
+        """Flag supported features."""
+        return SUPPORT_BRIGHTNESS | SUPPORT_COLOR_TEMP
+
+    @property
+    def is_on(self) -> bool:
+        """Return the state of the light."""
+        return bool(self._state)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn off the light."""
+        await self.async_turn_on(on=False)
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn on the light."""
+        data = {}
+
+        data[ATTR_ON] = True
+        if ATTR_ON in kwargs:
+            data[ATTR_ON] = kwargs[ATTR_ON]
+
+        if ATTR_COLOR_TEMP in kwargs:
+            data[ATTR_TEMPERATURE] = kwargs[ATTR_COLOR_TEMP]
+
+        if ATTR_BRIGHTNESS in kwargs:
+            data[ATTR_BRIGHTNESS] = round((kwargs[ATTR_BRIGHTNESS] / 255) * 100)
+
+        try:
+            await self.elgato.light(**data)
+        except ElgatoError:
+            _LOGGER.error("An error occurred while updating the Elgato Key Light")
+            self._available = False
+
+    async def async_update(self) -> None:
+        """Update Elgato entity."""
+        try:
+            state: State = await self.elgato.state()
+        except ElgatoError:
+            if self._available:
+                _LOGGER.error("An error occurred while updating the Elgato Key Light")
+            self._available = False
+            return
+
+        self._available = True
+        self._brightness = round((state.brightness * 255) / 100)
+        self._state = state.on
+        self._temperature = state.temperature
+
+    @property
+    def device_info(self) -> Dict[str, Any]:
+        """Return device information about this Elgato Key Light."""
+        return {
+            ATTR_IDENTIFIERS: {(DOMAIN, self._info.serial_number)},
+            ATTR_NAME: self._info.product_name,
+            ATTR_MANUFACTURER: "Elgato",
+            ATTR_MODEL: self._info.product_name,
+            ATTR_SOFTWARE_VERSION: f"{self._info.firmware_version} ({self._info.firmware_build_number})",
+        }

homeassistant/components/elgato/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "elgato",
+  "name": "Elgato Key Light",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/elgato",
+  "requirements": ["elgato==0.1.0"],
+  "dependencies": [],
+  "zeroconf": ["_elg._tcp.local."],
+  "codeowners": ["@frenck"]
+}

homeassistant/components/elgato/strings.json

@@ -0,0 +1,27 @@
+{
+  "config": {
+    "title": "Elgato Key Light",
+    "flow_title": "Elgato Key Light: {serial_number}",
+    "step": {
+      "user": {
+        "title": "Link your Elgato Key Light",
+        "description": "Set up your Elgato Key Light to integrate with Home Assistant.",
+        "data": {
+          "host": "Host or IP address",
+          "port": "Port number"
+        }
+      },
+      "zeroconf_confirm": {
+        "description": "Do you want to add the Elgato Key Light with serial number `{serial_number}` to Home Assistant?",
+        "title": "Discovered Elgato Key Light device"
+      }
+    },
+    "error": {
+      "connection_error": "Failed to connect to Elgato Key Light device."
+    },
+    "abort": {
+      "already_configured": "This Elgato Key Light device is already configured.",
+      "connection_error": "Failed to connect to Elgato Key Light device."
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -20,6 +20,7 @@ FLOWS = [
     "deconz",
     "dialogflow",
     "ecobee",
+    "elgato",
     "emulated_roku",
     "esphome",
     "geofency",

homeassistant/generated/zeroconf.py

@@ -12,6 +12,9 @@ ZEROCONF = {
     "_coap._udp.local.": [
         "tradfri"
     ],
+    "_elg._tcp.local.": [
+        "elgato"
+    ],
     "_esphomelib._tcp.local.": [
         "esphome"
     ],

requirements_all.txt

@@ -459,6 +459,9 @@ ecoaliface==0.4.0
 # homeassistant.components.ee_brightbox
 eebrightbox==0.0.4
 
+# homeassistant.components.elgato
+elgato==0.1.0
+
 # homeassistant.components.eliqonline
 eliqonline==1.2.2
 

requirements_test_all.txt

@@ -158,6 +158,9 @@ dsmr_parser==0.12
 # homeassistant.components.ee_brightbox
 eebrightbox==0.0.4
 
+# homeassistant.components.elgato
+elgato==0.1.0
+
 # homeassistant.components.emulated_roku
 emulated_roku==0.1.8
 

tests/components/elgato/__init__.py

@@ -0,0 +1,49 @@
+"""Tests for the Elgato Key Light integration."""
+
+from homeassistant.components.elgato.const import CONF_SERIAL_NUMBER, DOMAIN
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry, load_fixture
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+async def init_integration(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, skip_setup: bool = False,
+) -> MockConfigEntry:
+    """Set up the Elgato Key Light integration in Home Assistant."""
+
+    aioclient_mock.get(
+        "http://example.local:9123/elgato/accessory-info",
+        text=load_fixture("elgato/info.json"),
+        headers={"Content-Type": "application/json"},
+    )
+
+    aioclient_mock.put(
+        "http://example.local:9123/elgato/lights",
+        text=load_fixture("elgato/state.json"),
+        headers={"Content-Type": "application/json"},
+    )
+
+    aioclient_mock.get(
+        "http://example.local:9123/elgato/lights",
+        text=load_fixture("elgato/state.json"),
+        headers={"Content-Type": "application/json"},
+    )
+
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            CONF_HOST: "example.local",
+            CONF_PORT: 9123,
+            CONF_SERIAL_NUMBER: "CN11A1A00001",
+        },
+    )
+
+    entry.add_to_hass(hass)
+
+    if not skip_setup:
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    return entry

tests/components/elgato/test_config_flow.py

@@ -0,0 +1,238 @@
+"""Tests for the Elgato Key Light config flow."""
+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.config_entries import SOURCE_USER, SOURCE_ZEROCONF
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.core import HomeAssistant
+
+from . import init_integration
+
+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."""
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_USER}
+    result = await flow.async_step_user(user_input=None)
+
+    assert result["step_id"] == "user"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+
+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["description_placeholders"] == {CONF_SERIAL_NUMBER: "12345"}
+    assert result["step_id"] == "zeroconf_confirm"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+
+async def test_show_zerconf_form(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test that the zeroconf confirmation form is served."""
+    aioclient_mock.get(
+        "http://example.local:9123/elgato/accessory-info",
+        text=load_fixture("elgato/info.json"),
+        headers={"Content-Type": "application/json"},
+    )
+
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_ZEROCONF}
+    result = await flow.async_step_zeroconf(
+        {"hostname": "example.local.", "port": 9123}
+    )
+
+    assert flow.context[CONF_HOST] == "example.local"
+    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
+
+
+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://example.local/elgato/accessory-info", exc=aiohttp.ClientError
+    )
+
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_USER}
+    result = await flow.async_step_user(
+        user_input={CONF_HOST: "example.local", CONF_PORT: 9123}
+    )
+
+    assert result["errors"] == {"base": "connection_error"}
+    assert result["step_id"] == "user"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+
+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://example.local/elgato/accessory-info", exc=aiohttp.ClientError
+    )
+
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_ZEROCONF}
+    result = await flow.async_step_zeroconf(
+        user_input={"hostname": "example.local.", "port": 9123}
+    )
+
+    assert result["reason"] == "connection_error"
+    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://example.local/elgato/accessory-info", exc=aiohttp.ClientError
+    )
+
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {
+        "source": SOURCE_ZEROCONF,
+        CONF_HOST: "example.local",
+        CONF_PORT: 9123,
+    }
+    result = await flow.async_step_zeroconf_confirm(
+        user_input={CONF_HOST: "example.local", CONF_PORT: 9123}
+    )
+
+    assert result["reason"] == "connection_error"
+    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"] == "connection_error"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+
+
+async def test_user_device_exists_abort(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we abort zeroconf flow if Elgato Key Light device already configured."""
+    await init_integration(hass, aioclient_mock)
+
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_USER}
+    result = await flow.async_step_user({CONF_HOST: "example.local", CONF_PORT: 9123})
+
+    assert result["reason"] == "already_configured"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+
+
+async def test_zeroconf_device_exists_abort(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we abort zeroconf flow if Elgato Key Light device already configured."""
+    await init_integration(hass, aioclient_mock)
+
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_ZEROCONF}
+    result = await flow.async_step_zeroconf(
+        {"hostname": "example.local.", "port": 9123}
+    )
+
+    assert result["reason"] == "already_configured"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+
+    flow.context = {"source": SOURCE_ZEROCONF, CONF_HOST: "example.local", "port": 9123}
+    result = await flow.async_step_zeroconf_confirm(
+        {"hostname": "example.local.", "port": 9123}
+    )
+
+    assert result["reason"] == "already_configured"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+
+
+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://example.local:9123/elgato/accessory-info",
+        text=load_fixture("elgato/info.json"),
+        headers={"Content-Type": "application/json"},
+    )
+
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_USER}
+    result = await flow.async_step_user(user_input=None)
+
+    assert result["step_id"] == "user"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+    result = await flow.async_step_user(
+        user_input={CONF_HOST: "example.local", CONF_PORT: 9123}
+    )
+    assert result["data"][CONF_HOST] == "example.local"
+    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_full_zeroconf_flow_implementation(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the full manual user flow from start to finish."""
+    aioclient_mock.get(
+        "http://example.local:9123/elgato/accessory-info",
+        text=load_fixture("elgato/info.json"),
+        headers={"Content-Type": "application/json"},
+    )
+
+    flow = config_flow.ElgatoFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_ZEROCONF}
+    result = await flow.async_step_zeroconf(
+        {"hostname": "example.local.", "port": 9123}
+    )
+
+    assert flow.context[CONF_HOST] == "example.local"
+    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: "example.local"}
+    )
+    assert result["data"][CONF_HOST] == "example.local"
+    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

tests/components/elgato/test_init.py

@@ -0,0 +1,33 @@
+"""Tests for the Elgato Key Light integration."""
+import aiohttp
+
+from homeassistant.components.elgato.const import DOMAIN
+from homeassistant.config_entries import ENTRY_STATE_SETUP_RETRY
+from homeassistant.core import HomeAssistant
+
+from tests.components.elgato import init_integration
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+async def test_config_entry_not_ready(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the Elgato Key Light configuration entry not ready."""
+    aioclient_mock.get(
+        "http://example.local:9123/elgato/accessory-info", exc=aiohttp.ClientError
+    )
+
+    entry = await init_integration(hass, aioclient_mock)
+    assert entry.state == ENTRY_STATE_SETUP_RETRY
+
+
+async def test_unload_config_entry(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the Elgato Key Light configuration entry unloading."""
+    entry = await init_integration(hass, aioclient_mock)
+    assert hass.data[DOMAIN]
+
+    await hass.config_entries.async_unload(entry.entry_id)
+    await hass.async_block_till_done()
+    assert not hass.data.get(DOMAIN)

tests/components/elgato/test_light.py

@@ -0,0 +1,104 @@
+"""Tests for the Elgato Key Light light platform."""
+from unittest.mock import patch
+
+from homeassistant.components.elgato.light import ElgatoError
+from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
+    ATTR_COLOR_TEMP,
+    DOMAIN as LIGHT_DOMAIN,
+)
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    SERVICE_TURN_OFF,
+    SERVICE_TURN_ON,
+    STATE_ON,
+    STATE_UNAVAILABLE,
+)
+from homeassistant.core import HomeAssistant
+
+from tests.common import mock_coro
+from tests.components.elgato import init_integration
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+async def test_light_state(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the creation and values of the Elgato Key Lights."""
+    await init_integration(hass, aioclient_mock)
+
+    entity_registry = await hass.helpers.entity_registry.async_get_registry()
+
+    # First segment of the strip
+    state = hass.states.get("light.frenck")
+    assert state
+    assert state.attributes.get(ATTR_BRIGHTNESS) == 54
+    assert state.attributes.get(ATTR_COLOR_TEMP) == 297
+    assert state.state == STATE_ON
+
+    entry = entity_registry.async_get("light.frenck")
+    assert entry
+    assert entry.unique_id == "CN11A1A00001"
+
+
+async def test_light_change_state(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the change of state of a Elgato Key Light device."""
+    await init_integration(hass, aioclient_mock)
+
+    state = hass.states.get("light.frenck")
+    assert state.state == STATE_ON
+
+    with patch(
+        "homeassistant.components.elgato.light.Elgato.light", return_value=mock_coro(),
+    ) as mock_light:
+        await hass.services.async_call(
+            LIGHT_DOMAIN,
+            SERVICE_TURN_ON,
+            {
+                ATTR_ENTITY_ID: "light.frenck",
+                ATTR_BRIGHTNESS: 255,
+                ATTR_COLOR_TEMP: 100,
+            },
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        assert len(mock_light.mock_calls) == 1
+        mock_light.assert_called_with(on=True, brightness=100, temperature=100)
+
+    with patch(
+        "homeassistant.components.elgato.light.Elgato.light", return_value=mock_coro(),
+    ) as mock_light:
+        await hass.services.async_call(
+            LIGHT_DOMAIN,
+            SERVICE_TURN_OFF,
+            {ATTR_ENTITY_ID: "light.frenck"},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        assert len(mock_light.mock_calls) == 1
+        mock_light.assert_called_with(on=False)
+
+
+async def test_light_unavailable(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test error/unavailable handling of an Elgato Key Light."""
+    await init_integration(hass, aioclient_mock)
+    with patch(
+        "homeassistant.components.elgato.light.Elgato.light", side_effect=ElgatoError,
+    ):
+        with patch(
+            "homeassistant.components.elgato.light.Elgato.state",
+            side_effect=ElgatoError,
+        ):
+            await hass.services.async_call(
+                LIGHT_DOMAIN,
+                SERVICE_TURN_OFF,
+                {ATTR_ENTITY_ID: "light.frenck"},
+                blocking=True,
+            )
+            await hass.async_block_till_done()
+            state = hass.states.get("light.frenck")
+            assert state.state == STATE_UNAVAILABLE

tests/fixtures/elgato/info.json

@@ -0,0 +1,9 @@
+{
+  "productName": "Elgato Key Light",
+  "hardwareBoardType": 53,
+  "firmwareBuildNumber": 192,
+  "firmwareVersion": "1.0.3",
+  "serialNumber": "CN11A1A00001",
+  "displayName": "Frenck",
+  "features": ["lights"]
+}

tests/fixtures/elgato/state.json

@@ -0,0 +1,10 @@
+{
+  "numberOfLights": 1,
+  "lights": [
+    {
+      "on": 1,
+      "brightness": 21,
+      "temperature": 297
+    }
+  ]
+}