Add twinkly DHCP support (#61434)

* Add twinkly DHCP support

* fix typing import

* fix format

* Fix imports v2

* Using IP

* Fix tests

* Apply suggestions from code review

Thanks @bdraco

Co-authored-by: J. Nick Koston <nick@koston.org>

* fix black

* Add confirm step

* Add more tests

* Update homeassistant/components/twinkly/config_flow.py

Co-authored-by: J. Nick Koston <nick@koston.org>

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/twinkly/config_flow.py

@@ -1,13 +1,16 @@
 """Config flow to configure the Twinkly integration."""
+from __future__ import annotations
 
 import asyncio
 import logging
+from typing import Any
 
 from aiohttp import ClientError
 import twinkly_client
 from voluptuous import Required, Schema
 
-from homeassistant import config_entries
+from homeassistant import config_entries, data_entry_flow
+from homeassistant.components import dhcp
 from homeassistant.const import CONF_HOST
 
 from .const import (
@@ -29,6 +32,10 @@ class TwinklyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
 
     VERSION = 1
 
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._discovered_device: tuple[dict[str, Any], str] | None = None
+
     async def async_step_user(self, user_input=None):
         """Handle config steps."""
         host = user_input[CONF_HOST] if user_input else None
@@ -43,15 +50,8 @@ class TwinklyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
                 await self.async_set_unique_id(device_info[DEV_ID])
                 self._abort_if_unique_id_configured()
 
-                return self.async_create_entry(
-                    title=device_info[DEV_NAME],
-                    data={
-                        CONF_ENTRY_HOST: host,
-                        CONF_ENTRY_ID: device_info[DEV_ID],
-                        CONF_ENTRY_NAME: device_info[DEV_NAME],
-                        CONF_ENTRY_MODEL: device_info[DEV_MODEL],
-                    },
-                )
+                return self._create_entry_from_device(device_info, host)
+
             except (asyncio.TimeoutError, ClientError) as err:
                 _LOGGER.info("Cannot reach Twinkly '%s' (client)", host, exc_info=err)
                 errors[CONF_HOST] = "cannot_connect"
@@ -59,3 +59,53 @@ class TwinklyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         return self.async_show_form(
             step_id="user", data_schema=Schema(schema), errors=errors
         )
+
+    async def async_step_dhcp(
+        self, discovery_info: dhcp.DhcpServiceInfo
+    ) -> data_entry_flow.FlowResult:
+        """Handle dhcp discovery for twinkly."""
+        self._async_abort_entries_match({CONF_ENTRY_HOST: discovery_info.ip})
+        device_info = await twinkly_client.TwinklyClient(
+            discovery_info.ip
+        ).get_device_info()
+        await self.async_set_unique_id(device_info[DEV_ID])
+        self._abort_if_unique_id_configured(
+            updates={CONF_ENTRY_HOST: discovery_info.ip}
+        )
+
+        self._discovered_device = (device_info, discovery_info.ip)
+        return await self.async_step_discovery_confirm()
+
+    async def async_step_discovery_confirm(
+        self, user_input=None
+    ) -> data_entry_flow.FlowResult:
+        """Confirm discovery."""
+        assert self._discovered_device is not None
+        device_info, host = self._discovered_device
+
+        if user_input is not None:
+            return self._create_entry_from_device(device_info, host)
+
+        self._set_confirm_only()
+        placeholders = {
+            "model": device_info[DEV_MODEL],
+            "name": device_info[DEV_NAME],
+            "host": host,
+        }
+        return self.async_show_form(
+            step_id="discovery_confirm", description_placeholders=placeholders
+        )
+
+    def _create_entry_from_device(
+        self, device_info: dict[str, Any], host: str
+    ) -> data_entry_flow.FlowResult:
+        """Create entry from device data."""
+        return self.async_create_entry(
+            title=device_info[DEV_NAME],
+            data={
+                CONF_ENTRY_HOST: host,
+                CONF_ENTRY_ID: device_info[DEV_ID],
+                CONF_ENTRY_NAME: device_info[DEV_NAME],
+                CONF_ENTRY_MODEL: device_info[DEV_MODEL],
+            },
+        )

homeassistant/components/twinkly/manifest.json

@@ -6,5 +6,6 @@
   "dependencies": [],
   "codeowners": ["@dr1rrb"],
   "config_flow": true,
+  "dhcp": [{ "hostname": "twinkly_*" }],
   "iot_class": "local_polling"
 }

homeassistant/components/twinkly/strings.json

@@ -7,6 +7,9 @@
         "data": {
           "host": "Host (or IP address) of your twinkly device"
         }
+      },
+      "discovery_confirm": {
+        "description": "Do you want to setup {name} - {model} ({host})?"
       }
     },
     "error": {

homeassistant/generated/dhcp.py

@@ -520,6 +520,10 @@ DHCP = [
         "domain": "tuya",
         "macaddress": "D81F12*"
     },
+    {
+        "domain": "twinkly",
+        "hostname": "twinkly_*"
+    },
     {
         "domain": "verisure",
         "macaddress": "0023C1*"

tests/components/twinkly/test_config_flow.py

@@ -1,8 +1,8 @@
 """Tests for the config_flow of the twinly component."""
-
 from unittest.mock import patch
 
 from homeassistant import config_entries
+from homeassistant.components import dhcp
 from homeassistant.components.twinkly.const import (
     CONF_ENTRY_HOST,
     CONF_ENTRY_ID,
@@ -13,6 +13,8 @@ from homeassistant.components.twinkly.const import (
 
 from . import TEST_MODEL, ClientMock
 
+from tests.common import MockConfigEntry
+
 
 async def test_invalid_host(hass):
     """Test the failure when invalid host provided."""
@@ -60,3 +62,84 @@ async def test_success_flow(hass):
         CONF_ENTRY_NAME: client.id,
         CONF_ENTRY_MODEL: TEST_MODEL,
     }
+
+
+async def test_dhcp_can_confirm(hass):
+    """Test DHCP discovery flow can confirm right away."""
+    client = ClientMock()
+    with patch("twinkly_client.TwinklyClient", return_value=client):
+        result = await hass.config_entries.flow.async_init(
+            TWINKLY_DOMAIN,
+            context={"source": config_entries.SOURCE_DHCP},
+            data=dhcp.DhcpServiceInfo(
+                hostname="Twinkly_XYZ",
+                ip="1.2.3.4",
+                macaddress="aa:bb:cc:dd:ee:ff",
+            ),
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == "form"
+    assert result["step_id"] == "discovery_confirm"
+
+
+async def test_dhcp_success(hass):
+    """Test DHCP discovery flow success."""
+    client = ClientMock()
+    with patch("twinkly_client.TwinklyClient", return_value=client):
+        result = await hass.config_entries.flow.async_init(
+            TWINKLY_DOMAIN,
+            context={"source": config_entries.SOURCE_DHCP},
+            data=dhcp.DhcpServiceInfo(
+                hostname="Twinkly_XYZ",
+                ip="1.2.3.4",
+                macaddress="aa:bb:cc:dd:ee:ff",
+            ),
+        )
+        await hass.async_block_till_done()
+
+        assert result["type"] == "form"
+        assert result["step_id"] == "discovery_confirm"
+
+        result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == client.id
+    assert result["data"] == {
+        CONF_ENTRY_HOST: "1.2.3.4",
+        CONF_ENTRY_ID: client.id,
+        CONF_ENTRY_NAME: client.id,
+        CONF_ENTRY_MODEL: TEST_MODEL,
+    }
+
+
+async def test_dhcp_already_exists(hass):
+    """Test DHCP discovery flow that fails to connect."""
+    client = ClientMock()
+
+    entry = MockConfigEntry(
+        domain=TWINKLY_DOMAIN,
+        data={
+            CONF_ENTRY_HOST: "1.2.3.4",
+            CONF_ENTRY_ID: client.id,
+            CONF_ENTRY_NAME: client.id,
+            CONF_ENTRY_MODEL: TEST_MODEL,
+        },
+        unique_id=client.id,
+    )
+    entry.add_to_hass(hass)
+
+    with patch("twinkly_client.TwinklyClient", return_value=client):
+        result = await hass.config_entries.flow.async_init(
+            TWINKLY_DOMAIN,
+            context={"source": config_entries.SOURCE_DHCP},
+            data=dhcp.DhcpServiceInfo(
+                hostname="Twinkly_XYZ",
+                ip="1.2.3.4",
+                macaddress="aa:bb:cc:dd:ee:ff",
+            ),
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == "abort"
+    assert result["reason"] == "already_configured"