Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-14 17:23:33 +01:00
Code Activity

Add WS command config/device_registry/list_composite_splits (#176693)

Browse Source
This commit is contained in:
Erik Montnemery
2026-07-28 13:18:16 +02:00
committed by GitHub
parent 1fe5631276
commit 82ca35fd45
4 changed files with 208 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/config/device_registry.py
+40
View File
@@ -17,6 +17,7 @@ from homeassistant.helpers.device_registry import DeviceEntry, DeviceEntryDisabl
def async_setup(hass: HomeAssistant) -> bool:
"""Enable the Device Registry views."""
websocket_api.async_register_command(hass, websocket_list_composite_splits)
websocket_api.async_register_command(hass, websocket_list_devices)
websocket_api.async_register_command(hass, websocket_update_device)
websocket_api.async_register_command(
@@ -25,6 +26,45 @@ def async_setup(hass: HomeAssistant) -> bool:
return True
@callback
@websocket_api.websocket_command(
{
vol.Required("type"): "config/device_registry/list_composite_splits",
}
)
def websocket_list_composite_splits(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Handle list composite device splits command.
Maps every pre-migration composite device id that was removed by splitting the
device into one device per config entry to the ids of the devices which replaced
it, and which of those (if any) belongs to the composite's former primary config
entry.
"""
registry = dr.async_get(hass)
connection.send_result(
msg["id"],
{
composite_id: {
"split_ids": [device.id for device in devices],
"primary_id": next(
(
device.id
for device in devices
if device.config_entry_id
== device.composite_primary_config_entry
),
None,
),
}
for composite_id, devices in registry.devices.get_composite_splits().items()
},
)
@callback
@websocket_api.websocket_command(
{
homeassistant/helpers/device_registry.py
+8
View File
@@ -1221,6 +1221,14 @@ class ActiveDeviceRegistryItems(DeviceRegistryItems[DeviceEntry]):
for key in self._composite_device_id_index.get(composite_device_id, ())
]
def get_composite_splits(self) -> dict[str, list[DeviceEntry]]:
"""Get the pre-migration composite device ids and the devices split from them."""
data = self.data
return {
composite_device_id: [data[key] for key in keys]
for composite_device_id, keys in self._composite_device_id_index.items()
}
class DeletedDeviceRegistryItems(DeviceRegistryItems[DeletedDeviceEntry]):
"""Container for deleted device registry entries.
tests/components/config/test_device_registry.py
+116
View File
@@ -1,6 +1,7 @@
"""Test device_registry API."""
from datetime import datetime
from typing import Any
from freezegun.api import FrozenDateTimeFactory
import pytest
@@ -151,6 +152,121 @@ async def test_list_devices(
device_registry.async_remove_device(device2.id)
def _storage_device_v1_12(
device_id: str,
config_entries: list[str],
primary_config_entry: str | None,
identifier: str,
) -> dict[str, Any]:
"""Return a stored device in version 1.12 format."""
return {
"area_id": None,
"config_entries": config_entries,
"config_entries_subentries": {
config_entry: [None] for config_entry in config_entries
},
"configuration_url": None,
"connections": [],
"created_at": "1970-01-01T00:00:00+00:00",
"disabled_by": None,
"entry_type": None,
"hw_version": None,
"id": device_id,
"identifiers": [["test", identifier]],
"labels": [],
"manufacturer": None,
"model": None,
"model_id": None,
"modified_at": "1970-01-01T00:00:00+00:00",
"name": None,
"name_by_user": None,
"primary_config_entry": primary_config_entry,
"serial_number": None,
"sw_version": None,
"via_device_id": None,
}
@pytest.mark.parametrize("load_registries", [False])
async def test_list_composite_splits(
hass: HomeAssistant,
client: MockHAClientWebSocket,
hass_storage: dict[str, Any],
) -> None:
"""Test listing the devices pre-migration composite devices were split into."""
entry_1 = MockConfigEntry()
entry_1.add_to_hass(hass)
entry_2 = MockConfigEntry()
entry_2.add_to_hass(hass)
entry_3 = MockConfigEntry()
entry_3.add_to_hass(hass)
hass_storage[dr.STORAGE_KEY] = {
"version": 1,
"minor_version": 12,
"key": dr.STORAGE_KEY,
"data": {
"devices": [
# Composite spanning two config entries, with entry_1 as primary
_storage_device_v1_12(
"compositea000000000000000000000",
[entry_1.entry_id, entry_2.entry_id],
entry_1.entry_id,
"a",
),
# Composite spanning two config entries, without a primary
_storage_device_v1_12(
"compositeb000000000000000000000",
[entry_1.entry_id, entry_3.entry_id],
None,
"b",
),
# Single config entry device, not split
_storage_device_v1_12(
"single0000000000000000000000000",
[entry_1.entry_id],
entry_1.entry_id,
"c",
),
],
"deleted_devices": [],
},
}
dr.async_setup(hass)
await dr.async_load(hass)
registry = dr.async_get(hass)
splits_a = registry.async_get_devices_for_composite_device_id(
"compositea000000000000000000000"
)
splits_b = registry.async_get_devices_for_composite_device_id(
"compositeb000000000000000000000"
)
assert len(splits_a) == 2
assert len(splits_b) == 2
primary_a = next(
device for device in splits_a if device.config_entry_id == entry_1.entry_id
)
await client.send_json_auto_id(
{"type": "config/device_registry/list_composite_splits"}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] == {
"compositea000000000000000000000": {
"split_ids": unordered([device.id for device in splits_a]),
"primary_id": primary_a.id,
},
"compositeb000000000000000000000": {
"split_ids": unordered([device.id for device in splits_b]),
"primary_id": None,
},
}
@pytest.mark.parametrize(
("payload_key", "payload_value"),
[
tests/helpers/test_device_registry.py
+44
View File
@@ -8371,6 +8371,50 @@ async def test_restored_composite_preserves_primary_config_entry(
assert composite.primary_config_entry in composite.config_entries
@pytest.mark.parametrize("load_registries", [False])
async def test_get_composite_splits(
hass: HomeAssistant, hass_storage: dict[str, Any]
) -> None:
"""Test getting the mapping of composite device ids to their split devices."""
entry_a = MockConfigEntry(domain="domain_a")
entry_a.add_to_hass(hass)
entry_b = MockConfigEntry(domain="domain_b")
entry_b.add_to_hass(hass)
hass_storage[dr.STORAGE_KEY] = _composite_device_storage(entry_a, entry_b)
dr.async_setup(hass)
await dr.async_load(hass)
device_registry = dr.async_get(hass)
split_a = _get_device_for_config_entry(
device_registry, entry_a.entry_id, identifiers={("domain_a", "1")}
)
split_b = _get_device_for_config_entry(
device_registry, entry_b.entry_id, identifiers={("domain_b", "1")}
)
splits = device_registry.devices.get_composite_splits()
assert set(splits) == {COMPOSITE_ID}
assert {device.id for device in splits[COMPOSITE_ID]} == {split_a.id, split_b.id}
# A device which is not split from a composite is not included
device_registry.async_get_or_create(
config_entry_id=entry_a.entry_id, identifiers={("domain_a", "2")}
)
splits = device_registry.devices.get_composite_splits()
assert set(splits) == {COMPOSITE_ID}
assert {device.id for device in splits[COMPOSITE_ID]} == {split_a.id, split_b.id}
# A removed split is dropped from the mapping
device_registry.async_remove_device(split_a.id)
splits = device_registry.devices.get_composite_splits()
assert {device.id for device in splits[COMPOSITE_ID]} == {split_b.id}
# Removing the last split drops the composite id from the mapping
device_registry.async_remove_device(split_b.id)
assert device_registry.devices.get_composite_splits() == {}
@pytest.mark.parametrize("load_registries", [False])
async def test_clear_config_entry_clears_composite_primary_config_entry(
hass: HomeAssistant, hass_storage: dict[str, Any]