home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2026-05-05 20:34:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

Ignore discovery for existing ZHA entries (#152984)

Browse Source
This commit is contained in:
puddly
2025-09-26 01:17:01 -04:00
committed by Franck Nijhof
parent 68c51dc7aa
commit 99a0380ec5
2 changed files with 115 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/zha/config_flow.py
+35 -14
View File
@@ -23,6 +23,7 @@ from homeassistant.components.homeassistant_hardware import silabs_multiprotocol
from homeassistant.components.homeassistant_yellow import hardware as yellow_hardware
from homeassistant.config_entries import (
SOURCE_IGNORE,
SOURCE_ZEROCONF,
ConfigEntry,
ConfigEntryBaseFlow,
ConfigEntryState,
@@ -183,27 +184,17 @@ class BaseZhaFlow(ConfigEntryBaseFlow):
self._hass = hass
self._radio_mgr.hass = hass
async def _get_config_entry_data(self) -> dict:
def _get_config_entry_data(self) -> dict[str, Any]:
"""Extract ZHA config entry data from the radio manager."""
assert self._radio_mgr.radio_type is not None
assert self._radio_mgr.device_path is not None
assert self._radio_mgr.device_settings is not None
try:
device_path = await self.hass.async_add_executor_job(
usb.get_serial_by_id, self._radio_mgr.device_path
)
except OSError as error:
raise AbortFlow(
reason="cannot_resolve_path",
description_placeholders={"path": self._radio_mgr.device_path},
) from error
return {
CONF_DEVICE: DEVICE_SCHEMA(
{
**self._radio_mgr.device_settings,
CONF_DEVICE_PATH: device_path,
CONF_DEVICE_PATH: self._radio_mgr.device_path,
}
),
CONF_RADIO_TYPE: self._radio_mgr.radio_type.name,
@@ -703,6 +694,36 @@ class ZhaConfigFlowHandler(BaseZhaFlow, ConfigFlow, domain=DOMAIN):
DOMAIN, include_ignore=False
)
if self._radio_mgr.device_path is not None:
# Ensure the radio manager device path is unique and will match ZHA's
try:
self._radio_mgr.device_path = await self.hass.async_add_executor_job(
usb.get_serial_by_id, self._radio_mgr.device_path
)
except OSError as error:
raise AbortFlow(
reason="cannot_resolve_path",
description_placeholders={"path": self._radio_mgr.device_path},
) from error
# mDNS discovery can advertise the same adapter on multiple IPs or via a
# hostname, which should be considered a duplicate
current_device_paths = {self._radio_mgr.device_path}
if self.source == SOURCE_ZEROCONF:
discovery_info = self.init_data
current_device_paths |= {
f"socket://{ip}:{discovery_info.port}"
for ip in discovery_info.ip_addresses
}
for entry in zha_config_entries:
path = entry.data.get(CONF_DEVICE, {}).get(CONF_DEVICE_PATH)
# Abort discovery if the device path is already configured
if path is not None and path in current_device_paths:
return self.async_abort(reason="single_instance_allowed")
# Without confirmation, discovery can automatically progress into parts of the
# config flow logic that interacts with hardware.
if user_input is not None or (
@@ -873,7 +894,7 @@ class ZhaConfigFlowHandler(BaseZhaFlow, ConfigFlow, domain=DOMAIN):
zha_config_entries = self.hass.config_entries.async_entries(
DOMAIN, include_ignore=False
)
data = await self._get_config_entry_data()
data = self._get_config_entry_data()
if len(zha_config_entries) == 1:
return self.async_update_reload_and_abort(
@@ -976,7 +997,7 @@ class ZhaOptionsFlowHandler(BaseZhaFlow, OptionsFlow):
# Avoid creating both `.options` and `.data` by directly writing `data` here
self.hass.config_entries.async_update_entry(
entry=self.config_entry,
data=await self._get_config_entry_data(),
data=self._get_config_entry_data(),
options=self.config_entry.options,
)
tests/components/zha/test_config_flow.py
+80 -19
View File
@@ -857,6 +857,40 @@ async def test_discovery_via_usb_zha_ignored_updates(hass: HomeAssistant) -> Non
}
async def test_discovery_via_usb_same_device_already_setup(hass: HomeAssistant) -> None:
"""Test discovery aborting if ZHA is already setup."""
MockConfigEntry(
domain=DOMAIN,
data={CONF_DEVICE: {CONF_DEVICE_PATH: "/dev/serial/by-id/usb-device123"}},
).add_to_hass(hass)
# Discovery info with the same device but different path format
discovery_info = UsbServiceInfo(
device="/dev/ttyUSB0",
pid="AAAA",
vid="AAAA",
serial_number="1234",
description="zigbee radio",
manufacturer="test",
)
with patch(
"homeassistant.components.zha.config_flow.usb.get_serial_by_id",
return_value="/dev/serial/by-id/usb-device123",
) as mock_get_serial_by_id:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USB}, data=discovery_info
)
await hass.async_block_till_done()
# Verify get_serial_by_id was called to normalize the path
assert mock_get_serial_by_id.mock_calls == [call("/dev/ttyUSB0")]
# Should abort since it's the same device
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True))
async def test_legacy_zeroconf_discovery_already_setup(hass: HomeAssistant) -> None:
@@ -890,6 +924,39 @@ async def test_legacy_zeroconf_discovery_already_setup(hass: HomeAssistant) -> N
assert confirm_result["step_id"] == "choose_migration_strategy"
async def test_zeroconf_discovery_via_socket_already_setup_with_ip_match(
hass: HomeAssistant,
) -> None:
"""Test zeroconf discovery aborting when ZHA is already setup with socket and one IP matches."""
MockConfigEntry(
domain=DOMAIN,
data={CONF_DEVICE: {CONF_DEVICE_PATH: "socket://192.168.1.101:6638"}},
).add_to_hass(hass)
service_info = ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.100"),
ip_addresses=[
ip_address("192.168.1.100"),
ip_address("192.168.1.101"), # Matches config entry
],
hostname="tube-zigbee-gw.local.",
name="mock_name",
port=6638,
properties={"name": "tube_123456"},
type="mock_type",
)
# Discovery should abort due to single instance check
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=service_info
)
await hass.async_block_till_done()
# Should abort since one of the advertised IPs matches existing socket path
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
@patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type",
mock_detect_radio_type(radio_type=RadioType.deconz),
@@ -2289,34 +2356,28 @@ async def test_config_flow_serial_resolution_oserror(
) -> None:
"""Test that OSError during serial port resolution is handled."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "manual_pick_radio_type"},
data={CONF_RADIO_TYPE: RadioType.ezsp.description},
discovery_info = UsbServiceInfo(
device="/dev/ttyZIGBEE",
pid="AAAA",
vid="AAAA",
serial_number="1234",
description="zigbee radio",
manufacturer="test",
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={zigpy.config.CONF_DEVICE_PATH: "/dev/ttyUSB33"},
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "choose_setup_strategy"
with (
patch(
"homeassistant.components.usb.get_serial_by_id",
"homeassistant.components.zha.config_flow.usb.get_serial_by_id",
side_effect=OSError("Test error"),
),
):
setup_result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": config_flow.SETUP_STRATEGY_RECOMMENDED},
result_init = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USB}, data=discovery_info
)
assert setup_result["type"] is FlowResultType.ABORT
assert setup_result["reason"] == "cannot_resolve_path"
assert setup_result["description_placeholders"] == {"path": "/dev/ttyUSB33"}
assert result_init["type"] is FlowResultType.ABORT
assert result_init["reason"] == "cannot_resolve_path"
assert result_init["description_placeholders"] == {"path": "/dev/ttyZIGBEE"}
@patch("homeassistant.components.zha.radio_manager._allow_overwrite_ezsp_ieee")