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

Adjust device_registry tests which create devices (#98215)

pull/95121/head
Erik Montnemery 2023-08-10 19:28:33 +02:00 committed by GitHub
parent c7b4d4f361
commit 4e8b81370e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 218 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

330
tests/helpers/test_device_registry.py
View File

@ -34,15 +34,24 @@ def update_events(hass):
return events
@pytest.fixture
def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Create a mock config entry and add it to hass."""
entry = MockConfigEntry(title=None)
entry.add_to_hass(hass)
return entry
async def test_get_or_create_returns_same_entry(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
area_registry: ar.AreaRegistry,
mock_config_entry: MockConfigEntry,
update_events,
) -> None:
"""Make sure we do not duplicate entries."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
sw_version="sw-version",
@ -52,7 +61,7 @@ async def test_get_or_create_returns_same_entry(
suggested_area="Game Room",
)
entry2 = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "11:22:33:66:77:88")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
@ -60,7 +69,7 @@ async def test_get_or_create_returns_same_entry(
suggested_area="Game Room",
)
entry3 = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
@ -99,17 +108,18 @@ async def test_get_or_create_returns_same_entry(
async def test_requirement_for_identifier_or_connection(
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Make sure we do require some descriptor of device."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers=set(),
manufacturer="manufacturer",
model="model",
)
entry2 = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections=set(),
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
@ -122,7 +132,7 @@ async def test_requirement_for_identifier_or_connection(
with pytest.raises(HomeAssistantError):
device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections=set(),
identifiers=set(),
manufacturer="manufacturer",
@ -130,24 +140,31 @@ async def test_requirement_for_identifier_or_connection(
)
async def test_multiple_config_entries(device_registry: dr.DeviceRegistry) -> None:
async def test_multiple_config_entries(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Make sure we do not get duplicate entries."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
entry = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
model="model",
)
entry2 = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
model="model",
)
entry3 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
@ -157,12 +174,14 @@ async def test_multiple_config_entries(device_registry: dr.DeviceRegistry) -> No
assert len(device_registry.devices) == 1
assert entry.id == entry2.id
assert entry.id == entry3.id
assert entry2.config_entries == {"123", "456"}
assert entry2.config_entries == {config_entry_1.entry_id, config_entry_2.entry_id}
@pytest.mark.parametrize("load_registries", [False])
async def test_loading_from_storage(
hass: HomeAssistant, hass_storage: dict[str, Any]
hass: HomeAssistant,
hass_storage: dict[str, Any],
mock_config_entry: MockConfigEntry,
) -> None:
"""Test loading stored devices on start."""
hass_storage[dr.STORAGE_KEY] = {
@ -172,7 +191,7 @@ async def test_loading_from_storage(
"devices": [
{
"area_id": "12345A",
"config_entries": ["1234"],
"config_entries": [mock_config_entry.entry_id],
"configuration_url": "https://example.com/config",
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": dr.DeviceEntryDisabler.USER,
@ -190,7 +209,7 @@ async def test_loading_from_storage(
],
"deleted_devices": [
{
"config_entries": ["1234"],
"config_entries": [mock_config_entry.entry_id],
"connections": [["Zigbee", "23.45.67.89.01"]],
"id": "bcdefghijklmn",
"identifiers": [["serial", "34:56:AB:CD:EF:12"]],
@ -206,7 +225,7 @@ async def test_loading_from_storage(
assert len(registry.deleted_devices) == 1
entry = registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={("Zigbee", "01.23.45.67.89")},
identifiers={("serial", "12:34:56:AB:CD:EF")},
manufacturer="manufacturer",
@ -214,7 +233,7 @@ async def test_loading_from_storage(
)
assert entry == dr.DeviceEntry(
area_id="12345A",
config_entries={"1234"},
config_entries={mock_config_entry.entry_id},
configuration_url="https://example.com/config",
connections={("Zigbee", "01.23.45.67.89")},
disabled_by=dr.DeviceEntryDisabler.USER,
@ -235,14 +254,14 @@ async def test_loading_from_storage(
# Restore a device, id should be reused from the deleted device entry
entry = registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={("Zigbee", "23.45.67.89.01")},
identifiers={("serial", "34:56:AB:CD:EF:12")},
manufacturer="manufacturer",
model="model",
)
assert entry == dr.DeviceEntry(
config_entries={"1234"},
config_entries={mock_config_entry.entry_id},
connections={("Zigbee", "23.45.67.89.01")},
id="bcdefghijklmn",
identifiers={("serial", "34:56:AB:CD:EF:12")},
@ -257,7 +276,9 @@ async def test_loading_from_storage(
@pytest.mark.parametrize("load_registries", [False])
async def test_migration_1_1_to_1_3(
hass: HomeAssistant, hass_storage: dict[str, Any]
hass: HomeAssistant,
hass_storage: dict[str, Any],
mock_config_entry: MockConfigEntry,
) -> None:
"""Test migration from version 1.1 to 1.3."""
hass_storage[dr.STORAGE_KEY] = {
@ -266,7 +287,7 @@ async def test_migration_1_1_to_1_3(
"data": {
"devices": [
{
"config_entries": ["1234"],
"config_entries": [mock_config_entry.entry_id],
"connections": [["Zigbee", "01.23.45.67.89"]],
"entry_type": "service",
"id": "abcdefghijklm",
@ -310,7 +331,7 @@ async def test_migration_1_1_to_1_3(
# Test data was loaded
entry = registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={("Zigbee", "01.23.45.67.89")},
identifiers={("serial", "12:34:56:AB:CD:EF")},
)
@ -318,7 +339,7 @@ async def test_migration_1_1_to_1_3(
# Update to trigger a store
entry = registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={("Zigbee", "01.23.45.67.89")},
identifiers={("serial", "12:34:56:AB:CD:EF")},
sw_version="new_version",
@ -335,7 +356,7 @@ async def test_migration_1_1_to_1_3(
"devices": [
{
"area_id": None,
"config_entries": ["1234"],
"config_entries": [mock_config_entry.entry_id],
"configuration_url": None,
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": None,
@ -383,7 +404,9 @@ async def test_migration_1_1_to_1_3(
@pytest.mark.parametrize("load_registries", [False])
async def test_migration_1_2_to_1_3(
hass: HomeAssistant, hass_storage: dict[str, Any]
hass: HomeAssistant,
hass_storage: dict[str, Any],
mock_config_entry: MockConfigEntry,
) -> None:
"""Test migration from version 1.2 to 1.3."""
hass_storage[dr.STORAGE_KEY] = {
@ -394,7 +417,7 @@ async def test_migration_1_2_to_1_3(
"devices": [
{
"area_id": None,
"config_entries": ["1234"],
"config_entries": [mock_config_entry.entry_id],
"configuration_url": None,
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": None,
@ -434,7 +457,7 @@ async def test_migration_1_2_to_1_3(
# Test data was loaded
entry = registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={("Zigbee", "01.23.45.67.89")},
identifiers={("serial", "12:34:56:AB:CD:EF")},
)
@ -442,7 +465,7 @@ async def test_migration_1_2_to_1_3(
# Update to trigger a store
entry = registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={("Zigbee", "01.23.45.67.89")},
identifiers={("serial", "12:34:56:AB:CD:EF")},
sw_version="new_version",
@ -460,7 +483,7 @@ async def test_migration_1_2_to_1_3(
"devices": [
{
"area_id": None,
"config_entries": ["1234"],
"config_entries": [mock_config_entry.entry_id],
"configuration_url": None,
"connections": [["Zigbee", "01.23.45.67.89"]],
"disabled_by": None,
@ -502,22 +525,27 @@ async def test_removing_config_entries(
hass: HomeAssistant, device_registry: dr.DeviceRegistry, update_events
) -> None:
"""Make sure we do not get duplicate entries."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
entry = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
model="model",
)
entry2 = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
model="model",
)
entry3 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "34:56:78:CD:EF:12")},
identifiers={("bridgeid", "4567")},
manufacturer="manufacturer",
@ -527,15 +555,15 @@ async def test_removing_config_entries(
assert len(device_registry.devices) == 2
assert entry.id == entry2.id
assert entry.id != entry3.id
assert entry2.config_entries == {"123", "456"}
assert entry2.config_entries == {config_entry_1.entry_id, config_entry_2.entry_id}
device_registry.async_clear_config_entry("123")
device_registry.async_clear_config_entry(config_entry_1.entry_id)
entry = device_registry.async_get_device(identifiers={("bridgeid", "0123")})
entry3_removed = device_registry.async_get_device(
identifiers={("bridgeid", "4567")}
)
assert entry.config_entries == {"456"}
assert entry.config_entries == {config_entry_2.entry_id}
assert entry3_removed is None
await hass.async_block_till_done()
@ -546,13 +574,15 @@ async def test_removing_config_entries(
assert "changes" not in update_events[0]
assert update_events[1]["action"] == "update"
assert update_events[1]["device_id"] == entry2.id
assert update_events[1]["changes"] == {"config_entries": {"123"}}
assert update_events[1]["changes"] == {"config_entries": {config_entry_1.entry_id}}
assert update_events[2]["action"] == "create"
assert update_events[2]["device_id"] == entry3.id
assert "changes" not in update_events[2]
assert update_events[3]["action"] == "update"
assert update_events[3]["device_id"] == entry.id
assert update_events[3]["changes"] == {"config_entries": {"456", "123"}}
assert update_events[3]["changes"] == {
"config_entries": {config_entry_1.entry_id, config_entry_2.entry_id}
}
assert update_events[4]["action"] == "remove"
assert update_events[4]["device_id"] == entry3.id
assert "changes" not in update_events[4]
@ -562,22 +592,27 @@ async def test_deleted_device_removing_config_entries(
hass: HomeAssistant, device_registry: dr.DeviceRegistry, update_events
) -> None:
"""Make sure we do not get duplicate entries."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
entry = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
model="model",
)
entry2 = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
model="model",
)
entry3 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "34:56:78:CD:EF:12")},
identifiers={("bridgeid", "4567")},
manufacturer="manufacturer",
@ -588,7 +623,7 @@ async def test_deleted_device_removing_config_entries(
assert len(device_registry.deleted_devices) == 0
assert entry.id == entry2.id
assert entry.id != entry3.id
assert entry2.config_entries == {"123", "456"}
assert entry2.config_entries == {config_entry_1.entry_id, config_entry_2.entry_id}
device_registry.async_remove_device(entry.id)
device_registry.async_remove_device(entry3.id)
@ -603,7 +638,7 @@ async def test_deleted_device_removing_config_entries(
assert "changes" not in update_events[0]
assert update_events[1]["action"] == "update"
assert update_events[1]["device_id"] == entry2.id
assert update_events[1]["changes"] == {"config_entries": {"123"}}
assert update_events[1]["changes"] == {"config_entries": {config_entry_1.entry_id}}
assert update_events[2]["action"] == "create"
assert update_events[2]["device_id"] == entry3.id
assert "changes" not in update_events[2]["device_id"]
@ -614,11 +649,11 @@ async def test_deleted_device_removing_config_entries(
assert update_events[4]["device_id"] == entry3.id
assert "changes" not in update_events[4]
device_registry.async_clear_config_entry("123")
device_registry.async_clear_config_entry(config_entry_1.entry_id)
assert len(device_registry.devices) == 0
assert len(device_registry.deleted_devices) == 2
device_registry.async_clear_config_entry("456")
device_registry.async_clear_config_entry(config_entry_2.entry_id)
assert len(device_registry.devices) == 0
assert len(device_registry.deleted_devices) == 2
@ -628,7 +663,7 @@ async def test_deleted_device_removing_config_entries(
# Re-add, expect to keep the device id
entry2 = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
@ -644,7 +679,7 @@ async def test_deleted_device_removing_config_entries(
# Re-add, expect to get a new device id after the purge
entry4 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
@ -653,10 +688,12 @@ async def test_deleted_device_removing_config_entries(
assert entry3.id != entry4.id
async def test_removing_area_id(device_registry: dr.DeviceRegistry) -> None:
async def test_removing_area_id(
device_registry: dr.DeviceRegistry, mock_config_entry: MockConfigEntry
) -> None:
"""Make sure we can clear area id."""
entry = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
@ -672,10 +709,17 @@ async def test_removing_area_id(device_registry: dr.DeviceRegistry) -> None:
assert entry_w_area != entry_wo_area
async def test_specifying_via_device_create(device_registry: dr.DeviceRegistry) -> None:
async def test_specifying_via_device_create(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test specifying a via_device and removal of the hub device."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
via = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("hue", "0123")},
manufacturer="manufacturer",
@ -683,7 +727,7 @@ async def test_specifying_via_device_create(device_registry: dr.DeviceRegistry)
)
light = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections=set(),
identifiers={("hue", "456")},
manufacturer="manufacturer",
@ -698,10 +742,17 @@ async def test_specifying_via_device_create(device_registry: dr.DeviceRegistry)
assert light.via_device_id is None
async def test_specifying_via_device_update(device_registry: dr.DeviceRegistry) -> None:
async def test_specifying_via_device_update(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test specifying a via_device and updating."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
light = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections=set(),
identifiers={("hue", "456")},
manufacturer="manufacturer",
@ -712,7 +763,7 @@ async def test_specifying_via_device_update(device_registry: dr.DeviceRegistry)
assert light.via_device_id is None
via = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("hue", "0123")},
manufacturer="manufacturer",
@ -720,7 +771,7 @@ async def test_specifying_via_device_update(device_registry: dr.DeviceRegistry)
)
light = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections=set(),
identifiers={("hue", "456")},
manufacturer="manufacturer",
@ -735,8 +786,19 @@ async def test_loading_saving_data(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test that we load/save data correctly."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
config_entry_3 = MockConfigEntry()
config_entry_3.add_to_hass(hass)
config_entry_4 = MockConfigEntry()
config_entry_4.add_to_hass(hass)
config_entry_5 = MockConfigEntry()
config_entry_5.add_to_hass(hass)
orig_via = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("hue", "0123")},
manufacturer="manufacturer",
@ -747,7 +809,7 @@ async def test_loading_saving_data(
)
orig_light = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections=set(),
identifiers={("hue", "456")},
manufacturer="manufacturer",
@ -757,7 +819,7 @@ async def test_loading_saving_data(
)
orig_light2 = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections=set(),
identifiers={("hue", "789")},
manufacturer="manufacturer",
@ -768,7 +830,7 @@ async def test_loading_saving_data(
device_registry.async_remove_device(orig_light2.id)
orig_light3 = device_registry.async_get_or_create(
config_entry_id="789",
config_entry_id=config_entry_3.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "34:56:AB:CD:EF:12")},
identifiers={("hue", "abc")},
manufacturer="manufacturer",
@ -776,7 +838,7 @@ async def test_loading_saving_data(
)
device_registry.async_get_or_create(
config_entry_id="abc",
config_entry_id=config_entry_4.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "34:56:AB:CD:EF:12")},
identifiers={("abc", "123")},
manufacturer="manufacturer",
@ -786,7 +848,7 @@ async def test_loading_saving_data(
device_registry.async_remove_device(orig_light3.id)
orig_light4 = device_registry.async_get_or_create(
config_entry_id="789",
config_entry_id=config_entry_3.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "34:56:AB:CD:EF:12")},
identifiers={("hue", "abc")},
manufacturer="manufacturer",
@ -797,7 +859,7 @@ async def test_loading_saving_data(
assert orig_light4.id == orig_light3.id
orig_kitchen_light = device_registry.async_get_or_create(
config_entry_id="999",
config_entry_id=config_entry_5.entry_id,
connections=set(),
identifiers={("hue", "999")},
manufacturer="manufacturer",
@ -851,10 +913,12 @@ async def test_loading_saving_data(
assert orig_kitchen_light_witout_suggested_area == new_kitchen_light
async def test_no_unnecessary_changes(device_registry: dr.DeviceRegistry) -> None:
async def test_no_unnecessary_changes(
device_registry: dr.DeviceRegistry, mock_config_entry: MockConfigEntry
) -> None:
"""Make sure we do not consider devices changes."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={("ethernet", "12:34:56:78:90:AB:CD:EF")},
identifiers={("hue", "456"), ("bla", "123")},
)
@ -862,22 +926,24 @@ async def test_no_unnecessary_changes(device_registry: dr.DeviceRegistry) -> Non
"homeassistant.helpers.device_registry.DeviceRegistry.async_schedule_save"
) as mock_save:
entry2 = device_registry.async_get_or_create(
config_entry_id="1234", identifiers={("hue", "456")}
config_entry_id=mock_config_entry.entry_id, identifiers={("hue", "456")}
)
assert entry.id == entry2.id
assert len(mock_save.mock_calls) == 0
async def test_format_mac(device_registry: dr.DeviceRegistry) -> None:
async def test_format_mac(
device_registry: dr.DeviceRegistry, mock_config_entry: MockConfigEntry
) -> None:
"""Make sure we normalize mac addresses."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
for mac in ["123456ABCDEF", "123456abcdef", "12:34:56:ab:cd:ef", "1234.56ab.cdef"]:
test_entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, mac)},
)
assert test_entry.id == entry.id, mac
@ -895,18 +961,21 @@ async def test_format_mac(device_registry: dr.DeviceRegistry) -> None:
"123.456.abc.def", # too many .
]:
invalid_mac_entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, invalid)},
)
assert list(invalid_mac_entry.connections)[0][1] == invalid
async def test_update(
hass: HomeAssistant, device_registry: dr.DeviceRegistry, update_events
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
update_events,
) -> None:
"""Verify that we can update some attributes of a device."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("hue", "456"), ("bla", "123")},
)
@ -936,7 +1005,7 @@ async def test_update(
assert updated_entry != entry
assert updated_entry == dr.DeviceEntry(
area_id="12345A",
config_entries={"1234"},
config_entries={mock_config_entry.entry_id},
configuration_url="https://example.com/config",
connections={("mac", "12:34:56:ab:cd:ef")},
disabled_by=dr.DeviceEntryDisabler.USER,
@ -1001,22 +1070,27 @@ async def test_update_remove_config_entries(
hass: HomeAssistant, device_registry: dr.DeviceRegistry, update_events
) -> None:
"""Make sure we do not get duplicate entries."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
entry = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
model="model",
)
entry2 = device_registry.async_get_or_create(
config_entry_id="456",
config_entry_id=config_entry_2.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
model="model",
)
entry3 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "34:56:78:CD:EF:12")},
identifiers={("bridgeid", "4567")},
manufacturer="manufacturer",
@ -1026,16 +1100,16 @@ async def test_update_remove_config_entries(
assert len(device_registry.devices) == 2
assert entry.id == entry2.id
assert entry.id != entry3.id
assert entry2.config_entries == {"123", "456"}
assert entry2.config_entries == {config_entry_1.entry_id, config_entry_2.entry_id}
updated_entry = device_registry.async_update_device(
entry2.id, remove_config_entry_id="123"
entry2.id, remove_config_entry_id=config_entry_1.entry_id
)
removed_entry = device_registry.async_update_device(
entry3.id, remove_config_entry_id="123"
entry3.id, remove_config_entry_id=config_entry_1.entry_id
)
assert updated_entry.config_entries == {"456"}
assert updated_entry.config_entries == {config_entry_2.entry_id}
assert removed_entry is None
removed_entry = device_registry.async_get_device(identifiers={("bridgeid", "4567")})
@ -1050,13 +1124,15 @@ async def test_update_remove_config_entries(
assert "changes" not in update_events[0]
assert update_events[1]["action"] == "update"
assert update_events[1]["device_id"] == entry2.id
assert update_events[1]["changes"] == {"config_entries": {"123"}}
assert update_events[1]["changes"] == {"config_entries": {config_entry_1.entry_id}}
assert update_events[2]["action"] == "create"
assert update_events[2]["device_id"] == entry3.id
assert "changes" not in update_events[2]
assert update_events[3]["action"] == "update"
assert update_events[3]["device_id"] == entry.id
assert update_events[3]["changes"] == {"config_entries": {"456", "123"}}
assert update_events[3]["changes"] == {
"config_entries": {config_entry_1.entry_id, config_entry_2.entry_id}
}
assert update_events[4]["action"] == "remove"
assert update_events[4]["device_id"] == entry3.id
assert "changes" not in update_events[4]
@ -1066,11 +1142,12 @@ async def test_update_suggested_area(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
area_registry: ar.AreaRegistry,
mock_config_entry: MockConfigEntry,
update_events,
) -> None:
"""Verify that we can update the suggested area version of a device."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bla", "123")},
)
@ -1122,6 +1199,8 @@ async def test_cleanup_device_registry(
"""Test cleanup works."""
config_entry = MockConfigEntry(domain="hue")
config_entry.add_to_hass(hass)
ghost_config_entry = MockConfigEntry()
ghost_config_entry.add_to_hass(hass)
d1 = device_registry.async_get_or_create(
identifiers={("hue", "d1")}, config_entry_id=config_entry.entry_id
@ -1133,14 +1212,17 @@ async def test_cleanup_device_registry(
identifiers={("hue", "d3")}, config_entry_id=config_entry.entry_id
)
device_registry.async_get_or_create(
identifiers={("something", "d4")}, config_entry_id="non_existing"
identifiers={("something", "d4")}, config_entry_id=ghost_config_entry.entry_id
)
# Remove the config entry without triggering the normal cleanup
hass.config_entries._entries.pop(ghost_config_entry.entry_id)
ent_reg = er.async_get(hass)
ent_reg.async_get_or_create("light", "hue", "e1", device_id=d1.id)
ent_reg.async_get_or_create("light", "hue", "e2", device_id=d1.id)
ent_reg.async_get_or_create("light", "hue", "e3", device_id=d3.id)
# Manual cleanup should detect the orphaned config entry
dr.async_cleanup(hass, device_registry, ent_reg)
assert device_registry.async_get_device(identifiers={("hue", "d1")}) is not None
@ -1233,11 +1315,14 @@ async def test_cleanup_entity_registry_change(hass: HomeAssistant) -> None:
async def test_restore_device(
hass: HomeAssistant, device_registry: dr.DeviceRegistry, update_events
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
update_events,
) -> None:
"""Make sure device id is stable."""
entry = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
@ -1253,14 +1338,14 @@ async def test_restore_device(
assert len(device_registry.deleted_devices) == 1
entry2 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "34:56:78:CD:EF:12")},
identifiers={("bridgeid", "4567")},
manufacturer="manufacturer",
model="model",
)
entry3 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
manufacturer="manufacturer",
@ -1294,11 +1379,14 @@ async def test_restore_device(
async def test_restore_simple_device(
hass: HomeAssistant, device_registry: dr.DeviceRegistry, update_events
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
update_events,
) -> None:
"""Make sure device id is stable."""
entry = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
)
@ -1312,12 +1400,12 @@ async def test_restore_simple_device(
assert len(device_registry.deleted_devices) == 1
entry2 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "34:56:78:CD:EF:12")},
identifiers={("bridgeid", "4567")},
)
entry3 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
)
@ -1348,8 +1436,13 @@ async def test_restore_shared_device(
hass: HomeAssistant, device_registry: dr.DeviceRegistry, update_events
) -> None:
"""Make sure device id is stable for shared devices."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
entry = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("entry_123", "0123")},
manufacturer="manufacturer",
@ -1360,7 +1453,7 @@ async def test_restore_shared_device(
assert len(device_registry.deleted_devices) == 0
device_registry.async_get_or_create(
config_entry_id="234",
config_entry_id=config_entry_2.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("entry_234", "2345")},
manufacturer="manufacturer",
@ -1376,7 +1469,7 @@ async def test_restore_shared_device(
assert len(device_registry.deleted_devices) == 1
entry2 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("entry_123", "0123")},
manufacturer="manufacturer",
@ -1394,7 +1487,7 @@ async def test_restore_shared_device(
device_registry.async_remove_device(entry.id)
entry3 = device_registry.async_get_or_create(
config_entry_id="234",
config_entry_id=config_entry_2.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("entry_234", "2345")},
manufacturer="manufacturer",
@ -1410,7 +1503,7 @@ async def test_restore_shared_device(
assert isinstance(entry3.identifiers, set)
entry4 = device_registry.async_get_or_create(
config_entry_id="123",
config_entry_id=config_entry_1.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("entry_123", "0123")},
manufacturer="manufacturer",
@ -1434,7 +1527,7 @@ async def test_restore_shared_device(
assert update_events[1]["action"] == "update"
assert update_events[1]["device_id"] == entry.id
assert update_events[1]["changes"] == {
"config_entries": {"123"},
"config_entries": {config_entry_1.entry_id},
"identifiers": {("entry_123", "0123")},
}
assert update_events[2]["action"] == "remove"
@ -1452,17 +1545,18 @@ async def test_restore_shared_device(
assert update_events[6]["action"] == "update"
assert update_events[6]["device_id"] == entry.id
assert update_events[6]["changes"] == {
"config_entries": {"234"},
"config_entries": {config_entry_2.entry_id},
"identifiers": {("entry_234", "2345")},
}
async def test_get_or_create_empty_then_set_default_values(
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating an entry, then setting default name, model, manufacturer."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
assert entry.name is None
@ -1470,7 +1564,7 @@ async def test_get_or_create_empty_then_set_default_values(
assert entry.manufacturer is None
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
default_name="default name 1",
default_model="default model 1",
@ -1481,7 +1575,7 @@ async def test_get_or_create_empty_then_set_default_values(
assert entry.manufacturer == "default manufacturer 1"
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
default_name="default name 2",
default_model="default model 2",
@ -1494,10 +1588,11 @@ async def test_get_or_create_empty_then_set_default_values(
async def test_get_or_create_empty_then_update(
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating an entry, then setting name, model, manufacturer."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
assert entry.name is None
@ -1505,7 +1600,7 @@ async def test_get_or_create_empty_then_update(
assert entry.manufacturer is None
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
name="name 1",
model="model 1",
@ -1516,7 +1611,7 @@ async def test_get_or_create_empty_then_update(
assert entry.manufacturer == "manufacturer 1"
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
default_name="default name 1",
default_model="default model 1",
@ -1529,10 +1624,11 @@ async def test_get_or_create_empty_then_update(
async def test_get_or_create_sets_default_values(
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating an entry, then setting default name, model, manufacturer."""
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
default_name="default name 1",
default_model="default model 1",
@ -1543,7 +1639,7 @@ async def test_get_or_create_sets_default_values(
assert entry.manufacturer == "default manufacturer 1"
entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
default_name="default name 2",
default_model="default model 2",
@ -1555,13 +1651,15 @@ async def test_get_or_create_sets_default_values(
async def test_verify_suggested_area_does_not_overwrite_area_id(
device_registry: dr.DeviceRegistry, area_registry: ar.AreaRegistry
device_registry: dr.DeviceRegistry,
area_registry: ar.AreaRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Make sure suggested area does not override a set area id."""
game_room_area = area_registry.async_create("Game Room")
original_entry = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
sw_version="sw-version",
@ -1576,7 +1674,7 @@ async def test_verify_suggested_area_does_not_overwrite_area_id(
assert entry.area_id == game_room_area.id
entry2 = device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=mock_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
identifiers={("bridgeid", "0123")},
sw_version="sw-version",
@ -1713,16 +1811,21 @@ async def test_device_info_configuration_url_validation(
expectation,
) -> None:
"""Test configuration URL of device info is properly validated."""
config_entry_1 = MockConfigEntry()
config_entry_1.add_to_hass(hass)
config_entry_2 = MockConfigEntry()
config_entry_2.add_to_hass(hass)
with expectation:
device_registry.async_get_or_create(
config_entry_id="1234",
config_entry_id=config_entry_1.entry_id,
identifiers={("something", "1234")},
name="name",
configuration_url=configuration_url,
)
update_device = device_registry.async_get_or_create(
config_entry_id="5678",
config_entry_id=config_entry_2.entry_id,
identifiers={("something", "5678")},
name="name",
)
@ -1734,7 +1837,9 @@ async def test_device_info_configuration_url_validation(
@pytest.mark.parametrize("load_registries", [False])
async def test_loading_invalid_configuration_url_from_storage(
hass: HomeAssistant, hass_storage: dict[str, Any]
hass: HomeAssistant,
hass_storage: dict[str, Any],
mock_config_entry: MockConfigEntry,
) -> None:
"""Test loading stored devices with an invalid URL."""
hass_storage[dr.STORAGE_KEY] = {
@ -1768,6 +1873,7 @@ async def test_loading_invalid_configuration_url_from_storage(
registry = dr.async_get(hass)
assert len(registry.devices) == 1
entry = registry.async_get_or_create(
config_entry_id="1234", identifiers={("serial", "12:34:56:AB:CD:EF")}
config_entry_id=mock_config_entry.entry_id,
identifiers={("serial", "12:34:56:AB:CD:EF")},
)
assert entry.configuration_url == "invalid"