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

Improve some device registry tests (#77659)

Browse Source
This commit is contained in:
Erik Montnemery 2022-09-02 08:54:02 +02:00 committed by GitHub
parent 8bab2a9bea
commit 8924725d69
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 94 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -160,6 +160,7 @@ def _entry_dict(entry):
"connections": list(entry.connections),
"disabled_by": entry.disabled_by,
"entry_type": entry.entry_type,
"hw_version": entry.hw_version,
"id": entry.id,
"identifiers": list(entry.identifiers),
"manufacturer": entry.manufacturer,
@ -167,6 +168,5 @@ 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,
}

30
homeassistant/helpers/device_registry.py
View File

@ -83,6 +83,7 @@ class DeviceEntry:
connections: set[tuple[str, str]] = attr.ib(converter=set, factory=set)
disabled_by: DeviceEntryDisabler | None = attr.ib(default=None)
entry_type: DeviceEntryType | None = attr.ib(default=None)
hw_version: str | None = attr.ib(default=None)
id: str = attr.ib(factory=uuid_util.random_uuid_hex)
identifiers: set[tuple[str, str]] = attr.ib(converter=set, factory=set)
manufacturer: str | None = attr.ib(default=None)
@ -91,7 +92,6 @@ 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)
@ -318,13 +318,13 @@ class DeviceRegistry:
# To disable a device if it gets created
disabled_by: DeviceEntryDisabler | None | UndefinedType = UNDEFINED,
entry_type: DeviceEntryType | None | UndefinedType = UNDEFINED,
hw_version: str | None | UndefinedType = UNDEFINED,
identifiers: set[tuple[str, str]] | None = None,
manufacturer: str | None | UndefinedType = UNDEFINED,
model: str | None | UndefinedType = UNDEFINED,
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."""
@ -382,6 +382,7 @@ class DeviceRegistry:
configuration_url=configuration_url,
disabled_by=disabled_by,
entry_type=entry_type,
hw_version=hw_version,
manufacturer=manufacturer,
merge_connections=connections or UNDEFINED,
merge_identifiers=identifiers or UNDEFINED,
@ -389,7 +390,6 @@ class DeviceRegistry:
name=name,
suggested_area=suggested_area,
sw_version=sw_version,
hw_version=hw_version,
via_device_id=via_device_id,
)
@ -408,6 +408,7 @@ class DeviceRegistry:
configuration_url: str | None | UndefinedType = UNDEFINED,
disabled_by: DeviceEntryDisabler | None | UndefinedType = UNDEFINED,
entry_type: DeviceEntryType | None | UndefinedType = UNDEFINED,
hw_version: str | None | UndefinedType = UNDEFINED,
manufacturer: str | None | UndefinedType = UNDEFINED,
merge_connections: set[tuple[str, str]] | UndefinedType = UNDEFINED,
merge_identifiers: set[tuple[str, str]] | UndefinedType = UNDEFINED,
@ -418,7 +419,6 @@ 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."""
@ -492,17 +492,17 @@ class DeviceRegistry:
old_values["identifiers"] = old.identifiers
for attr_name, value in (
("area_id", area_id),
("configuration_url", configuration_url),
("disabled_by", disabled_by),
("entry_type", entry_type),
("hw_version", hw_version),
("manufacturer", manufacturer),
("model", model),
("name", name),
("name_by_user", name_by_user),
("area_id", area_id),
("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):
@ -584,6 +584,7 @@ class DeviceRegistry:
entry_type=DeviceEntryType(device["entry_type"])
if device["entry_type"]
else None,
hw_version=device["hw_version"],
id=device["id"],
identifiers={tuple(iden) for iden in device["identifiers"]}, # type: ignore[misc]
manufacturer=device["manufacturer"],
@ -591,7 +592,6 @@ 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
@ -617,25 +617,25 @@ class DeviceRegistry:
@callback
def _data_to_save(self) -> dict[str, list[dict[str, Any]]]:
"""Return data of device registry to store in a file."""
data = {}
data: dict[str, list[dict[str, Any]]] = {}
data["devices"] = [
{
"area_id": entry.area_id,
"config_entries": list(entry.config_entries),
"configuration_url": entry.configuration_url,
"connections": list(entry.connections),
"disabled_by": entry.disabled_by,
"entry_type": entry.entry_type,
"hw_version": entry.hw_version,
"id": entry.id,
"identifiers": list(entry.identifiers),
"manufacturer": entry.manufacturer,
"model": entry.model,
"name_by_user": entry.name_by_user,
"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,
"area_id": entry.area_id,
"name_by_user": entry.name_by_user,
"disabled_by": entry.disabled_by,
"configuration_url": entry.configuration_url,
}
for entry in self.devices.values()
]

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

@ -52,36 +52,36 @@ async def test_list_devices(hass, client, registry):
assert msg["result"] == [
{
"area_id": None,
"config_entries": ["1234"],
"configuration_url": None,
"connections": [["ethernet", "12:34:56:78:90:AB:CD:EF"]],
"disabled_by": None,
"entry_type": None,
"hw_version": None,
"identifiers": [["bridgeid", "0123"]],
"manufacturer": "manufacturer",
"model": "model",
"name_by_user": None,
"name": None,
"sw_version": None,
"hw_version": None,
"entry_type": None,
"via_device_id": None,
"area_id": None,
"name_by_user": None,
"disabled_by": None,
"configuration_url": None,
},
{
"area_id": None,
"config_entries": ["1234"],
"configuration_url": None,
"connections": [],
"disabled_by": None,
"entry_type": helpers_dr.DeviceEntryType.SERVICE,
"hw_version": None,
"identifiers": [["bridgeid", "1234"]],
"manufacturer": "manufacturer",
"model": "model",
"name_by_user": None,
"name": None,
"sw_version": None,
"hw_version": None,
"entry_type": helpers_dr.DeviceEntryType.SERVICE,
"via_device_id": dev1,
"area_id": None,
"name_by_user": None,
"disabled_by": None,
"configuration_url": None,
},
]
@ -104,8 +104,8 @@ async def test_list_devices(hass, client, registry):
"identifiers": [["bridgeid", "0123"]],
"manufacturer": "manufacturer",
"model": "model",
"name": None,
"name_by_user": None,
"name": None,
"sw_version": None,
"via_device_id": None,
}

145
tests/helpers/test_device_registry.py
View File

@ -178,10 +178,11 @@ async def test_loading_from_storage(hass, hass_storage):
{
"area_id": "12345A",
"config_entries": ["1234"],
"configuration_url": None,
"configuration_url": "configuration_url",
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": device_registry.DeviceEntryDisabler.USER,
"entry_type": device_registry.DeviceEntryType.SERVICE,
"hw_version": "hw_version",
"id": "abcdefghijklm",
"identifiers": [["serial", "12:34:56:AB:CD:EF"]],
"manufacturer": "manufacturer",
@ -189,7 +190,6 @@ 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,
}
],
@ -217,16 +217,28 @@ async def test_loading_from_storage(hass, hass_storage):
manufacturer="manufacturer",
model="model",
)
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 entry == device_registry.DeviceEntry(
area_id="12345A",
config_entries={"1234"},
configuration_url="configuration_url",
connections={("Zigbee", "01.23.45.67.89")},
disabled_by=device_registry.DeviceEntryDisabler.USER,
entry_type=device_registry.DeviceEntryType.SERVICE,
hw_version="hw_version",
id="abcdefghijklm",
identifiers={("serial", "12:34:56:AB:CD:EF")},
manufacturer="manufacturer",
model="model",
name_by_user="Test Friendly Name",
name="name",
suggested_area=None, # Not stored
sw_version="version",
)
assert isinstance(entry.config_entries, set)
assert isinstance(entry.connections, set)
assert isinstance(entry.identifiers, set)
# Restore a device, id should be reused from the deleted device entry
entry = registry.async_get_or_create(
config_entry_id="1234",
connections={("Zigbee", "23.45.67.89.01")},
@ -234,6 +246,14 @@ async def test_loading_from_storage(hass, hass_storage):
manufacturer="manufacturer",
model="model",
)
assert entry == device_registry.DeviceEntry(
config_entries={"1234"},
connections={("Zigbee", "23.45.67.89.01")},
id="bcdefghijklmn",
identifiers={("serial", "34:56:AB:CD:EF:12")},
manufacturer="manufacturer",
model="model",
)
assert entry.id == "bcdefghijklmn"
assert isinstance(entry.config_entries, set)
assert isinstance(entry.connections, set)
@ -323,6 +343,7 @@ async def test_migration_1_1_to_1_3(hass, hass_storage):
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": None,
"entry_type": "service",
"hw_version": None,
"id": "abcdefghijklm",
"identifiers": [["serial", "12:34:56:AB:CD:EF"]],
"manufacturer": "manufacturer",
@ -330,7 +351,6 @@ async def test_migration_1_1_to_1_3(hass, hass_storage):
"name": "name",
"name_by_user": None,
"sw_version": "new_version",
"hw_version": None,
"via_device_id": None,
},
{
@ -340,6 +360,7 @@ async def test_migration_1_1_to_1_3(hass, hass_storage):
"connections": [],
"disabled_by": None,
"entry_type": None,
"hw_version": None,
"id": "invalid-entry-type",
"identifiers": [["serial", "mock-id-invalid-entry"]],
"manufacturer": None,
@ -347,7 +368,6 @@ async def test_migration_1_1_to_1_3(hass, hass_storage):
"name_by_user": None,
"name": None,
"sw_version": None,
"hw_version": None,
"via_device_id": None,
},
],
@ -386,8 +406,7 @@ async def test_migration_1_2_to_1_3(hass, hass_storage):
"model": "model",
"name": "name",
"name_by_user": None,
"sw_version": "new_version",
"hw_version": None,
"sw_version": "version",
"via_device_id": None,
},
{
@ -404,7 +423,6 @@ async def test_migration_1_2_to_1_3(hass, hass_storage):
"name_by_user": None,
"name": None,
"sw_version": None,
"hw_version": None,
"via_device_id": None,
},
],
@ -428,7 +446,7 @@ async def test_migration_1_2_to_1_3(hass, hass_storage):
config_entry_id="1234",
connections={("Zigbee", "01.23.45.67.89")},
identifiers={("serial", "12:34:56:AB:CD:EF")},
hw_version="new_version",
sw_version="new_version",
)
assert entry.id == "abcdefghijklm"
@ -448,6 +466,7 @@ async def test_migration_1_2_to_1_3(hass, hass_storage):
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": None,
"entry_type": "service",
"hw_version": None,
"id": "abcdefghijklm",
"identifiers": [["serial", "12:34:56:AB:CD:EF"]],
"manufacturer": "manufacturer",
@ -455,7 +474,6 @@ async def test_migration_1_2_to_1_3(hass, hass_storage):
"name": "name",
"name_by_user": None,
"sw_version": "new_version",
"hw_version": "new_version",
"via_device_id": None,
},
{
@ -465,6 +483,7 @@ async def test_migration_1_2_to_1_3(hass, hass_storage):
"connections": [],
"disabled_by": None,
"entry_type": None,
"hw_version": None,
"id": "invalid-entry-type",
"identifiers": [["serial", "mock-id-invalid-entry"]],
"manufacturer": None,
@ -472,7 +491,6 @@ async def test_migration_1_2_to_1_3(hass, hass_storage):
"name_by_user": None,
"name": None,
"sw_version": None,
"hw_version": None,
"via_device_id": None,
},
],
@ -921,23 +939,40 @@ async def test_update(hass, registry, update_events):
updated_entry = registry.async_update_device(
entry.id,
area_id="12345A",
configuration_url="configuration_url",
disabled_by=device_registry.DeviceEntryDisabler.USER,
entry_type=device_registry.DeviceEntryType.SERVICE,
hw_version="hw_version",
manufacturer="Test Producer",
model="Test Model",
name_by_user="Test Friendly Name",
name="name",
new_identifiers=new_identifiers,
suggested_area="suggested_area",
sw_version="version",
via_device_id="98765B",
disabled_by=device_registry.DeviceEntryDisabler.USER,
)
assert mock_save.call_count == 1
assert updated_entry != entry
assert updated_entry.area_id == "12345A"
assert updated_entry.manufacturer == "Test Producer"
assert updated_entry.model == "Test Model"
assert updated_entry.name_by_user == "Test Friendly Name"
assert updated_entry.identifiers == new_identifiers
assert updated_entry.via_device_id == "98765B"
assert updated_entry.disabled_by is device_registry.DeviceEntryDisabler.USER
assert updated_entry == device_registry.DeviceEntry(
area_id="12345A",
config_entries={"1234"},
configuration_url="configuration_url",
connections={("mac", "12:34:56:ab:cd:ef")},
disabled_by=device_registry.DeviceEntryDisabler.USER,
entry_type=device_registry.DeviceEntryType.SERVICE,
hw_version="hw_version",
id=entry.id,
identifiers={("bla", "321"), ("hue", "654")},
manufacturer="Test Producer",
model="Test Model",
name_by_user="Test Friendly Name",
name="name",
suggested_area="suggested_area",
sw_version="version",
via_device_id="98765B",
)
assert registry.async_get_device({("hue", "456")}) is None
assert registry.async_get_device({("bla", "123")}) is None
@ -964,11 +999,17 @@ async def test_update(hass, registry, update_events):
assert update_events[1]["device_id"] == entry.id
assert update_events[1]["changes"] == {
"area_id": None,
"configuration_url": None,
"disabled_by": None,
"entry_type": None,
"hw_version": None,
"identifiers": {("bla", "123"), ("hue", "456")},
"manufacturer": None,
"model": None,
"name": None,
"name_by_user": None,
"suggested_area": None,
"sw_version": None,
"via_device_id": None,
}
@ -1036,62 +1077,6 @@ async def test_update_remove_config_entries(hass, registry, update_events):
assert "changes" not in update_events[4]
async def test_update_sw_version(hass, registry, update_events):
"""Verify that we can update software 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.sw_version
sw_version = "0x20020263"
with patch.object(registry, "async_schedule_save") as mock_save:
updated_entry = registry.async_update_device(entry.id, sw_version=sw_version)
assert mock_save.call_count == 1
assert updated_entry != entry
assert updated_entry.sw_version == sw_version
await hass.async_block_till_done()
assert len(update_events) == 2
assert update_events[0]["action"] == "create"
assert update_events[0]["device_id"] == entry.id
assert "changes" not in update_events[0]
assert update_events[1]["action"] == "update"
assert update_events[1]["device_id"] == entry.id
assert update_events[1]["changes"] == {"sw_version": None}
async def test_update_hw_version(hass, registry, update_events):
"""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
await hass.async_block_till_done()
assert len(update_events) == 2
assert update_events[0]["action"] == "create"
assert update_events[0]["device_id"] == entry.id
assert "changes" not in update_events[0]
assert update_events[1]["action"] == "update"
assert update_events[1]["device_id"] == entry.id
assert update_events[1]["changes"] == {"hw_version": None}
async def test_update_suggested_area(hass, registry, area_registry, update_events):
"""Verify that we can update the suggested area version of a device."""
entry = registry.async_get_or_create(