Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-14 20:48:32 +01:00
Code Activity

Use Tibber device IDs for Data API entities (#177527)

Browse Source
This commit is contained in:
Niek van der Maas
2026-09-09 20:17:27 +02:00
committed by GitHub
parent 2166c17956
commit 2edb623a71
6 changed files with 353 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/tibber/__init__.py
+90 -1
View File
@@ -11,7 +11,11 @@ import tibber
from homeassistant.const import CONF_ACCESS_TOKEN, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_registry as er,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.config_entry_oauth2_flow import (
OAuth2Session,
@@ -38,6 +42,89 @@ CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
_LOGGER = logging.getLogger(__name__)
def _migrate_data_api_registry_entries(
hass: HomeAssistant,
entry: TibberConfigEntry,
coordinator: TibberDataAPICoordinator,
home_ids: set[str],
) -> None:
"""Migrate Data API registry entries to Tibber device IDs."""
entity_registry = er.async_get(hass)
entity_entries = er.async_entries_for_config_entry(entity_registry, entry.entry_id)
device_registry = dr.async_get(hass)
legacy_device = device_registry.async_get_device_by_identifier(
(DOMAIN, ""), entry.entry_id
)
legacy_device_name = legacy_device.name if legacy_device else None
migrations: dict[str, tuple[str, str]] = {}
device_migrations: dict[str, set[str]] = {}
device_id_by_identifier = {home_id: home_id for home_id in home_ids}
for device in sorted(
coordinator.data.values(),
key=lambda device: (device.name != legacy_device_name, device.id),
):
if (
legacy_device
and not device.external_id
and device.name == legacy_device_name
):
device_id_by_identifier.setdefault("", device.id)
device_id_by_identifier[device.id] = device.id
if device.external_id:
device_id_by_identifier[device.external_id] = device.id
for sensor in device.sensors:
new_unique_id = f"{device.id}_{sensor.id}"
migration = (new_unique_id, device.id)
migrations[new_unique_id] = migration
if device.external_id:
migrations[f"{device.external_id}_{sensor.id}"] = migration
else:
# An empty external ID produced a legacy leading-underscore unique ID.
migrations.setdefault(f"_{sensor.id}", migration)
for entity_entry in entity_entries:
if not (registry_migration := migrations.get(entity_entry.unique_id)):
continue
new_unique_id, device_id = registry_migration
if entity_entry.device_id:
# Empty external IDs may have grouped multiple Tibber devices.
device_migrations.setdefault(entity_entry.device_id, set()).add(device_id)
if entity_entry.unique_id != new_unique_id:
entity_registry.async_update_entity(
entity_entry.entity_id, new_unique_id=new_unique_id
)
for registry_device in dr.async_entries_for_config_entry(
device_registry, entry.entry_id
):
if tibber_device_ids := device_migrations.get(registry_device.id):
tibber_device_id = min(
(
coordinator.data[device_id].name != registry_device.name,
device_id,
)
for device_id in tibber_device_ids
)[1]
else:
expected_device_ids = {
device_id_by_identifier[identifier]
for domain, identifier in registry_device.identifiers
if domain == DOMAIN and identifier in device_id_by_identifier
}
if len(expected_device_ids) != 1:
device_registry.async_remove_device(registry_device.id)
continue
tibber_device_id = expected_device_ids.pop()
identifiers = {(DOMAIN, tibber_device_id)}
if registry_device.identifiers != identifiers:
device_registry.async_update_device(
registry_device.id,
new_identifiers=identifiers,
)
@dataclass
class TibberRuntimeData:
"""Runtime data for Tibber API entries."""
@@ -152,6 +239,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: TibberConfigEntry) -> bo
coordinator = TibberDataAPICoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data.data_api_coordinator = coordinator
home_ids = {home.home_id for home in tibber_connection.get_homes(only_active=False)}
_migrate_data_api_registry_entries(hass, entry, coordinator, home_ids)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
homeassistant/components/tibber/binary_sensor.py
+1 -1
View File
@@ -103,7 +103,7 @@ class TibberDataAPIBinarySensor(
self._attr_unique_id = f"{device.id}_{entity_description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device.external_id)},
identifiers={(DOMAIN, device.id)},
name=device.name,
manufacturer=device.brand,
model=device.model,
homeassistant/components/tibber/sensor.py
+2 -2
View File
@@ -688,10 +688,10 @@ class TibberDataAPISensor(CoordinatorEntity[TibberDataAPICoordinator], SensorEnt
self.entity_description = entity_description
self._attr_translation_key = entity_description.translation_key
self._attr_unique_id = f"{device.external_id}_{self.entity_description.key}"
self._attr_unique_id = f"{device.id}_{self.entity_description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device.external_id)},
identifiers={(DOMAIN, device.id)},
name=device.name,
manufacturer=device.brand,
model=device.model,
tests/components/tibber/test_binary_sensor.py
+93 -1
View File
@@ -6,9 +6,14 @@ import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.recorder import Recorder
from homeassistant.components.tibber.const import DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import (
area_registry as ar,
device_registry as dr,
entity_registry as er,
)
from .conftest import create_tibber_device
@@ -46,6 +51,93 @@ async def test_binary_sensor_snapshot(
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
async def test_binary_sensors_with_empty_external_ids(
recorder_mock: Recorder,
hass: HomeAssistant,
config_entry: MockConfigEntry,
data_api_client_mock: AsyncMock,
setup_credentials: None,
area_registry: ar.AreaRegistry,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test binary sensors migrate empty external ID registry entries."""
area = area_registry.async_get_or_create("Outside")
legacy_device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, "")},
name="Charger left",
)
legacy_device = device_registry.async_update_device(
legacy_device.id, area_id=area.id
)
assert legacy_device is not None
right_legacy_entity = entity_registry.async_get_or_create(
"binary_sensor",
DOMAIN,
"charger-right_connector.status",
suggested_object_id="legacy_right_connector_status",
config_entry=config_entry,
device_id=legacy_device.id,
)
left_legacy_entity = entity_registry.async_get_or_create(
"binary_sensor",
DOMAIN,
"charger-left_connector.status",
suggested_object_id="legacy_connector_status",
config_entry=config_entry,
device_id=legacy_device.id,
)
devices = {
device_id: create_tibber_device(
device_id=device_id,
external_id="",
name=name,
connector_status="connected",
)
for device_id, name in (
("charger-left", "Charger left"),
("charger-right", "Charger right"),
)
}
data_api_client_mock.get_all_devices = AsyncMock(return_value=devices)
data_api_client_mock.update_devices = AsyncMock(return_value=devices)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
left_entity_id = entity_registry.async_get_entity_id(
"binary_sensor", DOMAIN, "charger-left_connector.status"
)
right_entity_id = entity_registry.async_get_entity_id(
"binary_sensor", DOMAIN, "charger-right_connector.status"
)
assert left_entity_id is not None
assert right_entity_id is not None
assert left_entity_id == left_legacy_entity.entity_id
assert right_entity_id == right_legacy_entity.entity_id
left_entity = entity_registry.async_get(left_entity_id)
right_entity = entity_registry.async_get(right_entity_id)
assert left_entity is not None
assert right_entity is not None
assert left_entity.device_id != right_entity.device_id
left_device = device_registry.async_get_device_by_identifier(
(DOMAIN, "charger-left"), config_entry.entry_id
)
right_device = device_registry.async_get_device_by_identifier(
(DOMAIN, "charger-right"), config_entry.entry_id
)
assert left_device is not None
assert right_device is not None
assert left_device.id == legacy_device.id
assert left_device.area_id == area.id
assert left_entity.device_id == left_device.id
assert right_entity.device_id == right_device.id
@pytest.mark.parametrize(
(
"entity_suffix",
tests/components/tibber/test_coordinator.py
+1 -1
View File
@@ -59,7 +59,7 @@ async def _async_setup_data_api_sensor(
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, "external-id_storage.stateOfCharge"
"sensor", DOMAIN, "device-id_storage.stateOfCharge"
)
assert entity_id is not None
assert hass.states.get(entity_id) is not None
tests/components/tibber/test_sensor.py
+166 -11
View File
@@ -7,7 +7,11 @@ import pytest
from homeassistant.components.recorder import Recorder
from homeassistant.components.tibber.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import (
area_registry as ar,
device_registry as dr,
entity_registry as er,
)
from homeassistant.helpers.entity_component import async_update_entity
from .conftest import create_tibber_device, create_tibber_home
@@ -67,20 +71,58 @@ async def test_price_sensor_state_unit_and_attributes(
assert state.attributes["off_peak_2"] == 1.0
async def test_data_api_sensors_are_created(
recorder_mock: Recorder,
@pytest.mark.usefixtures("recorder_mock", "setup_credentials")
async def test_data_api_sensors_migrate_to_device_id(
hass: HomeAssistant,
config_entry: MockConfigEntry,
data_api_client_mock: AsyncMock,
setup_credentials: None,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
tibber_mock: MagicMock,
) -> None:
"""Ensure Data API sensors are created and expose values from the coordinator."""
"""Test Data API sensors migrate to Tibber device IDs."""
home = create_tibber_home()
tibber_mock.get_homes.return_value = [home]
home_device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, home.home_id)},
)
legacy_device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, "external-id")},
name="Test Device",
)
orphaned_device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, "orphaned-external-id")},
)
stale_device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, "stale-device")},
)
legacy_entity = entity_registry.async_get_or_create(
"sensor",
DOMAIN,
"external-id_storage.stateOfCharge",
suggested_object_id="legacy_state_of_charge",
config_entry=config_entry,
device_id=legacy_device.id,
)
orphaned_tibber_device = create_tibber_device(
device_id="orphaned-device-id",
external_id="orphaned-external-id",
)
data_api_client_mock.get_all_devices = AsyncMock(
return_value={"device-id": create_tibber_device(state_of_charge=72.0)}
return_value={
"device-id": create_tibber_device(state_of_charge=72.0),
"orphaned-device-id": orphaned_tibber_device,
}
)
data_api_client_mock.update_devices = AsyncMock(
return_value={"device-id": create_tibber_device(state_of_charge=83.0)}
return_value={
"device-id": create_tibber_device(state_of_charge=83.0),
"orphaned-device-id": orphaned_tibber_device,
}
)
await hass.config_entries.async_setup(config_entry.entry_id)
@@ -89,15 +131,128 @@ async def test_data_api_sensors_are_created(
data_api_client_mock.get_all_devices.assert_awaited_once()
data_api_client_mock.update_devices.assert_awaited_once()
unique_id = "external-id_storage.stateOfCharge"
unique_id = "device-id_storage.stateOfCharge"
entity_id = entity_registry.async_get_entity_id("sensor", DOMAIN, unique_id)
assert entity_id is not None
assert entity_id == legacy_entity.entity_id
device = device_registry.async_get_device_by_identifier(
(DOMAIN, "device-id"), config_entry.entry_id
)
assert device is not None
assert device.id == legacy_device.id
migrated_orphaned_device = device_registry.async_get_device_by_identifier(
(DOMAIN, "orphaned-device-id"), config_entry.entry_id
)
assert migrated_orphaned_device is not None
assert migrated_orphaned_device.id == orphaned_device.id
assert device_registry.async_get(home_device.id) is not None
assert device_registry.async_get(stale_device.id) is None
state = hass.states.get(entity_id)
assert state is not None
assert float(state.state) == 83.0
@pytest.mark.usefixtures("recorder_mock", "setup_credentials")
@pytest.mark.parametrize(
"unsupported_device_name", ["Unsupported device", "Charger left"]
)
async def test_data_api_sensors_with_empty_external_ids(
hass: HomeAssistant,
config_entry: MockConfigEntry,
data_api_client_mock: AsyncMock,
area_registry: ar.AreaRegistry,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
unsupported_device_name: str,
) -> None:
"""Test Data API sensors migrate empty external ID registry entries."""
area = area_registry.async_get_or_create("Outside")
legacy_device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, "")},
name="Charger left",
)
legacy_device = device_registry.async_update_device(
legacy_device.id, area_id=area.id
)
assert legacy_device is not None
legacy_entities = {
sensor_id: entity_registry.async_get_or_create(
"sensor",
DOMAIN,
f"_{sensor_id}",
suggested_object_id=f"legacy_{sensor_id}",
config_entry=config_entry,
device_id=legacy_device.id,
)
for sensor_id in (
"charging.current.max",
"charging.current.offlineFallback",
)
}
devices = {
"a-unsupported-device": create_tibber_device(
device_id="a-unsupported-device",
external_id="",
name=unsupported_device_name,
sensor_values={"unknown.sensor.id": None},
),
**{
device_id: create_tibber_device(
device_id=device_id,
external_id="",
name=name,
sensor_values={
"charging.current.max": 32.0,
"charging.current.offlineFallback": 16.0,
},
)
for device_id, name in (
("charger-right", "Charger right"),
("charger-left", "Charger left"),
)
},
}
data_api_client_mock.get_all_devices = AsyncMock(return_value=devices)
data_api_client_mock.update_devices = AsyncMock(return_value=devices)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
left_entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, "charger-left_charging.current.max"
)
right_entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, "charger-right_charging.current.max"
)
assert left_entity_id is not None
assert right_entity_id is not None
assert left_entity_id == legacy_entities["charging.current.max"].entity_id
fallback_entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, "charger-left_charging.current.offlineFallback"
)
assert (
fallback_entity_id
== legacy_entities["charging.current.offlineFallback"].entity_id
)
left_entity = entity_registry.async_get(left_entity_id)
right_entity = entity_registry.async_get(right_entity_id)
assert left_entity is not None
assert right_entity is not None
assert left_entity.device_id != right_entity.device_id
left_device = device_registry.async_get_device_by_identifier(
(DOMAIN, "charger-left"), config_entry.entry_id
)
assert left_device is not None
assert left_device.id == legacy_device.id
assert left_device.area_id == area.id
@pytest.mark.parametrize(
("sensor_id", "expected_value", "description"),
[
@@ -151,7 +306,7 @@ async def test_new_data_api_sensor_values(
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
unique_id = f"external-id_{sensor_id}"
unique_id = f"device-id_{sensor_id}"
entity_id = entity_registry.async_get_entity_id("sensor", DOMAIN, unique_id)
assert entity_id is not None, f"Entity not found for {description}"
@@ -211,7 +366,7 @@ async def test_new_data_api_sensors_with_disabled_by_default(
]
for sensor_id in disabled_sensors:
unique_id = f"external-id_{sensor_id}"
unique_id = f"device-id_{sensor_id}"
entity_id = entity_registry.async_get_entity_id("sensor", DOMAIN, unique_id)
assert entity_id is not None, f"Entity not found for sensor {sensor_id}"