jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 03:07:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix Magic Home devices with multiple network interfaces (#68029)

Browse Source
This commit is contained in:
J. Nick Koston 2022-03-14 08:38:54 -10:00 committed by GitHub
parent 0e602dd390
commit 8a8d7741d5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 215 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
homeassistant/components/flux_led/__init__.py
View File

@ -41,6 +41,7 @@ from .discovery import (
async_trigger_discovery,
async_update_entry_from_discovery,
)
from .util import mac_matches_by_one
_LOGGER = logging.getLogger(__name__)
@ -90,18 +91,27 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def _async_migrate_unique_ids(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Migrate entities when the mac address gets discovered."""
if not (unique_id := entry.unique_id):
return
entry_id = entry.entry_id
@callback
def _async_migrator(entity_entry: er.RegistryEntry) -> dict[str, Any] | None:
# Old format {entry_id}.....
# New format {unique_id}....
entity_unique_id = entity_entry.unique_id
if not entity_unique_id.startswith(entry_id):
if not (unique_id := entry.unique_id):
return None
entry_id = entry.entry_id
entity_unique_id = entity_entry.unique_id
entity_mac = entity_unique_id[: len(unique_id)]
new_unique_id = None
if entity_unique_id.startswith(entry_id):
# Old format {entry_id}....., New format {unique_id}....
new_unique_id = f"{unique_id}{entity_unique_id[len(entry_id):]}"
elif (
":" in entity_mac
and entity_mac != unique_id
and mac_matches_by_one(entity_mac, unique_id)
):
# Old format {dhcp_mac}....., New format {discovery_mac}....
new_unique_id = f"{unique_id}{entity_unique_id[len(unique_id):]}"
else:
return None
new_unique_id = f"{unique_id}{entity_unique_id[len(entry_id):]}"
_LOGGER.info(
"Migrating unique_id from [%s] to [%s]",
entity_unique_id,
@ -148,7 +158,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if entry.unique_id and discovery.get(ATTR_ID):
mac = dr.format_mac(cast(str, discovery[ATTR_ID]))
if mac != entry.unique_id:
if not mac_matches_by_one(mac, entry.unique_id):
# The device is offline and another flux_led device is now using the ip address
raise ConfigEntryNotReady(
f"Unexpected device found at {host}; Expected {entry.unique_id}, found {mac}"
@ -157,7 +167,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if not discovery_cached:
# Only update the entry once we have verified the unique id
# is either missing or we have verified it matches
async_update_entry_from_discovery(hass, entry, discovery, device.model_num)
async_update_entry_from_discovery(
hass, entry, discovery, device.model_num, True
)
await _async_migrate_unique_ids(hass, entry)

57
homeassistant/components/flux_led/config_flow.py
View File

@ -19,7 +19,7 @@ from homeassistant import config_entries
from homeassistant.components import dhcp
from homeassistant.const import CONF_HOST
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import DiscoveryInfoType
@ -43,7 +43,7 @@ from .discovery import (
async_populate_data_from_discovery,
async_update_entry_from_discovery,
)
from .util import format_as_flux_mac
from .util import format_as_flux_mac, mac_matches_by_one
CONF_DEVICE: Final = "device"
@ -57,6 +57,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize the config flow."""
self._discovered_devices: dict[str, FluxLEDDiscovery] = {}
self._discovered_device: FluxLEDDiscovery | None = None
self._allow_update_mac = False
@staticmethod
@callback
@ -85,37 +86,65 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self, discovery_info: DiscoveryInfoType
) -> FlowResult:
"""Handle integration discovery."""
self._allow_update_mac = True
self._discovered_device = cast(FluxLEDDiscovery, discovery_info)
return await self._async_handle_discovery()
async def _async_set_discovered_mac(
self, device: FluxLEDDiscovery, allow_update_mac: bool
) -> None:
"""Set the discovered mac.
We only allow it to be updated if it comes from udp
discovery since the dhcp mac can be one digit off from
the udp discovery mac for devices with multiple network interfaces
"""
mac_address = device[ATTR_ID]
assert mac_address is not None
mac = dr.format_mac(mac_address)
await self.async_set_unique_id(mac)
for entry in self._async_current_entries(include_ignore=False):
if entry.data[CONF_HOST] == device[ATTR_IPADDR] or (
entry.unique_id
and ":" in entry.unique_id
and mac_matches_by_one(entry.unique_id, mac)
):
if async_update_entry_from_discovery(
self.hass, entry, device, None, allow_update_mac
):
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)
)
raise AbortFlow("already_configured")
async def _async_handle_discovery(self) -> FlowResult:
"""Handle any discovery."""
device = self._discovered_device
assert device is not None
mac_address = device[ATTR_ID]
assert mac_address is not None
mac = dr.format_mac(mac_address)
await self._async_set_discovered_mac(device, self._allow_update_mac)
host = device[ATTR_IPADDR]
await self.async_set_unique_id(mac)
for entry in self._async_current_entries(include_ignore=False):
if entry.unique_id == mac or entry.data[CONF_HOST] == host:
if async_update_entry_from_discovery(self.hass, entry, device, None):
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)
)
return self.async_abort(reason="already_configured")
self.context[CONF_HOST] = host
for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == host:
return self.async_abort(reason="already_in_progress")
if not device[ATTR_MODEL_DESCRIPTION]:
mac_address = device[ATTR_ID]
assert mac_address is not None
mac = dr.format_mac(mac_address)
try:
device = await self._async_try_connect(host, device)
except FLUX_LED_EXCEPTIONS:
return self.async_abort(reason="cannot_connect")
else:
if device[ATTR_MODEL_DESCRIPTION]:
discovered_mac = device[ATTR_ID]
if device[ATTR_MODEL_DESCRIPTION] or (
discovered_mac is not None
and (formatted_discovered_mac := dr.format_mac(discovered_mac))
and formatted_discovered_mac != mac
and mac_matches_by_one(discovered_mac, mac)
):
self._discovered_device = device
await self._async_set_discovered_mac(device, True)
return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm(

21
homeassistant/components/flux_led/discovery.py
View File

@ -19,6 +19,7 @@ from flux_led.const import (
ATTR_REMOTE_ACCESS_PORT,
ATTR_VERSION_NUM,
)
from flux_led.models_db import get_model_description
from flux_led.scanner import FluxLEDDiscovery
from homeassistant import config_entries
@ -42,7 +43,7 @@ from .const import (
DOMAIN,
FLUX_LED_DISCOVERY,
)
from .util import format_as_flux_mac
from .util import format_as_flux_mac, mac_matches_by_one
_LOGGER = logging.getLogger(__name__)
@ -80,13 +81,17 @@ def async_build_cached_discovery(entry: ConfigEntry) -> FluxLEDDiscovery:
@callback
def async_name_from_discovery(device: FluxLEDDiscovery) -> str:
def async_name_from_discovery(
device: FluxLEDDiscovery, model_num: int | None = None
) -> str:
"""Convert a flux_led discovery to a human readable name."""
if (mac_address := device[ATTR_ID]) is None:
return device[ATTR_IPADDR]
short_mac = mac_address[-6:]
if device[ATTR_MODEL_DESCRIPTION]:
return f"{device[ATTR_MODEL_DESCRIPTION]} {short_mac}"
if model_num is not None:
return f"{get_model_description(model_num, None)} {short_mac}"
return f"{device[ATTR_MODEL]} {short_mac}"
@ -113,19 +118,25 @@ def async_update_entry_from_discovery(
entry: config_entries.ConfigEntry,
device: FluxLEDDiscovery,
model_num: int | None,
allow_update_mac: bool,
) -> bool:
"""Update a config entry from a flux_led discovery."""
data_updates: dict[str, Any] = {}
mac_address = device[ATTR_ID]
assert mac_address is not None
updates: dict[str, Any] = {}
if not entry.unique_id:
updates["unique_id"] = dr.format_mac(mac_address)
formatted_mac = dr.format_mac(mac_address)
if not entry.unique_id or (
allow_update_mac
and entry.unique_id != formatted_mac
and mac_matches_by_one(formatted_mac, entry.unique_id)
):
updates["unique_id"] = formatted_mac
if model_num and entry.data.get(CONF_MODEL_NUM) != model_num:
data_updates[CONF_MODEL_NUM] = model_num
async_populate_data_from_discovery(entry.data, data_updates, device)
if is_ip_address(entry.title):
updates["title"] = async_name_from_discovery(device)
updates["title"] = async_name_from_discovery(device, model_num)
title_matches_name = entry.title == entry.data.get(CONF_NAME)
if data_updates or title_matches_name:
updates["data"] = {**entry.data, **data_updates}

12
homeassistant/components/flux_led/util.py
View File

@ -28,6 +28,18 @@ def _human_readable_option(const_option: str) -> str:
return const_option.replace("_", " ").title()
def mac_matches_by_one(formatted_mac_1: str, formatted_mac_2: str) -> bool:
"""Check if a mac address is only one digit off.
Some of the devices have two mac addresses which are
one off from each other. We need to treat them as the same
since its the same device.
"""
mac_int_1 = int(formatted_mac_1.replace(":", ""), 16)
mac_int_2 = int(formatted_mac_2.replace(":", ""), 16)
return abs(mac_int_1 - mac_int_2) < 2
def _flux_color_mode_to_hass(
flux_color_mode: str | None, flux_color_modes: set[str]
) -> str:

2
tests/components/flux_led/__init__.py
View File

@ -39,6 +39,8 @@ MODEL_NUM = 0x35
MODEL = "AK001-ZJ2149"
MODEL_DESCRIPTION = "Bulb RGBCW"
MAC_ADDRESS = "aa:bb:cc:dd:ee:ff"
MAC_ADDRESS_ONE_OFF = "aa:bb:cc:dd:ee:fe"
FLUX_MAC_ADDRESS = "AABBCCDDEEFF"
SHORT_MAC_ADDRESS = "DDEEFF"

42
tests/components/flux_led/test_config_flow.py
View File

@ -34,6 +34,7 @@ from . import (
FLUX_DISCOVERY_PARTIAL,
IP_ADDRESS,
MAC_ADDRESS,
MAC_ADDRESS_ONE_OFF,
MODEL,
MODEL_DESCRIPTION,
MODEL_NUM,
@ -546,6 +547,7 @@ async def test_discovered_by_dhcp_partial_udp_response_fallback_tcp(hass):
CONF_MODEL_NUM: MODEL_NUM,
CONF_MODEL_DESCRIPTION: MODEL_DESCRIPTION,
}
assert result2["title"] == "Bulb RGBCW DDEEFF"
assert mock_async_setup.called
assert mock_async_setup_entry.called
@ -589,6 +591,46 @@ async def test_discovered_by_dhcp_or_discovery_adds_missing_unique_id(
assert config_entry.unique_id == MAC_ADDRESS
async def test_mac_address_off_by_one_updated_via_discovery(hass):
"""Test the mac address is updated when its off by one from integration discovery."""
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=MAC_ADDRESS_ONE_OFF
)
config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_wifibulb():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data=FLUX_DISCOVERY,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert config_entry.unique_id == MAC_ADDRESS
async def test_mac_address_off_by_one_not_updated_from_dhcp(hass):
"""Test the mac address is NOT updated when its off by one from dhcp discovery."""
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=MAC_ADDRESS_ONE_OFF
)
config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_wifibulb():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=DHCP_DISCOVERY
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert config_entry.unique_id == MAC_ADDRESS_ONE_OFF
@pytest.mark.parametrize(
"source, data",
[

82
tests/components/flux_led/test_init.py
View File

@ -26,6 +26,7 @@ from . import (
FLUX_DISCOVERY_PARTIAL,
IP_ADDRESS,
MAC_ADDRESS,
MAC_ADDRESS_ONE_OFF,
_mocked_bulb,
_patch_discovery,
_patch_wifibulb,
@ -82,7 +83,9 @@ async def test_configuring_flux_led_causes_discovery_multiple_addresses(
async def test_config_entry_reload(hass: HomeAssistant) -> None:
"""Test that a config entry can be reloaded."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=MAC_ADDRESS)
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=MAC_ADDRESS
)
config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_wifibulb():
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
@ -126,11 +129,11 @@ async def test_config_entry_fills_unique_id_with_directed_discovery(
# Only return discovery results when doing directed discovery
nonlocal last_address
last_address = address
return [FLUX_DISCOVERY] if address == IP_ADDRESS else []
return [discovery] if address == IP_ADDRESS else []
def _mock_getBulbInfo(*args, **kwargs):
nonlocal last_address
return [FLUX_DISCOVERY] if last_address == IP_ADDRESS else []
return [discovery] if last_address == IP_ADDRESS else []
with patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.async_scan",
@ -149,7 +152,9 @@ async def test_config_entry_fills_unique_id_with_directed_discovery(
async def test_time_sync_startup_and_next_day(hass: HomeAssistant) -> None:
"""Test that time is synced on startup and next day."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=MAC_ADDRESS)
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=MAC_ADDRESS
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
with _patch_discovery(), _patch_wifibulb(device=bulb):
@ -204,3 +209,72 @@ async def test_unique_id_migrate_when_mac_discovered(hass: HomeAssistant) -> Non
entity_registry.async_get("switch.bulb_rgbcw_ddeeff_remote_access").unique_id
== f"{config_entry.unique_id}_remote_access"
)
async def test_unique_id_migrate_when_mac_discovered_via_discovery(
hass: HomeAssistant,
) -> None:
"""Test unique id migrated when mac discovered via discovery and the mac address from dhcp was one off."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_REMOTE_ACCESS_HOST: "any",
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_PORT: 1234,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
},
unique_id=MAC_ADDRESS_ONE_OFF,
)
config_entry.add_to_hass(hass)
bulb = _mocked_bulb()
with _patch_discovery(no_device=True), _patch_wifibulb(device=bulb):
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
assert config_entry.unique_id == MAC_ADDRESS_ONE_OFF
entity_registry = er.async_get(hass)
assert (
entity_registry.async_get("light.bulb_rgbcw_ddeeff").unique_id
== MAC_ADDRESS_ONE_OFF
)
assert (
entity_registry.async_get("switch.bulb_rgbcw_ddeeff_remote_access").unique_id
== f"{MAC_ADDRESS_ONE_OFF}_remote_access"
)
for _ in range(2):
with _patch_discovery(), _patch_wifibulb(device=bulb):
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.async_block_till_done()
assert (
entity_registry.async_get("light.bulb_rgbcw_ddeeff").unique_id
== config_entry.unique_id
)
assert (
entity_registry.async_get(
"switch.bulb_rgbcw_ddeeff_remote_access"
).unique_id
== f"{config_entry.unique_id}_remote_access"
)
async def test_name_removed_when_it_matches_entry_title(hass: HomeAssistant) -> None:
"""Test name is removed when it matches the entry title."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_REMOTE_ACCESS_HOST: "any",
CONF_REMOTE_ACCESS_ENABLED: True,
CONF_REMOTE_ACCESS_PORT: 1234,
CONF_HOST: IP_ADDRESS,
CONF_NAME: DEFAULT_ENTRY_TITLE,
},
title=DEFAULT_ENTRY_TITLE,
)
config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_wifibulb():
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
assert CONF_NAME not in config_entry.data