Add DHCP discovery to Knocki (#144048)

* Add DHCP discovery to Knocki

* Update homeassistant/components/knocki/quality_scale.yaml

Co-authored-by: Josef Zweck <josef@zweck.dev>

---------

Co-authored-by: Josef Zweck <josef@zweck.dev>
This commit is contained in:
Joost Lekkerkerker 2025-05-02 18:34:09 +02:00 committed by GitHub
parent 5e463d6af4
commit 4967c287f8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 104 additions and 5 deletions

View File

@ -10,7 +10,9 @@ import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME from homeassistant.const import CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from .const import DOMAIN, LOGGER from .const import DOMAIN, LOGGER
@ -62,3 +64,19 @@ class KnockiConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors, errors=errors,
data_schema=DATA_SCHEMA, data_schema=DATA_SCHEMA,
) )
async def async_step_dhcp(
self, discovery_info: DhcpServiceInfo
) -> ConfigFlowResult:
"""Handle a DHCP discovery."""
device_registry = dr.async_get(self.hass)
if device_entry := device_registry.async_get_device(
identifiers={(DOMAIN, discovery_info.hostname)}
):
device_registry.async_update_device(
device_entry.id,
new_connections={
(dr.CONNECTION_NETWORK_MAC, discovery_info.macaddress)
},
)
return await super().async_step_dhcp(discovery_info)

View File

@ -3,6 +3,11 @@
"name": "Knocki", "name": "Knocki",
"codeowners": ["@joostlek", "@jgatto1", "@JakeBosh"], "codeowners": ["@joostlek", "@jgatto1", "@JakeBosh"],
"config_flow": true, "config_flow": true,
"dhcp": [
{
"hostname": "knc*"
}
],
"documentation": "https://www.home-assistant.io/integrations/knocki", "documentation": "https://www.home-assistant.io/integrations/knocki",
"integration_type": "hub", "integration_type": "hub",
"iot_class": "cloud_push", "iot_class": "cloud_push",

View File

@ -50,10 +50,8 @@ rules:
# Gold # Gold
devices: done devices: done
diagnostics: todo diagnostics: todo
discovery-update-info: discovery-update-info: done
status: exempt discovery: done
comment: This is a cloud service and does not benefit from device updates.
discovery: todo
docs-data-update: todo docs-data-update: todo
docs-examples: todo docs-examples: todo
docs-known-limitations: todo docs-known-limitations: todo

View File

@ -311,6 +311,10 @@ DHCP: Final[list[dict[str, str | bool]]] = [
"hostname": "polisy*", "hostname": "polisy*",
"macaddress": "000DB9*", "macaddress": "000DB9*",
}, },
{
"domain": "knocki",
"hostname": "knc*",
},
{ {
"domain": "lamarzocco", "domain": "lamarzocco",
"registered_devices": True, "registered_devices": True,

View File

@ -10,3 +10,4 @@ async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry)
config_entry.add_to_hass(hass) config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id) await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

View File

@ -6,13 +6,23 @@ from knocki import KnockiConnectionError, KnockiInvalidAuthError
import pytest import pytest
from homeassistant.components.knocki.const import DOMAIN from homeassistant.components.knocki.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER from homeassistant.config_entries import SOURCE_DHCP, SOURCE_USER
from homeassistant.const import CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME from homeassistant.const import CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from . import setup_integration
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
DHCP_DISCOVERY = DhcpServiceInfo(
ip="1.1.1.1",
hostname="KNC1-W-00000214",
macaddress="aa:bb:cc:dd:ee:ff",
)
async def test_full_flow( async def test_full_flow(
hass: HomeAssistant, hass: HomeAssistant,
@ -111,3 +121,66 @@ async def test_exceptions(
}, },
) )
assert result["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_dhcp(
hass: HomeAssistant,
mock_knocki_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test DHCP discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == "test-id"
async def test_dhcp_mac(
hass: HomeAssistant,
mock_knocki_client: AsyncMock,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test updating the mac address in the DHCP discovery."""
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device(identifiers={(DOMAIN, "KNC1-W-00000214")})
assert device
assert device.connections == set()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
device = device_registry.async_get_device(identifiers={(DOMAIN, "KNC1-W-00000214")})
assert device
assert device.connections == {(dr.CONNECTION_NETWORK_MAC, "aa:bb:cc:dd:ee:ff")}
async def test_dhcp_already_setup(
hass: HomeAssistant,
mock_knocki_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test DHCP discovery with already setup device."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=DHCP_DISCOVERY
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"