Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add hardware version to the device registry (#61650)

pull/61743/head
J. Nick Koston 2021-12-16 05:16:19 -06:00 committed by GitHub
parent b1b3079d07
commit 04153c0075
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 206 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
homeassistant/components/config/device_registry.py
View File

@ -84,5 +84,6 @@ def _entry_dict(entry):
"name_by_user": entry.name_by_user,
"name": entry.name,
"sw_version": entry.sw_version,
"hw_version": entry.hw_version,
"via_device_id": entry.via_device_id,
}

66
homeassistant/helpers/device_registry.py
View File

@ -33,7 +33,7 @@ DATA_REGISTRY = "device_registry"
EVENT_DEVICE_REGISTRY_UPDATED = "device_registry_updated"
STORAGE_KEY = "core.device_registry"
STORAGE_VERSION_MAJOR = 1
STORAGE_VERSION_MINOR = 2
STORAGE_VERSION_MINOR = 3
SAVE_DELAY = 10
CLEANUP_DELAY = 10
@ -87,6 +87,7 @@ class DeviceEntry:
name: str | None = attr.ib(default=None)
suggested_area: str | None = attr.ib(default=None)
sw_version: str | None = attr.ib(default=None)
hw_version: str | None = attr.ib(default=None)
via_device_id: str | None = attr.ib(default=None)
# This value is not stored, just used to keep track of events to fire.
is_new: bool = attr.ib(default=False)
@ -168,32 +169,37 @@ class DeviceRegistryStore(storage.Store):
self, old_major_version: int, old_minor_version: int, old_data: dict[str, Any]
) -> dict[str, Any]:
"""Migrate to the new version."""
if old_major_version < 2 and old_minor_version < 2:
# From version 1.1
for device in old_data["devices"]:
# Introduced in 0.110
try:
device["entry_type"] = DeviceEntryType(device.get("entry_type"))
except ValueError:
device["entry_type"] = None
if old_major_version < 2:
if old_minor_version < 2:
# From version 1.1
for device in old_data["devices"]:
# Introduced in 0.110
try:
device["entry_type"] = DeviceEntryType(device.get("entry_type"))
except ValueError:
device["entry_type"] = None
# Introduced in 0.79
# renamed in 0.95
device["via_device_id"] = device.get("via_device_id") or device.get(
"hub_device_id"
)
# Introduced in 0.87
device["area_id"] = device.get("area_id")
device["name_by_user"] = device.get("name_by_user")
# Introduced in 0.119
device["disabled_by"] = device.get("disabled_by")
# Introduced in 2021.11
device["configuration_url"] = device.get("configuration_url")
# Introduced in 0.111
old_data["deleted_devices"] = old_data.get("deleted_devices", [])
for device in old_data["deleted_devices"]:
# Introduced in 2021.2
device["orphaned_timestamp"] = device.get("orphaned_timestamp")
# Introduced in 0.79
# renamed in 0.95
device["via_device_id"] = device.get("via_device_id") or device.get(
"hub_device_id"
)
# Introduced in 0.87
device["area_id"] = device.get("area_id")
device["name_by_user"] = device.get("name_by_user")
# Introduced in 0.119
device["disabled_by"] = device.get("disabled_by")
# Introduced in 2021.11
device["configuration_url"] = device.get("configuration_url")
# Introduced in 0.111
old_data["deleted_devices"] = old_data.get("deleted_devices", [])
for device in old_data["deleted_devices"]:
# Introduced in 2021.2
device["orphaned_timestamp"] = device.get("orphaned_timestamp")
if old_minor_version < 3:
# Introduced in 2022.2
for device in old_data["devices"]:
device["hw_version"] = device.get("hw_version")
if old_major_version > 1:
raise NotImplementedError
@ -314,6 +320,7 @@ class DeviceRegistry:
name: str | None | UndefinedType = UNDEFINED,
suggested_area: str | None | UndefinedType = UNDEFINED,
sw_version: str | None | UndefinedType = UNDEFINED,
hw_version: str | None | UndefinedType = UNDEFINED,
via_device: tuple[str, str] | None = None,
) -> DeviceEntry:
"""Get device. Create if it doesn't exist."""
@ -378,6 +385,7 @@ class DeviceRegistry:
name=name,
suggested_area=suggested_area,
sw_version=sw_version,
hw_version=hw_version,
via_device_id=via_device_id,
)
@ -403,6 +411,7 @@ class DeviceRegistry:
remove_config_entry_id: str | UndefinedType = UNDEFINED,
suggested_area: str | None | UndefinedType = UNDEFINED,
sw_version: str | None | UndefinedType = UNDEFINED,
hw_version: str | None | UndefinedType = UNDEFINED,
via_device_id: str | None | UndefinedType = UNDEFINED,
) -> DeviceEntry | None:
"""Update properties of a device."""
@ -420,6 +429,7 @@ class DeviceRegistry:
remove_config_entry_id=remove_config_entry_id,
suggested_area=suggested_area,
sw_version=sw_version,
hw_version=hw_version,
via_device_id=via_device_id,
)
@ -443,6 +453,7 @@ class DeviceRegistry:
remove_config_entry_id: str | UndefinedType = UNDEFINED,
suggested_area: str | None | UndefinedType = UNDEFINED,
sw_version: str | None | UndefinedType = UNDEFINED,
hw_version: str | None | UndefinedType = UNDEFINED,
via_device_id: str | None | UndefinedType = UNDEFINED,
) -> DeviceEntry | None:
"""Update device attributes."""
@ -513,6 +524,7 @@ class DeviceRegistry:
("name", name),
("suggested_area", suggested_area),
("sw_version", sw_version),
("hw_version", hw_version),
("via_device_id", via_device_id),
):
if value is not UNDEFINED and value != getattr(old, attr_name):
@ -595,6 +607,7 @@ class DeviceRegistry:
name_by_user=device["name_by_user"],
name=device["name"],
sw_version=device["sw_version"],
hw_version=device["hw_version"],
via_device_id=device["via_device_id"],
)
# Introduced in 0.111
@ -631,6 +644,7 @@ class DeviceRegistry:
"model": entry.model,
"name": entry.name,
"sw_version": entry.sw_version,
"hw_version": entry.hw_version,
"entry_type": entry.entry_type,
"id": entry.id,
"via_device_id": entry.via_device_id,

