starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-06 14:29:21 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix orphaned HomeWizard external devices after unique ID migration (#172657)

Browse Source
This commit is contained in:
Matthias de Baat
2026-10-02 20:01:36 +02:00
committed by GitHub
parent a57de59dad
commit 076918139c
2 changed files with 171 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/homewizard/sensor.py
+58 -2
View File
@@ -3,6 +3,7 @@
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
from typing import Final, override
from homewizard_energy.const import Model
@@ -30,6 +31,7 @@ from homeassistant.const import (
UnitOfVolumeFlowRate,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import (
DeviceInfo,
async_get_device_id_by_identifier,
@@ -43,6 +45,7 @@ from .coordinator import HomeWizardConfigEntry, HWEnergyDeviceUpdateCoordinator
from .entity import HomeWizardEntity
PARALLEL_UPDATES = 1
_LOGGER = logging.getLogger(__name__)
@dataclass(frozen=True, kw_only=True)
@@ -774,20 +777,73 @@ async def async_setup_entry(
# Initialize external devices
measurement = entry.runtime_data.data.measurement
if measurement.external_devices is not None:
for unique_id, device in measurement.external_devices.items():
dev_reg = dr.async_get(hass)
parent_serial = entry.runtime_data.data.device.serial
# This cleanup must run at setup because the external device mapping
# needed for old->new identifier conversion is only available in runtime
# measurement data.
for new_unique_id, device in measurement.external_devices.items():
_async_migrate_external_device_identifier(
dev_reg,
entry,
str(device.unique_id),
new_unique_id,
parent_serial,
)
if device.type is not None and (
description := EXTERNAL_SENSORS.get(device.type)
):
# Add external device
entities.append(
HomeWizardExternalSensorEntity(
entry.runtime_data, description, unique_id
entry.runtime_data, description, new_unique_id
)
)
async_add_entities(entities)
def _async_migrate_external_device_identifier(
dev_reg: dr.DeviceRegistry,
entry: HomeWizardConfigEntry,
old_unique_id: str,
new_unique_id: str,
parent_serial: str | None,
) -> None:
"""Migrate a HomeWizard external device identifier when needed."""
old_device = dev_reg.async_get_device_by_identifier(
(DOMAIN, old_unique_id), entry.entry_id
)
if old_device is None:
return
if old_unique_id == parent_serial:
return
new_device = dev_reg.async_get_device_by_identifier(
(DOMAIN, new_unique_id), entry.entry_id
)
if new_device is None:
dev_reg.async_update_device(
old_device.id,
new_identifiers={(DOMAIN, new_unique_id)},
serial_number=new_unique_id,
)
return
if old_device.id == new_device.id:
return
_LOGGER.debug(
"Removing migrated HomeWizard external device %s in favor of %s",
old_unique_id,
new_unique_id,
)
dev_reg.async_remove_device(old_device.id)
class HomeWizardSensorEntity(HomeWizardEntity, SensorEntity):
"""Representation of a HomeWizard Sensor."""
tests/components/homewizard/test_sensor.py
+113
View File
@@ -19,6 +19,26 @@ pytestmark = [
pytest.mark.usefixtures("init_integration"),
]
EXTERNAL_DEVICE_OLD_UNIQUE_ID = "G001"
EXTERNAL_DEVICE_NEW_UNIQUE_ID = f"gas_meter_{EXTERNAL_DEVICE_OLD_UNIQUE_ID}"
def _cleanup_external_migration_devices(
device_registry: dr.DeviceRegistry, config_entry_id: str
) -> None:
"""Remove external devices used by migration tests."""
assert (
device_registry.async_get_device_by_identifier(
(DOMAIN, EXTERNAL_DEVICE_OLD_UNIQUE_ID), config_entry_id
)
is None
)
existing_new = device_registry.async_get_device_by_identifier(
(DOMAIN, EXTERNAL_DEVICE_NEW_UNIQUE_ID), config_entry_id
)
assert existing_new is not None
device_registry.async_remove_device(existing_new.id)
@pytest.mark.freeze_time("2025-01-28 21:45:00")
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@@ -621,6 +641,99 @@ async def test_external_sensors_unreachable(
assert state.state == STATE_UNAVAILABLE
async def test_external_device_registry_migration(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
init_integration: MockConfigEntry,
) -> None:
"""Test orphaned external devices are removed during setup."""
await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()
_cleanup_external_migration_devices(device_registry, init_integration.entry_id)
old_device = device_registry.async_get_or_create(
config_entry_id=init_integration.entry_id,
identifiers={(DOMAIN, EXTERNAL_DEVICE_OLD_UNIQUE_ID)},
manufacturer="HomeWizard",
model="HWE-P1",
name="Gas meter",
serial_number=EXTERNAL_DEVICE_OLD_UNIQUE_ID,
)
new_device = device_registry.async_get_or_create(
config_entry_id=init_integration.entry_id,
identifiers={(DOMAIN, EXTERNAL_DEVICE_NEW_UNIQUE_ID)},
manufacturer="HomeWizard",
model="HWE-P1",
name="Gas meter",
serial_number=EXTERNAL_DEVICE_NEW_UNIQUE_ID,
)
await hass.config_entries.async_setup(init_integration.entry_id)
await hass.async_block_till_done()
assert device_registry.async_get(old_device.id) is None
assert device_registry.async_get(new_device.id) is not None
async def test_external_device_registry_migration_old_only(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
init_integration: MockConfigEntry,
) -> None:
"""Test old-format external device identifiers are migrated during setup."""
await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()
_cleanup_external_migration_devices(device_registry, init_integration.entry_id)
old_device = device_registry.async_get_or_create(
config_entry_id=init_integration.entry_id,
identifiers={(DOMAIN, EXTERNAL_DEVICE_OLD_UNIQUE_ID)},
manufacturer="HomeWizard",
model="HWE-P1",
name="Gas meter",
serial_number=EXTERNAL_DEVICE_OLD_UNIQUE_ID,
)
await hass.config_entries.async_setup(init_integration.entry_id)
await hass.async_block_till_done()
migrated_device = device_registry.async_get(old_device.id)
assert migrated_device is not None
assert (DOMAIN, EXTERNAL_DEVICE_OLD_UNIQUE_ID) not in migrated_device.identifiers
assert (DOMAIN, EXTERNAL_DEVICE_NEW_UNIQUE_ID) in migrated_device.identifiers
async def test_external_device_registry_migration_new_only_noop(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
init_integration: MockConfigEntry,
) -> None:
"""Test already-migrated external device identifiers remain untouched."""
await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()
_cleanup_external_migration_devices(device_registry, init_integration.entry_id)
new_device = device_registry.async_get_or_create(
config_entry_id=init_integration.entry_id,
identifiers={(DOMAIN, EXTERNAL_DEVICE_NEW_UNIQUE_ID)},
manufacturer="HomeWizard",
model="HWE-P1",
name="Gas meter",
serial_number=EXTERNAL_DEVICE_NEW_UNIQUE_ID,
)
await hass.config_entries.async_setup(init_integration.entry_id)
await hass.async_block_till_done()
post_setup_device = device_registry.async_get(new_device.id)
assert post_setup_device is not None
assert (DOMAIN, EXTERNAL_DEVICE_OLD_UNIQUE_ID) not in post_setup_device.identifiers
assert (DOMAIN, EXTERNAL_DEVICE_NEW_UNIQUE_ID) in post_setup_device.identifiers
@pytest.mark.parametrize(
("device_fixture", "entity_ids"),
[