starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-25 07:51:46 -05:00
Code Issues Packages Projects Releases Wiki Activity

Match UniFi discovery against every address a console announces (#180343)

Browse Source
This commit is contained in:
Raphael Hehl
2026-08-27 20:51:03 +02:00
committed by GitHub
parent e81abcf8cd
commit c0e009933a
6 changed files with 216 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifi/config_flow.py
+11 -4
View File
@@ -228,13 +228,20 @@ class UnifiFlowHandler(ConfigFlow, domain=DOMAIN):
CONF_VERIFY_SSL: bool(direct_connect_domain),
}
for entry in self._async_current_entries(include_ignore=False):
if entry.data.get(CONF_HOST) in (source_ip, direct_connect_domain):
return self.async_abort(reason="already_configured")
# MAC first: an entry keyed by it gets its host refreshed here, and the
# host match below would otherwise abort before that can happen.
await self.async_set_unique_id(mac_address)
self._abort_if_unique_id_configured(updates=self.config, reload_on_update=False)
# A console answers on every VLAN interface but discovery reports only
# one of them, so match every address it announced for itself.
known_hosts = {source_ip, *discovery_info.get("announced_ips", ())}
if direct_connect_domain:
known_hosts.add(direct_connect_domain)
for entry in self._async_current_entries(include_ignore=False):
if entry.data.get(CONF_HOST) in known_hosts:
return self.async_abort(reason="already_configured")
self.context["title_placeholders"] = {
CONF_NAME: (
discovery_info.get("name")
homeassistant/components/unifi_access/config_flow.py
+4 -1
View File
@@ -131,10 +131,13 @@ class UnifiAccessConfigFlow(ConfigFlow, domain=DOMAIN):
source_ip = discovery_info["source_ip"]
mac = discovery_info["hw_addr"].replace(":", "").upper()
await self.async_set_unique_id(mac)
# A console answers on every VLAN interface but discovery reports only
# one of them, so match every address it announced for itself.
known_hosts = {source_ip, *discovery_info.get("announced_ips", ())}
for entry in self._async_current_entries():
if entry.source == SOURCE_IGNORE:
continue
if entry.data.get(CONF_HOST) == source_ip:
if entry.data.get(CONF_HOST) in known_hosts:
if not entry.unique_id:
self.hass.config_entries.async_update_entry(entry, unique_id=mac)
return self.async_abort(reason="already_configured")
homeassistant/components/unifi_discovery/discovery.py
+30
View File
@@ -11,6 +11,7 @@ from unifi_discovery import AIOUnifiScanner, UnifiDevice
from homeassistant import config_entries
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util.hass_dict import HassKey
@@ -23,6 +24,34 @@ DISCOVERY_INTERVAL = timedelta(minutes=60)
DATA_DISCOVERY_STARTED: HassKey[bool] = HassKey(DOMAIN)
def _announced_ips(device: UnifiDevice) -> list[str]:
"""Return the IPs a device announced as its own.
A console answers discovery on every VLAN interface it has and lists them
in ``ip_info`` as ``"mac;ip"``, alongside ``primary_addr``. Only one of
those answers survives the scanner's per-MAC collapse, so consumers cannot
rely on ``source_ip`` being the address an entry was configured with.
``ip_info`` also carries addresses that are not the device's own: the
upstream WAN, neighbouring hosts and all-zero placeholders. A device's
interface MACs share the first five octets with ``hw_addr``, which is what
separates them. Matching on the OUI alone would pull in every other
Ubiquiti device on the network.
"""
if not device.hw_addr:
return []
prefix = format_mac(device.hw_addr)[:14]
announced = [*(device.ip_info or ())]
if device.primary_addr:
announced.append(device.primary_addr)
ips: list[str] = []
for address in announced:
mac_address, _, ip_address = address.rpartition(";")
if ip_address and format_mac(mac_address).startswith(prefix):
ips.append(ip_address)
return list(dict.fromkeys(ips))
def _device_to_dict(device: UnifiDevice) -> dict[str, Any]:
"""Convert a UnifiDevice to a plain dict.
@@ -38,6 +67,7 @@ def _device_to_dict(device: UnifiDevice) -> dict[str, Any]:
if isinstance(value, Mapping):
value = dict(value)
data[f.name] = value
data["announced_ips"] = _announced_ips(device)
return data
tests/components/unifi/test_config_flow.py
+55 -3
View File
@@ -1,6 +1,7 @@
"""Test UniFi Network config flow."""
import socket
from typing import Any
from unittest.mock import PropertyMock, patch
import pytest
@@ -541,6 +542,40 @@ async def test_flow_integration_discovery_aborts_if_host_already_exists(
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("config_entry")
async def test_flow_integration_discovery_aborts_on_other_announced_address(
hass: HomeAssistant,
) -> None:
"""Test we abort when the entry uses another interface of the same console."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
**INTEGRATION_DISCOVERY_INFO,
"source_ip": "10.0.0.1",
"direct_connect_domain": None,
"announced_ips": ["10.0.0.1", "1.2.3.4"],
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_flow_integration_discovery_ignores_entry_without_host(
hass: HomeAssistant,
) -> None:
"""Test an entry carrying no host does not match a missing direct connect."""
MockConfigEntry(domain=DOMAIN, unique_id="site-id", data={}).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={**INTEGRATION_DISCOVERY_INFO, "direct_connect_domain": None},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
async def test_flow_integration_discovery_uses_direct_connect_domain(
hass: HomeAssistant,
) -> None:
@@ -580,15 +615,32 @@ async def test_flow_integration_discovery_aborts_on_direct_connect_host(
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("entry_host", "extra_info"),
[
pytest.param("old.host", {}, id="stale_host"),
pytest.param(
"10.0.0.99",
{"announced_ips": ["10.0.0.99"]},
id="other_announced_interface",
),
],
)
async def test_flow_integration_discovery_updates_existing_entry_on_rediscovery(
hass: HomeAssistant,
entry_host: str,
extra_info: dict[str, Any],
) -> None:
"""Test existing entry's host is refreshed when rediscovered with same MAC."""
"""Test existing entry's host is refreshed when rediscovered with same MAC.
This also holds when the entry sits on another interface the console
announces, which the host match must not abort on first.
"""
old_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=format_mac(INTEGRATION_DISCOVERY_INFO["hw_addr"]),
data={
CONF_HOST: "old.host",
CONF_HOST: entry_host,
CONF_VERIFY_SSL: False,
},
)
@@ -597,7 +649,7 @@ async def test_flow_integration_discovery_updates_existing_entry_on_rediscovery(
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data=INTEGRATION_DISCOVERY_INFO,
data={**INTEGRATION_DISCOVERY_INFO, **extra_info},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
tests/components/unifi_access/test_config_flow.py
+31
View File
@@ -749,6 +749,37 @@ async def test_discovery_sets_unique_id_on_manual_entry(
assert entry.unique_id == "AABBCCDDEEFF"
async def test_discovery_matches_other_announced_address(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
"""Test an entry on another interface of the same console is recognised.
A console answers on every VLAN interface but discovery reports only one of
them, so a manually configured entry on another one has to match too, which
is also what stamps its unique ID.
"""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "192.168.2.5",
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_VERIFY_SSL: False,
},
)
entry.add_to_hass(hass)
assert entry.unique_id is None
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_INTEGRATION_DISCOVERY},
data={**DISCOVERY_INFO, "announced_ips": ["10.0.0.5", "192.168.2.5"]},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.unique_id == "AABBCCDDEEFF"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_discovery_already_configured_by_host_with_unique_id(
hass: HomeAssistant, mock_client: MagicMock
tests/components/unifi_discovery/test_init.py
+85
View File
@@ -1,7 +1,16 @@
"""Test the UniFi Discovery init."""
from typing import Any
import pytest
from unifi_discovery import UnifiDevice
from homeassistant import config_entries
from homeassistant.components.unifi_discovery.const import DOMAIN
from homeassistant.components.unifi_discovery.discovery import (
_announced_ips,
_device_to_dict,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@@ -73,3 +82,79 @@ async def test_discovery_does_not_deepcopy_device(hass: HomeAssistant) -> None:
flows = hass.config_entries.flow.async_progress_by_handler("unifiprotect")
assert len(flows) == 1
assert flows[0]["context"]["source"] == config_entries.SOURCE_INTEGRATION_DISCOVERY
@pytest.mark.parametrize(
("device_kwargs", "expected"),
[
pytest.param({}, [], id="nothing_announced"),
pytest.param(
{
"ip_info": (
"aa:bb:cc:dd:ee:fd;192.168.1.1",
"aa:bb:cc:dd:ee:fd;192.168.2.1",
"aa:bb:cc:dd:ee:fb;10.0.0.5",
)
},
["192.168.1.1", "192.168.2.1", "10.0.0.5"],
id="every_interface",
),
pytest.param(
{
"ip_info": (
"aa:bb:cc:dd:ee:fd;192.168.1.1",
# Upstream WAN and neighbours the console also reports.
"00:00:00:00:00:00;198.51.100.7",
"00:00:00:00:00:00;192.168.0.0",
"5a:71:71:7a:68:8e;192.168.1.9",
# Another Ubiquiti device: same OUI, different unit.
"aa:bb:cc:99:99:01;192.168.1.10",
)
},
["192.168.1.1"],
id="foreign_addresses_dropped",
),
pytest.param(
{"ip_info": ("aa:bb:cc:dd:ee:fd;", "aa:bb:cc:dd:ee:fd")},
[],
id="malformed_entries",
),
pytest.param(
{"primary_addr": "aa:bb:cc:dd:ee:ff;192.168.1.1"},
["192.168.1.1"],
id="primary_addr",
),
pytest.param(
{
"ip_info": ("aa:bb:cc:dd:ee:fd;192.168.1.1",),
"primary_addr": "aa:bb:cc:dd:ee:ff;192.168.1.1",
},
["192.168.1.1"],
id="deduplicated",
),
],
)
def test_announced_ips(device_kwargs: dict[str, Any], expected: list[str]) -> None:
"""Test only the device's own announced addresses are returned."""
device = UnifiDevice(
source_ip="192.168.1.1", hw_addr="aa:bb:cc:dd:ee:ff", **device_kwargs
)
assert _announced_ips(device) == expected
def test_announced_ips_without_mac() -> None:
"""Test a device without hw_addr announces nothing identifiable."""
device = UnifiDevice(
source_ip="192.168.1.1", ip_info=("aa:bb:cc:dd:ee:fd;192.168.1.1",)
)
assert _announced_ips(device) == []
def test_device_to_dict_carries_announced_ips() -> None:
"""Test the announced addresses reach the payload the consumers receive."""
device = UnifiDevice(
source_ip="192.168.1.1",
hw_addr="aa:bb:cc:dd:ee:ff",
ip_info=("aa:bb:cc:dd:ee:fd;192.168.2.1",),
)
assert _device_to_dict(device)["announced_ips"] == ["192.168.2.1"]