1
homeassistant/helpers/entity.py
View File

@ -177,6 +177,7 @@ class DeviceInfo(TypedDict, total=False):
name: str | None
suggested_area: str | None
sw_version: str | None
hw_version: str | None
via_device: tuple[str, str]

2
tests/components/config/test_device_registry.py
View File

@ -53,6 +53,7 @@ async def test_list_devices(hass, client, registry):
"model": "model",
"name": None,
"sw_version": None,
"hw_version": None,
"entry_type": None,
"via_device_id": None,
"area_id": None,
@ -68,6 +69,7 @@ async def test_list_devices(hass, client, registry):
"model": "model",
"name": None,
"sw_version": None,
"hw_version": None,
"entry_type": helpers_dr.DeviceEntryType.SERVICE,
"via_device_id": dev1,
"area_id": None,

164
tests/helpers/test_device_registry.py
View File

@ -185,6 +185,7 @@ async def test_loading_from_storage(hass, hass_storage):
"name_by_user": "Test Friendly Name",
"name": "name",
"sw_version": "version",
"hw_version": "hw_version",
"via_device_id": None,
}
],
@ -215,6 +216,7 @@ async def test_loading_from_storage(hass, hass_storage):
assert entry.id == "abcdefghijklm"
assert entry.area_id == "12345A"
assert entry.name_by_user == "Test Friendly Name"
assert entry.hw_version == "hw_version"
assert entry.entry_type is device_registry.DeviceEntryType.SERVICE
assert entry.disabled_by is device_registry.DeviceEntryDisabler.USER
assert isinstance(entry.config_entries, set)
@ -235,8 +237,8 @@ async def test_loading_from_storage(hass, hass_storage):
@pytest.mark.parametrize("load_registries", [False])
async def test_migration_1_1_to_1_2(hass, hass_storage):
"""Test migration from version 1.1 to 1.2."""
async def test_migration_1_1_to_1_3(hass, hass_storage):
"""Test migration from version 1.1 to 1.3."""
hass_storage[device_registry.STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
@ -266,6 +268,19 @@ async def test_migration_1_1_to_1_2(hass, hass_storage):
"sw_version": None,
},
],
"deleted_devices": [
{
"config_entries": ["123456"],
"connections": [],
"entry_type": "service",
"id": "deletedid",
"identifiers": [["serial", "12:34:56:AB:CD:FF"]],
"manufacturer": "manufacturer",
"model": "model",
"name": "name",
"sw_version": "version",
}
],
},
}
@ -311,6 +326,7 @@ async def test_migration_1_1_to_1_2(hass, hass_storage):
"name": "name",
"name_by_user": None,
"sw_version": "new_version",
"hw_version": None,
"via_device_id": None,
},
{
@ -327,6 +343,132 @@ async def test_migration_1_1_to_1_2(hass, hass_storage):
"name_by_user": None,
"name": None,
"sw_version": None,
"hw_version": None,
"via_device_id": None,
},
],
"deleted_devices": [
{
"config_entries": ["123456"],
"connections": [],
"id": "deletedid",
"identifiers": [["serial", "12:34:56:AB:CD:FF"]],
"orphaned_timestamp": None,
}
],
},
}
@pytest.mark.parametrize("load_registries", [False])
async def test_migration_1_2_to_1_3(hass, hass_storage):
"""Test migration from version 1.2 to 1.3."""
hass_storage[device_registry.STORAGE_KEY] = {
"version": 1,
"minor_version": 2,
"key": device_registry.STORAGE_KEY,
"data": {
"devices": [
{
"area_id": None,
"config_entries": ["1234"],
"configuration_url": None,
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": None,
"entry_type": "service",
"id": "abcdefghijklm",
"identifiers": [["serial", "12:34:56:AB:CD:EF"]],
"manufacturer": "manufacturer",
"model": "model",
"name": "name",
"name_by_user": None,
"sw_version": "new_version",
"hw_version": None,
"via_device_id": None,
},
{
"area_id": None,
"config_entries": [None],
"configuration_url": None,
"connections": [],
"disabled_by": None,
"entry_type": None,
"id": "invalid-entry-type",
"identifiers": [["serial", "mock-id-invalid-entry"]],
"manufacturer": None,
"model": None,
"name_by_user": None,
"name": None,
"sw_version": None,
"hw_version": None,
"via_device_id": None,
},
],
"deleted_devices": [],
},
}
await device_registry.async_load(hass)
registry = device_registry.async_get(hass)
# Test data was loaded
entry = registry.async_get_or_create(
config_entry_id="1234",
connections={("Zigbee", "01.23.45.67.89")},
identifiers={("serial", "12:34:56:AB:CD:EF")},
)
assert entry.id == "abcdefghijklm"
# Update to trigger a store
entry = registry.async_get_or_create(
config_entry_id="1234",
connections={("Zigbee", "01.23.45.67.89")},
identifiers={("serial", "12:34:56:AB:CD:EF")},
hw_version="new_version",
)
assert entry.id == "abcdefghijklm"
# Check we store migrated data
await flush_store(registry._store)
assert hass_storage[device_registry.STORAGE_KEY] == {
"version": device_registry.STORAGE_VERSION_MAJOR,
"minor_version": device_registry.STORAGE_VERSION_MINOR,
"key": device_registry.STORAGE_KEY,
"data": {
"devices": [
{
"area_id": None,
"config_entries": ["1234"],
"configuration_url": None,
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": None,
"entry_type": "service",
"id": "abcdefghijklm",
"identifiers": [["serial", "12:34:56:AB:CD:EF"]],
"manufacturer": "manufacturer",
"model": "model",
"name": "name",
"name_by_user": None,
"sw_version": "new_version",
"hw_version": "new_version",
"via_device_id": None,
},
{
"area_id": None,
"config_entries": [None],
"configuration_url": None,
"connections": [],
"disabled_by": None,
"entry_type": None,
"id": "invalid-entry-type",
"identifiers": [["serial", "mock-id-invalid-entry"]],
"manufacturer": None,
"model": None,
"name_by_user": None,
"name": None,
"sw_version": None,
"hw_version": None,
"via_device_id": None,
},
],
@ -865,6 +1007,24 @@ async def test_update_sw_version(registry):
assert updated_entry.sw_version == sw_version
async def test_update_hw_version(registry):
"""Verify that we can update hardware version of a device."""
entry = registry.async_get_or_create(
config_entry_id="1234",
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bla", "123")},
)
assert not entry.hw_version
hw_version = "0x20020263"
with patch.object(registry, "async_schedule_save") as mock_save:
updated_entry = registry.async_update_device(entry.id, hw_version=hw_version)
assert mock_save.call_count == 1
assert updated_entry != entry
assert updated_entry.hw_version == hw_version
async def test_update_suggested_area(registry, area_registry):
"""Verify that we can update the suggested area version of a device."""
entry = registry.async_get_or_create(