"""Tests for the Device Registry.""" from collections.abc import Iterable from contextlib import AbstractContextManager, nullcontext from datetime import datetime from functools import partial import time from typing import Any from unittest.mock import patch from freezegun.api import FrozenDateTimeFactory import pytest from yarl import URL from homeassistant import config_entries from homeassistant.const import EVENT_HOMEASSISTANT_STARTED from homeassistant.core import CoreState, HomeAssistant, ReleaseChannel from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import ( area_registry as ar, device_registry as dr, entity_registry as er, ) from homeassistant.util.dt import utcnow from tests.common import ( MockConfigEntry, async_capture_events, flush_store, help_test_all, import_and_test_deprecated_constant_enum, ) @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, ) -> None: """Make sure we do not duplicate entries.""" update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) entry = device_registry.async_get_or_create( 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", name="name", manufacturer="manufacturer", model="model", suggested_area="Game Room", ) entry2 = device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "11:22:33:66:77:88")}, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", suggested_area="Game Room", ) entry3 = device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, ) game_room_area = area_registry.async_get_area_by_name("Game Room") assert game_room_area is not None assert len(area_registry.areas) == 1 assert len(device_registry.devices) == 1 assert entry.area_id == game_room_area.id assert entry.id == entry2.id assert entry.id == entry3.id assert entry.identifiers == {("bridgeid", "0123")} assert entry2.area_id == game_room_area.id assert entry3.manufacturer == "manufacturer" assert entry3.model == "model" assert entry3.name == "name" assert entry3.sw_version == "sw-version" assert entry3.suggested_area == "Game Room" assert entry3.area_id == game_room_area.id await hass.async_block_till_done() # Only 2 update events. The third entry did not generate any changes. assert len(update_events) == 2 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "update", "device_id": entry.id, "changes": {"connections": {("mac", "12:34:56:ab:cd:ef")}}, } 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=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=mock_config_entry.entry_id, connections=set(), identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) assert len(device_registry.devices) == 2 assert entry assert entry2 with pytest.raises(HomeAssistantError): device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections=set(), identifiers=set(), manufacturer="manufacturer", model="model", ) 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=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=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=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) assert len(device_registry.devices) == 1 assert entry.id == entry2.id assert entry.id == entry3.id assert entry2.config_entries == {config_entry_1.entry_id, config_entry_2.entry_id} assert entry2.primary_config_entry == config_entry_1.entry_id assert entry3.config_entries == {config_entry_1.entry_id, config_entry_2.entry_id} assert entry3.primary_config_entry == config_entry_1.entry_id @pytest.mark.parametrize("load_registries", [False]) @pytest.mark.usefixtures("freezer") async def test_loading_from_storage( hass: HomeAssistant, hass_storage: dict[str, Any], mock_config_entry: MockConfigEntry, ) -> None: """Test loading stored devices on start.""" created_at = "2024-01-01T00:00:00+00:00" modified_at = "2024-02-01T00:00:00+00:00" hass_storage[dr.STORAGE_KEY] = { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "data": { "devices": [ { "area_id": "12345A", "config_entries": [mock_config_entry.entry_id], "configuration_url": "https://example.com/config", "connections": [["Zigbee", "01.23.45.67.89"]], "created_at": created_at, "disabled_by": dr.DeviceEntryDisabler.USER, "entry_type": dr.DeviceEntryType.SERVICE, "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": {"label1", "label2"}, "manufacturer": "manufacturer", "model": "model", "model_id": "model_id", "modified_at": modified_at, "name_by_user": "Test Friendly Name", "name": "name", "primary_config_entry": mock_config_entry.entry_id, "serial_number": "serial_no", "sw_version": "version", "via_device_id": None, } ], "deleted_devices": [ { "config_entries": [mock_config_entry.entry_id], "connections": [["Zigbee", "23.45.67.89.01"]], "created_at": created_at, "id": "bcdefghijklmn", "identifiers": [["serial", "3456ABCDEF12"]], "modified_at": modified_at, "orphaned_timestamp": None, } ], }, } await dr.async_load(hass) registry = dr.async_get(hass) assert len(registry.devices) == 1 assert len(registry.deleted_devices) == 1 assert registry.deleted_devices["bcdefghijklmn"] == dr.DeletedDeviceEntry( config_entries={mock_config_entry.entry_id}, connections={("Zigbee", "23.45.67.89.01")}, created_at=datetime.fromisoformat(created_at), id="bcdefghijklmn", identifiers={("serial", "3456ABCDEF12")}, modified_at=datetime.fromisoformat(modified_at), orphaned_timestamp=None, ) entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, manufacturer="manufacturer", model="model", ) assert entry == dr.DeviceEntry( area_id="12345A", config_entries={mock_config_entry.entry_id}, configuration_url="https://example.com/config", connections={("Zigbee", "01.23.45.67.89")}, created_at=datetime.fromisoformat(created_at), disabled_by=dr.DeviceEntryDisabler.USER, entry_type=dr.DeviceEntryType.SERVICE, hw_version="hw_version", id="abcdefghijklm", identifiers={("serial", "123456ABCDEF")}, labels={"label1", "label2"}, manufacturer="manufacturer", model="model", model_id="model_id", modified_at=datetime.fromisoformat(modified_at), name_by_user="Test Friendly Name", name="name", primary_config_entry=mock_config_entry.entry_id, serial_number="serial_no", 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=mock_config_entry.entry_id, connections={("Zigbee", "23.45.67.89.01")}, identifiers={("serial", "3456ABCDEF12")}, manufacturer="manufacturer", model="model", ) assert entry == dr.DeviceEntry( config_entries={mock_config_entry.entry_id}, connections={("Zigbee", "23.45.67.89.01")}, created_at=datetime.fromisoformat(created_at), id="bcdefghijklmn", identifiers={("serial", "3456ABCDEF12")}, manufacturer="manufacturer", model="model", modified_at=utcnow(), primary_config_entry=mock_config_entry.entry_id, ) assert entry.id == "bcdefghijklmn" assert isinstance(entry.config_entries, set) assert isinstance(entry.connections, set) assert isinstance(entry.identifiers, set) @pytest.mark.parametrize("load_registries", [False]) @pytest.mark.usefixtures("freezer") async def test_migration_1_1_to_1_7( hass: HomeAssistant, hass_storage: dict[str, Any], mock_config_entry: MockConfigEntry, ) -> None: """Test migration from version 1.1 to 1.7.""" hass_storage[dr.STORAGE_KEY] = { "version": 1, "minor_version": 1, "data": { "devices": [ { "config_entries": [mock_config_entry.entry_id], "connections": [["Zigbee", "01.23.45.67.89"]], "entry_type": "service", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "manufacturer": "manufacturer", "model": "model", "name": "name", "sw_version": "version", }, # Invalid entry type { "config_entries": [None], "connections": [], "entry_type": "INVALID_VALUE", "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "manufacturer": None, "model": None, "name": None, "sw_version": None, }, ], "deleted_devices": [ { "config_entries": ["123456"], "connections": [], "entry_type": "service", "id": "deletedid", "identifiers": [["serial", "123456ABCDFF"]], "manufacturer": "manufacturer", "model": "model", "name": "name", "sw_version": "version", } ], }, } await dr.async_load(hass) registry = dr.async_get(hass) # Test data was loaded entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, ) assert entry.id == "abcdefghijklm" # Update to trigger a store entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, sw_version="new_version", ) assert entry.id == "abcdefghijklm" # Check we store migrated data await flush_store(registry._store) assert hass_storage[dr.STORAGE_KEY] == { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": "service", "hw_version": None, "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": [], "manufacturer": "manufacturer", "model": "model", "model_id": None, "modified_at": utcnow().isoformat(), "name": "name", "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": [], "manufacturer": None, "model": None, "model_id": None, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "name": None, "primary_config_entry": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [ { "config_entries": ["123456"], "connections": [], "created_at": "1970-01-01T00:00:00+00:00", "id": "deletedid", "identifiers": [["serial", "123456ABCDFF"]], "modified_at": "1970-01-01T00:00:00+00:00", "orphaned_timestamp": None, } ], }, } @pytest.mark.parametrize("load_registries", [False]) @pytest.mark.usefixtures("freezer") async def test_migration_1_2_to_1_7( hass: HomeAssistant, hass_storage: dict[str, Any], mock_config_entry: MockConfigEntry, ) -> None: """Test migration from version 1.2 to 1.7.""" hass_storage[dr.STORAGE_KEY] = { "version": 1, "minor_version": 2, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "disabled_by": None, "entry_type": "service", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "manufacturer": "manufacturer", "model": "model", "modified_at": utcnow().isoformat(), "name": "name", "name_by_user": None, "sw_version": "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, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "name": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } await dr.async_load(hass) registry = dr.async_get(hass) # Test data was loaded entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, ) assert entry.id == "abcdefghijklm" # Update to trigger a store entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, sw_version="new_version", ) assert entry.id == "abcdefghijklm" # Check we store migrated data await flush_store(registry._store) assert hass_storage[dr.STORAGE_KEY] == { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": "service", "hw_version": None, "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": [], "manufacturer": "manufacturer", "model": "model", "model_id": None, "modified_at": utcnow().isoformat(), "name": "name", "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": [], "manufacturer": None, "model": None, "model_id": None, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "name": None, "primary_config_entry": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } @pytest.mark.parametrize("load_registries", [False]) @pytest.mark.usefixtures("freezer") async def test_migration_1_3_to_1_7( hass: HomeAssistant, hass_storage: dict[str, Any], mock_config_entry: MockConfigEntry, ) -> None: """Test migration from version 1.3 to 1.7.""" hass_storage[dr.STORAGE_KEY] = { "version": 1, "minor_version": 3, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "manufacturer": "manufacturer", "model": "model", "name": "name", "name_by_user": None, "sw_version": "version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "manufacturer": None, "model": None, "name_by_user": None, "name": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } await dr.async_load(hass) registry = dr.async_get(hass) # Test data was loaded entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, ) assert entry.id == "abcdefghijklm" # Update to trigger a store entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, sw_version="new_version", ) assert entry.id == "abcdefghijklm" # Check we store migrated data await flush_store(registry._store) assert hass_storage[dr.STORAGE_KEY] == { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": [], "manufacturer": "manufacturer", "model": "model", "model_id": None, "modified_at": utcnow().isoformat(), "name": "name", "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "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": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } @pytest.mark.parametrize("load_registries", [False]) @pytest.mark.usefixtures("freezer") async def test_migration_1_4_to_1_7( hass: HomeAssistant, hass_storage: dict[str, Any], mock_config_entry: MockConfigEntry, ) -> None: """Test migration from version 1.4 to 1.7.""" hass_storage[dr.STORAGE_KEY] = { "version": 1, "minor_version": 4, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "manufacturer": "manufacturer", "model": "model", "name": "name", "name_by_user": None, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "manufacturer": None, "model": None, "name_by_user": None, "name": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } await dr.async_load(hass) registry = dr.async_get(hass) # Test data was loaded entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, ) assert entry.id == "abcdefghijklm" # Update to trigger a store entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, sw_version="new_version", ) assert entry.id == "abcdefghijklm" # Check we store migrated data await flush_store(registry._store) assert hass_storage[dr.STORAGE_KEY] == { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": [], "manufacturer": "manufacturer", "model": "model", "model_id": None, "modified_at": utcnow().isoformat(), "name": "name", "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": [], "manufacturer": None, "model": None, "model_id": None, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "name": None, "primary_config_entry": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } @pytest.mark.parametrize("load_registries", [False]) @pytest.mark.usefixtures("freezer") async def test_migration_1_5_to_1_7( hass: HomeAssistant, hass_storage: dict[str, Any], mock_config_entry: MockConfigEntry, ) -> None: """Test migration from version 1.5 to 1.7.""" hass_storage[dr.STORAGE_KEY] = { "version": 1, "minor_version": 5, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": ["blah"], "manufacturer": "manufacturer", "model": "model", "name": "name", "name_by_user": None, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": ["blah"], "manufacturer": None, "model": None, "name_by_user": None, "name": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } await dr.async_load(hass) registry = dr.async_get(hass) # Test data was loaded entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, ) assert entry.id == "abcdefghijklm" # Update to trigger a store entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, sw_version="new_version", ) assert entry.id == "abcdefghijklm" # Check we store migrated data await flush_store(registry._store) assert hass_storage[dr.STORAGE_KEY] == { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": ["blah"], "manufacturer": "manufacturer", "model": "model", "name": "name", "model_id": None, "modified_at": utcnow().isoformat(), "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": ["blah"], "manufacturer": None, "model": None, "model_id": None, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "name": None, "primary_config_entry": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } @pytest.mark.parametrize("load_registries", [False]) @pytest.mark.usefixtures("freezer") async def test_migration_1_6_to_1_8( hass: HomeAssistant, hass_storage: dict[str, Any], mock_config_entry: MockConfigEntry, ) -> None: """Test migration from version 1.6 to 1.8.""" hass_storage[dr.STORAGE_KEY] = { "version": 1, "minor_version": 6, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": ["blah"], "manufacturer": "manufacturer", "model": "model", "name": "name", "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": ["blah"], "manufacturer": None, "model": None, "name_by_user": None, "primary_config_entry": None, "name": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } await dr.async_load(hass) registry = dr.async_get(hass) # Test data was loaded entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, ) assert entry.id == "abcdefghijklm" # Update to trigger a store entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, sw_version="new_version", ) assert entry.id == "abcdefghijklm" # Check we store migrated data await flush_store(registry._store) assert hass_storage[dr.STORAGE_KEY] == { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": ["blah"], "manufacturer": "manufacturer", "model": "model", "name": "name", "model_id": None, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": ["blah"], "manufacturer": None, "model": None, "model_id": None, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "name": None, "primary_config_entry": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } @pytest.mark.parametrize("load_registries", [False]) @pytest.mark.usefixtures("freezer") async def test_migration_1_7_to_1_8( hass: HomeAssistant, hass_storage: dict[str, Any], mock_config_entry: MockConfigEntry, ) -> None: """Test migration from version 1.7 to 1.8.""" hass_storage[dr.STORAGE_KEY] = { "version": 1, "minor_version": 7, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": ["blah"], "manufacturer": "manufacturer", "model": "model", "model_id": None, "name": "name", "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": ["blah"], "manufacturer": None, "model": None, "model_id": None, "name_by_user": None, "primary_config_entry": None, "name": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } await dr.async_load(hass) registry = dr.async_get(hass) # Test data was loaded entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, ) assert entry.id == "abcdefghijklm" # Update to trigger a store entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={("Zigbee", "01.23.45.67.89")}, identifiers={("serial", "123456ABCDEF")}, sw_version="new_version", ) assert entry.id == "abcdefghijklm" # Check we store migrated data await flush_store(registry._store) assert hass_storage[dr.STORAGE_KEY] == { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "key": dr.STORAGE_KEY, "data": { "devices": [ { "area_id": None, "config_entries": [mock_config_entry.entry_id], "configuration_url": None, "connections": [["Zigbee", "01.23.45.67.89"]], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": "service", "hw_version": "hw_version", "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": ["blah"], "manufacturer": "manufacturer", "model": "model", "name": "name", "model_id": None, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "primary_config_entry": mock_config_entry.entry_id, "serial_number": None, "sw_version": "new_version", "via_device_id": None, }, { "area_id": None, "config_entries": [None], "configuration_url": None, "connections": [], "created_at": "1970-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": None, "hw_version": None, "id": "invalid-entry-type", "identifiers": [["serial", "mock-id-invalid-entry"]], "labels": ["blah"], "manufacturer": None, "model": None, "model_id": None, "modified_at": "1970-01-01T00:00:00+00:00", "name_by_user": None, "name": None, "primary_config_entry": None, "serial_number": None, "sw_version": None, "via_device_id": None, }, ], "deleted_devices": [], }, } async def test_removing_config_entries( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Make sure we do not get duplicate entries.""" update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) 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=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=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=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "34:56:78:CD:EF:12")}, identifiers={("bridgeid", "4567")}, manufacturer="manufacturer", model="model", ) assert len(device_registry.devices) == 2 assert entry.id == entry2.id assert entry.id != entry3.id assert entry2.config_entries == {config_entry_1.entry_id, config_entry_2.entry_id} 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 == {config_entry_2.entry_id} assert entry3_removed is None await hass.async_block_till_done() assert len(update_events) == 5 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "update", "device_id": entry.id, "changes": { "config_entries": {config_entry_1.entry_id}, }, } assert update_events[2].data == { "action": "create", "device_id": entry3.id, } assert update_events[3].data == { "action": "update", "device_id": entry.id, "changes": { "config_entries": {config_entry_1.entry_id, config_entry_2.entry_id}, "primary_config_entry": config_entry_1.entry_id, }, } assert update_events[4].data == { "action": "remove", "device_id": entry3.id, } async def test_deleted_device_removing_config_entries( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Make sure we do not get duplicate entries.""" update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) 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=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=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=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "34:56:78:CD:EF:12")}, identifiers={("bridgeid", "4567")}, manufacturer="manufacturer", model="model", ) assert len(device_registry.devices) == 2 assert len(device_registry.deleted_devices) == 0 assert entry.id == entry2.id assert entry.id != entry3.id 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) assert len(device_registry.devices) == 0 assert len(device_registry.deleted_devices) == 2 await hass.async_block_till_done() assert len(update_events) == 5 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "update", "device_id": entry2.id, "changes": { "config_entries": {config_entry_1.entry_id}, }, } assert update_events[2].data == { "action": "create", "device_id": entry3.id, } assert update_events[3].data == { "action": "remove", "device_id": entry.id, } assert update_events[4].data == { "action": "remove", "device_id": entry3.id, } 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(config_entry_2.entry_id) assert len(device_registry.devices) == 0 assert len(device_registry.deleted_devices) == 2 # No event when a deleted device is purged await hass.async_block_till_done() assert len(update_events) == 5 # Re-add, expect to keep the device id entry2 = device_registry.async_get_or_create( 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", ) assert entry.id == entry2.id future_time = time.time() + dr.ORPHANED_DEVICE_KEEP_SECONDS + 1 with patch("time.time", return_value=future_time): device_registry.async_purge_expired_orphaned_devices() # Re-add, expect to get a new device id after the purge entry4 = device_registry.async_get_or_create( 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", ) assert entry3.id != entry4.id 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=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) entry_w_area = device_registry.async_update_device(entry.id, area_id="12345A") device_registry.async_clear_area_id("12345A") entry_wo_area = device_registry.async_get_device(identifiers={("bridgeid", "0123")}) assert not entry_wo_area.area_id assert entry_w_area != entry_wo_area 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=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("hue", "0123")}, manufacturer="manufacturer", model="via", ) light = device_registry.async_get_or_create( config_entry_id=config_entry_2.entry_id, connections=set(), identifiers={("hue", "456")}, manufacturer="manufacturer", model="light", via_device=("hue", "0123"), ) assert light.via_device_id == via.id device_registry.async_remove_device(via.id) light = device_registry.async_get_device(identifiers={("hue", "456")}) assert light.via_device_id is 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=config_entry_2.entry_id, connections=set(), identifiers={("hue", "456")}, manufacturer="manufacturer", model="light", via_device=("hue", "0123"), ) assert light.via_device_id is None via = device_registry.async_get_or_create( config_entry_id=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("hue", "0123")}, manufacturer="manufacturer", model="via", ) light = device_registry.async_get_or_create( config_entry_id=config_entry_2.entry_id, connections=set(), identifiers={("hue", "456")}, manufacturer="manufacturer", model="light", via_device=("hue", "0123"), ) assert light.via_device_id == via.id 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=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("hue", "0123")}, manufacturer="manufacturer", model="via", name="Original Name", sw_version="Orig SW 1", entry_type=None, ) orig_light = device_registry.async_get_or_create( config_entry_id=config_entry_2.entry_id, connections=set(), identifiers={("hue", "456")}, manufacturer="manufacturer", model="light", via_device=("hue", "0123"), disabled_by=dr.DeviceEntryDisabler.USER, ) orig_light2 = device_registry.async_get_or_create( config_entry_id=config_entry_2.entry_id, connections=set(), identifiers={("hue", "789")}, manufacturer="manufacturer", model="light", via_device=("hue", "0123"), ) device_registry.async_remove_device(orig_light2.id) orig_light3 = device_registry.async_get_or_create( config_entry_id=config_entry_3.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "34:56:AB:CD:EF:12")}, identifiers={("hue", "abc")}, manufacturer="manufacturer", model="light", ) device_registry.async_get_or_create( config_entry_id=config_entry_4.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "34:56:AB:CD:EF:12")}, identifiers={("abc", "123")}, manufacturer="manufacturer", model="light", ) device_registry.async_remove_device(orig_light3.id) orig_light4 = device_registry.async_get_or_create( config_entry_id=config_entry_3.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "34:56:AB:CD:EF:12")}, identifiers={("hue", "abc")}, manufacturer="manufacturer", model="light", entry_type=dr.DeviceEntryType.SERVICE, ) assert orig_light4.id == orig_light3.id orig_kitchen_light = device_registry.async_get_or_create( config_entry_id=config_entry_5.entry_id, connections=set(), identifiers={("hue", "999")}, manufacturer="manufacturer", model="light", via_device=("hue", "0123"), disabled_by=dr.DeviceEntryDisabler.USER, suggested_area="Kitchen", ) assert len(device_registry.devices) == 4 assert len(device_registry.deleted_devices) == 1 orig_via = device_registry.async_update_device( orig_via.id, area_id="mock-area-id", name_by_user="mock-name-by-user", labels={"mock-label1", "mock-label2"}, ) # Now load written data in new registry registry2 = dr.DeviceRegistry(hass) await flush_store(device_registry._store) await registry2.async_load() # Ensure same order assert list(device_registry.devices) == list(registry2.devices) assert list(device_registry.deleted_devices) == list(registry2.deleted_devices) new_via = registry2.async_get_device(identifiers={("hue", "0123")}) new_light = registry2.async_get_device(identifiers={("hue", "456")}) new_light4 = registry2.async_get_device(identifiers={("hue", "abc")}) assert orig_via == new_via assert orig_light == new_light assert orig_light4 == new_light4 # Ensure enums converted for old, new in ( (orig_via, new_via), (orig_light, new_light), (orig_light4, new_light4), ): assert old.disabled_by is new.disabled_by assert old.entry_type is new.entry_type # Ensure a save/load cycle does not keep suggested area new_kitchen_light = registry2.async_get_device(identifiers={("hue", "999")}) assert orig_kitchen_light.suggested_area == "Kitchen" orig_kitchen_light_witout_suggested_area = device_registry.async_update_device( orig_kitchen_light.id, suggested_area=None ) assert orig_kitchen_light_witout_suggested_area.suggested_area is None assert orig_kitchen_light_witout_suggested_area == new_kitchen_light 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=mock_config_entry.entry_id, connections={("ethernet", "12:34:56:78:90:AB:CD:EF")}, identifiers={("hue", "456"), ("bla", "123")}, ) with patch( "homeassistant.helpers.device_registry.DeviceRegistry.async_schedule_save" ) as mock_save: entry2 = device_registry.async_get_or_create( 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, mock_config_entry: MockConfigEntry ) -> None: """Make sure we normalize mac addresses.""" entry = device_registry.async_get_or_create( 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=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, mac)}, ) assert test_entry.id == entry.id, mac assert test_entry.connections == { (dr.CONNECTION_NETWORK_MAC, "12:34:56:ab:cd:ef") } # This should not raise for invalid in ( "invalid_mac", "123456ABCDEFG", # 1 extra char "12:34:56:ab:cdef", # not enough : "12:34:56:ab:cd:e:f", # too many : "1234.56abcdef", # not enough . "123.456.abc.def", # too many . ): invalid_mac_entry = device_registry.async_get_or_create( 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, mock_config_entry: MockConfigEntry, freezer: FrozenDateTimeFactory, ) -> None: """Verify that we can update some attributes of a device.""" created_at = datetime.fromisoformat("2024-01-01T01:00:00+00:00") freezer.move_to(created_at) update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) entry = device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("hue", "456"), ("bla", "123")}, ) new_connections = {(dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA")} new_identifiers = {("hue", "654"), ("bla", "321")} assert not entry.area_id assert not entry.labels assert not entry.name_by_user assert entry.created_at == created_at assert entry.modified_at == created_at modified_at = datetime.fromisoformat("2024-02-01T01:00:00+00:00") freezer.move_to(modified_at) with patch.object(device_registry, "async_schedule_save") as mock_save: updated_entry = device_registry.async_update_device( entry.id, area_id="12345A", configuration_url="https://example.com/config", disabled_by=dr.DeviceEntryDisabler.USER, entry_type=dr.DeviceEntryType.SERVICE, hw_version="hw_version", labels={"label1", "label2"}, manufacturer="Test Producer", model="Test Model", model_id="Test Model Name", name_by_user="Test Friendly Name", name="name", new_connections=new_connections, new_identifiers=new_identifiers, serial_number="serial_no", suggested_area="suggested_area", sw_version="version", via_device_id="98765B", ) assert mock_save.call_count == 1 assert updated_entry != entry assert updated_entry == dr.DeviceEntry( area_id="12345A", config_entries={mock_config_entry.entry_id}, configuration_url="https://example.com/config", connections={("mac", "65:43:21:fe:dc:ba")}, created_at=created_at, disabled_by=dr.DeviceEntryDisabler.USER, entry_type=dr.DeviceEntryType.SERVICE, hw_version="hw_version", id=entry.id, identifiers={("bla", "321"), ("hue", "654")}, labels={"label1", "label2"}, manufacturer="Test Producer", model="Test Model", model_id="Test Model Name", modified_at=modified_at, name_by_user="Test Friendly Name", name="name", serial_number="serial_no", suggested_area="suggested_area", sw_version="version", via_device_id="98765B", ) assert device_registry.async_get_device(identifiers={("hue", "456")}) is None assert device_registry.async_get_device(identifiers={("bla", "123")}) is None assert ( device_registry.async_get_device(identifiers={("hue", "654")}) == updated_entry ) assert ( device_registry.async_get_device(identifiers={("bla", "321")}) == updated_entry ) assert ( device_registry.async_get_device( connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")} ) is None ) assert ( device_registry.async_get_device( connections={(dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA")} ) == updated_entry ) assert device_registry.async_get(updated_entry.id) is not None await hass.async_block_till_done() assert len(update_events) == 2 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "update", "device_id": entry.id, "changes": { "area_id": None, "connections": {("mac", "12:34:56:ab:cd:ef")}, "configuration_url": None, "disabled_by": None, "entry_type": None, "hw_version": None, "identifiers": {("bla", "123"), ("hue", "456")}, "labels": set(), "manufacturer": None, "model": None, "model_id": None, "name": None, "name_by_user": None, "serial_number": None, "suggested_area": None, "sw_version": None, "via_device_id": None, }, } with pytest.raises(HomeAssistantError): device_registry.async_update_device( entry.id, merge_connections=new_connections, new_connections=new_connections, ) with pytest.raises(HomeAssistantError): device_registry.async_update_device( entry.id, merge_identifiers=new_identifiers, new_identifiers=new_identifiers, ) @pytest.mark.parametrize( ("initial_connections", "new_connections", "updated_connections"), [ ( # No connection -> single connection None, {(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, {(dr.CONNECTION_NETWORK_MAC, "12:34:56:ab:cd:ef")}, ), ( # No connection -> double connection None, { (dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA"), (dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF"), }, { (dr.CONNECTION_NETWORK_MAC, "65:43:21:fe:dc:ba"), (dr.CONNECTION_NETWORK_MAC, "12:34:56:ab:cd:ef"), }, ), ( # single connection -> no connection {(dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA")}, set(), set(), ), ( # single connection -> single connection {(dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA")}, {(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, {(dr.CONNECTION_NETWORK_MAC, "12:34:56:ab:cd:ef")}, ), ( # single connection -> double connection {(dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA")}, { (dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA"), (dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF"), }, { (dr.CONNECTION_NETWORK_MAC, "65:43:21:fe:dc:ba"), (dr.CONNECTION_NETWORK_MAC, "12:34:56:ab:cd:ef"), }, ), ( # Double connection -> None { (dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF"), (dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA"), }, set(), set(), ), ( # Double connection -> single connection { (dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA"), (dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF"), }, {(dr.CONNECTION_NETWORK_MAC, "65:43:21:FE:DC:BA")}, {(dr.CONNECTION_NETWORK_MAC, "65:43:21:fe:dc:ba")}, ), ], ) async def test_update_connection( device_registry: dr.DeviceRegistry, mock_config_entry: MockConfigEntry, initial_connections: set[tuple[str, str]] | None, new_connections: set[tuple[str, str]] | None, updated_connections: set[tuple[str, str]] | None, ) -> None: """Verify that we can update some attributes of a device.""" entry = device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections=initial_connections, identifiers={("hue", "456"), ("bla", "123")}, ) with patch.object(device_registry, "async_schedule_save") as mock_save: updated_entry = device_registry.async_update_device( entry.id, new_connections=new_connections, ) assert mock_save.call_count == 1 assert updated_entry != entry assert updated_entry.connections == updated_connections assert ( device_registry.async_get_device(identifiers={("bla", "123")}) == updated_entry ) async def test_update_remove_config_entries( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Make sure we do not get duplicate entries.""" update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) 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) entry = device_registry.async_get_or_create( 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=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=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "34:56:78:CD:EF:12")}, identifiers={("bridgeid", "4567")}, manufacturer="manufacturer", model="model", ) entry4 = device_registry.async_update_device( entry2.id, add_config_entry_id=config_entry_3.entry_id ) # Try to add an unknown config entry with pytest.raises(HomeAssistantError): device_registry.async_update_device(entry2.id, add_config_entry_id="blabla") assert len(device_registry.devices) == 2 assert entry.id == entry2.id == entry4.id assert entry.id != entry3.id assert entry2.config_entries == {config_entry_1.entry_id, config_entry_2.entry_id} assert entry4.config_entries == { config_entry_1.entry_id, config_entry_2.entry_id, config_entry_3.entry_id, } device_registry.async_update_device( entry2.id, remove_config_entry_id=config_entry_1.entry_id ) updated_entry = device_registry.async_update_device( entry2.id, remove_config_entry_id=config_entry_3.entry_id ) removed_entry = device_registry.async_update_device( entry3.id, remove_config_entry_id=config_entry_1.entry_id ) 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")}) assert removed_entry is None await hass.async_block_till_done() assert len(update_events) == 7 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "update", "device_id": entry2.id, "changes": { "config_entries": {config_entry_1.entry_id}, }, } assert update_events[2].data == { "action": "create", "device_id": entry3.id, } assert update_events[3].data == { "action": "update", "device_id": entry.id, "changes": { "config_entries": {config_entry_1.entry_id, config_entry_2.entry_id} }, } assert update_events[4].data == { "action": "update", "device_id": entry2.id, "changes": { "config_entries": { config_entry_1.entry_id, config_entry_2.entry_id, config_entry_3.entry_id, }, "primary_config_entry": config_entry_1.entry_id, }, } assert update_events[5].data == { "action": "update", "device_id": entry2.id, "changes": { "config_entries": {config_entry_2.entry_id, config_entry_3.entry_id} }, } assert update_events[6].data == { "action": "remove", "device_id": entry3.id, } async def test_update_suggested_area( hass: HomeAssistant, device_registry: dr.DeviceRegistry, area_registry: ar.AreaRegistry, mock_config_entry: MockConfigEntry, ) -> None: """Verify that we can update the suggested area version of a device.""" update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) entry = device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bla", "123")}, ) assert not entry.suggested_area assert entry.area_id is None suggested_area = "Pool" with patch.object(device_registry, "async_schedule_save") as mock_save: updated_entry = device_registry.async_update_device( entry.id, suggested_area=suggested_area ) assert mock_save.call_count == 1 assert updated_entry != entry assert updated_entry.suggested_area == suggested_area pool_area = area_registry.async_get_area_by_name("Pool") assert pool_area is not None assert updated_entry.area_id == pool_area.id assert len(area_registry.areas) == 1 await hass.async_block_till_done() assert len(update_events) == 2 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "update", "device_id": entry.id, "changes": {"area_id": None, "suggested_area": None}, } # Do not save or fire the event if the suggested # area does not result in a change of area # but still update the actual entry with patch.object(device_registry, "async_schedule_save") as mock_save_2: updated_entry = device_registry.async_update_device( entry.id, suggested_area="Other" ) assert len(update_events) == 2 assert mock_save_2.call_count == 0 assert updated_entry != entry assert updated_entry.suggested_area == "Other" async def test_cleanup_device_registry( hass: HomeAssistant, device_registry: dr.DeviceRegistry, entity_registry: er.EntityRegistry, ) -> None: """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 ) device_registry.async_get_or_create( identifiers={("hue", "d2")}, config_entry_id=config_entry.entry_id ) d3 = device_registry.async_get_or_create( identifiers={("hue", "d3")}, config_entry_id=config_entry.entry_id ) device_registry.async_get_or_create( 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) entity_registry.async_get_or_create("light", "hue", "e1", device_id=d1.id) entity_registry.async_get_or_create("light", "hue", "e2", device_id=d1.id) entity_registry.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, entity_registry) assert device_registry.async_get_device(identifiers={("hue", "d1")}) is not None assert device_registry.async_get_device(identifiers={("hue", "d2")}) is not None assert device_registry.async_get_device(identifiers={("hue", "d3")}) is not None assert device_registry.async_get_device(identifiers={("something", "d4")}) is None async def test_cleanup_device_registry_removes_expired_orphaned_devices( hass: HomeAssistant, device_registry: dr.DeviceRegistry, entity_registry: er.EntityRegistry, ) -> None: """Test cleanup removes expired orphaned devices.""" config_entry = MockConfigEntry(domain="hue") config_entry.add_to_hass(hass) device_registry.async_get_or_create( identifiers={("hue", "d1")}, config_entry_id=config_entry.entry_id ) device_registry.async_get_or_create( identifiers={("hue", "d2")}, config_entry_id=config_entry.entry_id ) device_registry.async_get_or_create( identifiers={("hue", "d3")}, config_entry_id=config_entry.entry_id ) device_registry.async_clear_config_entry(config_entry.entry_id) assert len(device_registry.devices) == 0 assert len(device_registry.deleted_devices) == 3 dr.async_cleanup(hass, device_registry, entity_registry) assert len(device_registry.devices) == 0 assert len(device_registry.deleted_devices) == 3 future_time = time.time() + dr.ORPHANED_DEVICE_KEEP_SECONDS + 1 with patch("time.time", return_value=future_time): dr.async_cleanup(hass, device_registry, entity_registry) assert len(device_registry.devices) == 0 assert len(device_registry.deleted_devices) == 0 async def test_cleanup_startup(hass: HomeAssistant) -> None: """Test we run a cleanup on startup.""" hass.set_state(CoreState.not_running) with patch( "homeassistant.helpers.device_registry.Debouncer.async_call" ) as mock_call: hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) await hass.async_block_till_done() assert len(mock_call.mock_calls) == 1 @pytest.mark.parametrize("load_registries", [False]) async def test_cleanup_entity_registry_change(hass: HomeAssistant) -> None: """Test we run a cleanup when entity registry changes. Don't pre-load the registries as the debouncer will then not be waiting for EVENT_ENTITY_REGISTRY_UPDATED events. """ await dr.async_load(hass) await er.async_load(hass) ent_reg = er.async_get(hass) with patch( "homeassistant.helpers.device_registry.Debouncer.async_schedule_call" ) as mock_call: entity = ent_reg.async_get_or_create("light", "hue", "e1") await hass.async_block_till_done() assert len(mock_call.mock_calls) == 0 # Normal update does not trigger ent_reg.async_update_entity(entity.entity_id, name="updated") await hass.async_block_till_done() assert len(mock_call.mock_calls) == 0 # Device ID update triggers ent_reg.async_get_or_create("light", "hue", "e1", device_id="bla") await hass.async_block_till_done() assert len(mock_call.mock_calls) == 1 # Removal also triggers ent_reg.async_remove(entity.entity_id) await hass.async_block_till_done() assert len(mock_call.mock_calls) == 2 async def test_restore_device( hass: HomeAssistant, device_registry: dr.DeviceRegistry, mock_config_entry: MockConfigEntry, ) -> None: """Make sure device id is stable.""" update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) entry = device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) assert len(device_registry.devices) == 1 assert len(device_registry.deleted_devices) == 0 device_registry.async_remove_device(entry.id) assert len(device_registry.devices) == 0 assert len(device_registry.deleted_devices) == 1 entry2 = device_registry.async_get_or_create( 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=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) assert entry.id == entry3.id assert entry.id != entry2.id assert len(device_registry.devices) == 2 assert len(device_registry.deleted_devices) == 0 assert isinstance(entry3.config_entries, set) assert isinstance(entry3.connections, set) assert isinstance(entry3.identifiers, set) await hass.async_block_till_done() assert len(update_events) == 4 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "remove", "device_id": entry.id, } assert update_events[2].data == { "action": "create", "device_id": entry2.id, } assert update_events[3].data == { "action": "create", "device_id": entry3.id, } async def test_restore_simple_device( hass: HomeAssistant, device_registry: dr.DeviceRegistry, mock_config_entry: MockConfigEntry, ) -> None: """Make sure device id is stable.""" update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) entry = device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, ) assert len(device_registry.devices) == 1 assert len(device_registry.deleted_devices) == 0 device_registry.async_remove_device(entry.id) assert len(device_registry.devices) == 0 assert len(device_registry.deleted_devices) == 1 entry2 = device_registry.async_get_or_create( 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=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, ) assert entry.id == entry3.id assert entry.id != entry2.id assert len(device_registry.devices) == 2 assert len(device_registry.deleted_devices) == 0 await hass.async_block_till_done() assert len(update_events) == 4 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "remove", "device_id": entry.id, } assert update_events[2].data == { "action": "create", "device_id": entry2.id, } assert update_events[3].data == { "action": "create", "device_id": entry3.id, } async def test_restore_shared_device( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Make sure device id is stable for shared devices.""" update_events = async_capture_events(hass, dr.EVENT_DEVICE_REGISTRY_UPDATED) 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=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("entry_123", "0123")}, manufacturer="manufacturer", model="model", ) assert len(device_registry.devices) == 1 assert len(device_registry.deleted_devices) == 0 device_registry.async_get_or_create( 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", model="model", ) assert len(device_registry.devices) == 1 assert len(device_registry.deleted_devices) == 0 device_registry.async_remove_device(entry.id) assert len(device_registry.devices) == 0 assert len(device_registry.deleted_devices) == 1 entry2 = device_registry.async_get_or_create( 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", model="model", ) assert entry.id == entry2.id assert len(device_registry.devices) == 1 assert len(device_registry.deleted_devices) == 0 assert isinstance(entry2.config_entries, set) assert isinstance(entry2.connections, set) assert isinstance(entry2.identifiers, set) device_registry.async_remove_device(entry.id) entry3 = device_registry.async_get_or_create( 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", model="model", ) assert entry.id == entry3.id assert len(device_registry.devices) == 1 assert len(device_registry.deleted_devices) == 0 assert isinstance(entry3.config_entries, set) assert isinstance(entry3.connections, set) assert isinstance(entry3.identifiers, set) entry4 = device_registry.async_get_or_create( 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", model="model", ) assert entry.id == entry4.id assert len(device_registry.devices) == 1 assert len(device_registry.deleted_devices) == 0 assert isinstance(entry4.config_entries, set) assert isinstance(entry4.connections, set) assert isinstance(entry4.identifiers, set) await hass.async_block_till_done() assert len(update_events) == 7 assert update_events[0].data == { "action": "create", "device_id": entry.id, } assert update_events[1].data == { "action": "update", "device_id": entry.id, "changes": { "config_entries": {config_entry_1.entry_id}, "identifiers": {("entry_123", "0123")}, }, } assert update_events[2].data == { "action": "remove", "device_id": entry.id, } assert update_events[3].data == { "action": "create", "device_id": entry.id, } assert update_events[4].data == { "action": "remove", "device_id": entry.id, } assert update_events[5].data == { "action": "create", "device_id": entry.id, } assert update_events[6].data == { "action": "update", "device_id": entry.id, "changes": { "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=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, ) assert entry.name is None assert entry.model is None assert entry.manufacturer is None entry = device_registry.async_get_or_create( 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", default_manufacturer="default manufacturer 1", ) assert entry.name == "default name 1" assert entry.model == "default model 1" assert entry.manufacturer == "default manufacturer 1" entry = device_registry.async_get_or_create( 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", default_manufacturer="default manufacturer 2", ) assert entry.name == "default name 1" assert entry.model == "default model 1" assert entry.manufacturer == "default manufacturer 1" 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=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, ) assert entry.name is None assert entry.model is None assert entry.manufacturer is None entry = device_registry.async_get_or_create( 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", manufacturer="manufacturer 1", ) assert entry.name == "name 1" assert entry.model == "model 1" assert entry.manufacturer == "manufacturer 1" entry = device_registry.async_get_or_create( 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", default_manufacturer="default manufacturer 1", ) assert entry.name == "name 1" assert entry.model == "model 1" assert entry.manufacturer == "manufacturer 1" 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=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", default_manufacturer="default manufacturer 1", ) assert entry.name == "default name 1" assert entry.model == "default model 1" assert entry.manufacturer == "default manufacturer 1" entry = device_registry.async_get_or_create( 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", default_manufacturer="default manufacturer 2", ) assert entry.name == "default name 1" assert entry.model == "default model 1" assert entry.manufacturer == "default manufacturer 1" async def test_verify_suggested_area_does_not_overwrite_area_id( 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=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, sw_version="sw-version", name="name", manufacturer="manufacturer", model="model", ) entry = device_registry.async_update_device( original_entry.id, area_id=game_room_area.id ) assert entry.area_id == game_room_area.id entry2 = device_registry.async_get_or_create( 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", name="name", manufacturer="manufacturer", model="model", suggested_area="New Game Room", ) assert entry2.area_id == game_room_area.id async def test_disable_config_entry_disables_devices( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Test that we disable entities tied to a config entry.""" config_entry = MockConfigEntry(domain="light") config_entry.add_to_hass(hass) entry1 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, ) entry2 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "34:56:AB:CD:EF:12")}, disabled_by=dr.DeviceEntryDisabler.USER, ) assert not entry1.disabled assert entry2.disabled await hass.config_entries.async_set_disabled_by( config_entry.entry_id, config_entries.ConfigEntryDisabler.USER ) await hass.async_block_till_done() entry1 = device_registry.async_get(entry1.id) assert entry1.disabled assert entry1.disabled_by is dr.DeviceEntryDisabler.CONFIG_ENTRY entry2 = device_registry.async_get(entry2.id) assert entry2.disabled assert entry2.disabled_by is dr.DeviceEntryDisabler.USER await hass.config_entries.async_set_disabled_by(config_entry.entry_id, None) await hass.async_block_till_done() entry1 = device_registry.async_get(entry1.id) assert not entry1.disabled entry2 = device_registry.async_get(entry2.id) assert entry2.disabled assert entry2.disabled_by is dr.DeviceEntryDisabler.USER async def test_only_disable_device_if_all_config_entries_are_disabled( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Test that we only disable device if all related config entries are disabled.""" config_entry1 = MockConfigEntry(domain="light") config_entry1.add_to_hass(hass) config_entry2 = MockConfigEntry(domain="light") config_entry2.add_to_hass(hass) device_registry.async_get_or_create( config_entry_id=config_entry1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, ) entry1 = device_registry.async_get_or_create( config_entry_id=config_entry2.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, ) assert len(entry1.config_entries) == 2 assert not entry1.disabled await hass.config_entries.async_set_disabled_by( config_entry1.entry_id, config_entries.ConfigEntryDisabler.USER ) await hass.async_block_till_done() entry1 = device_registry.async_get(entry1.id) assert not entry1.disabled await hass.config_entries.async_set_disabled_by( config_entry2.entry_id, config_entries.ConfigEntryDisabler.USER ) await hass.async_block_till_done() entry1 = device_registry.async_get(entry1.id) assert entry1.disabled assert entry1.disabled_by is dr.DeviceEntryDisabler.CONFIG_ENTRY await hass.config_entries.async_set_disabled_by(config_entry1.entry_id, None) await hass.async_block_till_done() entry1 = device_registry.async_get(entry1.id) assert not entry1.disabled @pytest.mark.parametrize( ("configuration_url", "expectation"), [ ("http://localhost", nullcontext()), ("http://localhost:8123", nullcontext()), ("https://example.com", nullcontext()), ("http://localhost/config", nullcontext()), ("http://localhost:8123/config", nullcontext()), ("https://example.com/config", nullcontext()), ("homeassistant://config", nullcontext()), (URL("http://localhost"), nullcontext()), (URL("http://localhost:8123"), nullcontext()), (URL("https://example.com"), nullcontext()), (URL("http://localhost/config"), nullcontext()), (URL("http://localhost:8123/config"), nullcontext()), (URL("https://example.com/config"), nullcontext()), (URL("homeassistant://config"), nullcontext()), (None, nullcontext()), ("http://", pytest.raises(ValueError)), ("https://", pytest.raises(ValueError)), ("gopher://localhost", pytest.raises(ValueError)), ("homeassistant://", pytest.raises(ValueError)), (URL("http://"), pytest.raises(ValueError)), (URL("https://"), pytest.raises(ValueError)), (URL("gopher://localhost"), pytest.raises(ValueError)), (URL("homeassistant://"), pytest.raises(ValueError)), # Exception implements __str__ (Exception("https://example.com"), nullcontext()), (Exception("https://"), pytest.raises(ValueError)), (Exception(), pytest.raises(ValueError)), ], ) async def test_device_info_configuration_url_validation( hass: HomeAssistant, device_registry: dr.DeviceRegistry, configuration_url: str | URL | None, expectation: AbstractContextManager, ) -> 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=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=config_entry_2.entry_id, identifiers={("something", "5678")}, name="name", ) with expectation: device_registry.async_update_device( update_device.id, configuration_url=configuration_url ) @pytest.mark.parametrize("load_registries", [False]) async def test_loading_invalid_configuration_url_from_storage( 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] = { "version": dr.STORAGE_VERSION_MAJOR, "minor_version": dr.STORAGE_VERSION_MINOR, "data": { "devices": [ { "area_id": None, "config_entries": ["1234"], "configuration_url": "invalid", "connections": [], "created_at": "2024-01-01T00:00:00+00:00", "disabled_by": None, "entry_type": dr.DeviceEntryType.SERVICE, "hw_version": None, "id": "abcdefghijklm", "identifiers": [["serial", "123456ABCDEF"]], "labels": [], "manufacturer": None, "model": None, "model_id": None, "modified_at": "2024-02-01T00:00:00+00:00", "name_by_user": None, "name": None, "primary_config_entry": "1234", "serial_number": None, "sw_version": None, "via_device_id": None, } ], "deleted_devices": [], }, } await dr.async_load(hass) registry = dr.async_get(hass) assert len(registry.devices) == 1 entry = registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, identifiers={("serial", "123456ABCDEF")}, ) assert entry.configuration_url == "invalid" def test_all() -> None: """Test module.__all__ is correctly set.""" help_test_all(dr) @pytest.mark.parametrize(("enum"), list(dr.DeviceEntryDisabler)) def test_deprecated_constants( caplog: pytest.LogCaptureFixture, enum: dr.DeviceEntryDisabler, ) -> None: """Test deprecated constants.""" import_and_test_deprecated_constant_enum(caplog, dr, enum, "DISABLED_", "2025.1") async def test_removing_labels( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Make sure we can clear labels.""" config_entry = MockConfigEntry() config_entry.add_to_hass(hass) entry = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) entry = device_registry.async_update_device(entry.id, labels={"label1", "label2"}) device_registry.async_clear_label_id("label1") entry_cleared_label1 = device_registry.async_get_device({("bridgeid", "0123")}) device_registry.async_clear_label_id("label2") entry_cleared_label2 = device_registry.async_get_device({("bridgeid", "0123")}) assert entry_cleared_label1 assert entry_cleared_label2 assert entry != entry_cleared_label1 assert entry != entry_cleared_label2 assert entry_cleared_label1 != entry_cleared_label2 assert entry.labels == {"label1", "label2"} assert entry_cleared_label1.labels == {"label2"} assert not entry_cleared_label2.labels async def test_entries_for_label( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Test getting device entries by label.""" config_entry = MockConfigEntry() config_entry.add_to_hass(hass) device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:00")}, identifiers={("bridgeid", "0000")}, manufacturer="manufacturer", model="model", ) entry_1 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:23")}, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) entry_1 = device_registry.async_update_device(entry_1.id, labels={"label1"}) entry_2 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:56")}, identifiers={("bridgeid", "0456")}, manufacturer="manufacturer", model="model", ) entry_2 = device_registry.async_update_device(entry_2.id, labels={"label2"}) entry_1_and_2 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:89")}, identifiers={("bridgeid", "0789")}, manufacturer="manufacturer", model="model", ) entry_1_and_2 = device_registry.async_update_device( entry_1_and_2.id, labels={"label1", "label2"} ) entries = dr.async_entries_for_label(device_registry, "label1") assert len(entries) == 2 assert entries == [entry_1, entry_1_and_2] entries = dr.async_entries_for_label(device_registry, "label2") assert len(entries) == 2 assert entries == [entry_2, entry_1_and_2] assert not dr.async_entries_for_label(device_registry, "unknown") assert not dr.async_entries_for_label(device_registry, "") @pytest.mark.parametrize( ( "translation_key", "translations", "placeholders", "expected_device_name", ), [ (None, None, None, "Device Bla"), ( "test_device", { "en": {"component.test.device.test_device.name": "English device"}, }, None, "English device", ), ( "test_device", { "en": { "component.test.device.test_device.name": "{placeholder} English dev" }, }, {"placeholder": "special"}, "special English dev", ), ( "test_device", { "en": { "component.test.device.test_device.name": "English dev {placeholder}" }, }, {"placeholder": "special"}, "English dev special", ), ], ) async def test_device_name_translation_placeholders( hass: HomeAssistant, device_registry: dr.DeviceRegistry, translation_key: str | None, translations: dict[str, str] | None, placeholders: dict[str, str] | None, expected_device_name: str | None, ) -> None: """Test device name when the device name translation has placeholders.""" def async_get_cached_translations( hass: HomeAssistant, language: str, category: str, integrations: Iterable[str] | None = None, config_flow: bool | None = None, ) -> dict[str, Any]: """Return all backend translations.""" return translations[language] config_entry_1 = MockConfigEntry() config_entry_1.add_to_hass(hass) with patch( "homeassistant.helpers.device_registry.translation.async_get_cached_translations", side_effect=async_get_cached_translations, ): entry1 = device_registry.async_get_or_create( config_entry_id=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, name="Device Bla", translation_key=translation_key, translation_placeholders=placeholders, ) assert entry1.name == expected_device_name @pytest.mark.parametrize( ( "translation_key", "translations", "placeholders", "release_channel", "expectation", "expected_error", ), [ ( "test_device", { "en": { "component.test.device.test_device.name": "{placeholder} English dev {2ndplaceholder}" }, }, {"placeholder": "special"}, ReleaseChannel.STABLE, nullcontext(), ( "has translation placeholders '{'placeholder': 'special'}' which do " "not match the name '{placeholder} English dev {2ndplaceholder}'" ), ), ( "test_device", { "en": { "component.test.device.test_device.name": "{placeholder} English ent {2ndplaceholder}" }, }, {"placeholder": "special"}, ReleaseChannel.BETA, pytest.raises( HomeAssistantError, match="Missing placeholder '2ndplaceholder'" ), "", ), ( "test_device", { "en": { "component.test.device.test_device.name": "{placeholder} English dev" }, }, None, ReleaseChannel.STABLE, nullcontext(), ( "has translation placeholders '{}' which do " "not match the name '{placeholder} English dev'" ), ), ], ) async def test_device_name_translation_placeholders_errors( hass: HomeAssistant, device_registry: dr.DeviceRegistry, translation_key: str | None, translations: dict[str, str] | None, placeholders: dict[str, str] | None, release_channel: ReleaseChannel, expectation: AbstractContextManager, expected_error: str, caplog: pytest.LogCaptureFixture, ) -> None: """Test device name has placeholder issuess.""" def async_get_cached_translations( hass: HomeAssistant, language: str, category: str, integrations: Iterable[str] | None = None, config_flow: bool | None = None, ) -> dict[str, Any]: """Return all backend translations.""" return translations[language] config_entry_1 = MockConfigEntry() config_entry_1.add_to_hass(hass) with ( patch( "homeassistant.helpers.device_registry.translation.async_get_cached_translations", side_effect=async_get_cached_translations, ), patch( "homeassistant.helpers.device_registry.get_release_channel", return_value=release_channel, ), expectation, ): device_registry.async_get_or_create( config_entry_id=config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, name="Device Bla", translation_key=translation_key, translation_placeholders=placeholders, ) assert expected_error in caplog.text async def test_async_get_or_create_thread_safety( hass: HomeAssistant, device_registry: dr.DeviceRegistry, mock_config_entry: MockConfigEntry, ) -> None: """Test async_get_or_create raises when called from wrong thread.""" with pytest.raises( RuntimeError, match="Detected code that calls device_registry.async_update_device from a thread.", ): await hass.async_add_executor_job( partial( device_registry.async_get_or_create, 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", ) ) async def test_async_remove_device_thread_safety( hass: HomeAssistant, device_registry: dr.DeviceRegistry, mock_config_entry: MockConfigEntry, ) -> None: """Test async_remove_device raises when called from wrong thread.""" device = device_registry.async_get_or_create( 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", ) with pytest.raises( RuntimeError, match="Detected code that calls device_registry.async_remove_device from a thread.", ): await hass.async_add_executor_job( device_registry.async_remove_device, device.id ) async def test_device_registry_connections_collision( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Test connection collisions in the device registry.""" config_entry = MockConfigEntry() config_entry.add_to_hass(hass) device1 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "none")}, manufacturer="manufacturer", model="model", ) device2 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "none")}, manufacturer="manufacturer", model="model", ) assert device1.id == device2.id device3 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) # Attempt to merge connection for device3 with the same # connection that already exists in device1 with pytest.raises( HomeAssistantError, match=f"Connections.*already registered.*{device1.id}" ): device_registry.async_update_device( device3.id, merge_connections={ (dr.CONNECTION_NETWORK_MAC, "EE:EE:EE:EE:EE:EE"), (dr.CONNECTION_NETWORK_MAC, "none"), }, ) # Attempt to add new connections for device3 with the same # connection that already exists in device1 with pytest.raises( HomeAssistantError, match=f"Connections.*already registered.*{device1.id}" ): device_registry.async_update_device( device3.id, new_connections={ (dr.CONNECTION_NETWORK_MAC, "EE:EE:EE:EE:EE:EE"), (dr.CONNECTION_NETWORK_MAC, "none"), }, ) device3_refetched = device_registry.async_get(device3.id) assert device3_refetched.connections == set() assert device3_refetched.identifiers == {("bridgeid", "0123")} device1_refetched = device_registry.async_get(device1.id) assert device1_refetched.connections == {(dr.CONNECTION_NETWORK_MAC, "none")} assert device1_refetched.identifiers == set() device2_refetched = device_registry.async_get(device2.id) assert device2_refetched.connections == {(dr.CONNECTION_NETWORK_MAC, "none")} assert device2_refetched.identifiers == set() assert device2_refetched.id == device1_refetched.id assert len(device_registry.devices) == 2 # Attempt to implicitly merge connection for device3 with the same # connection that already exists in device1 device4 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, identifiers={("bridgeid", "0123")}, connections={ (dr.CONNECTION_NETWORK_MAC, "EE:EE:EE:EE:EE:EE"), (dr.CONNECTION_NETWORK_MAC, "none"), }, ) assert len(device_registry.devices) == 2 assert device4.id in (device1.id, device3.id) device3_refetched = device_registry.async_get(device3.id) device1_refetched = device_registry.async_get(device1.id) assert not device1_refetched.connections.isdisjoint(device3_refetched.connections) async def test_device_registry_identifiers_collision( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: """Test identifiers collisions in the device registry.""" config_entry = MockConfigEntry() config_entry.add_to_hass(hass) device1 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) device2 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, identifiers={("bridgeid", "0123")}, manufacturer="manufacturer", model="model", ) assert device1.id == device2.id device3 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, identifiers={("bridgeid", "4567")}, manufacturer="manufacturer", model="model", ) # Attempt to merge identifiers for device3 with the same # connection that already exists in device1 with pytest.raises( HomeAssistantError, match=f"Identifiers.*already registered.*{device1.id}" ): device_registry.async_update_device( device3.id, merge_identifiers={("bridgeid", "0123"), ("bridgeid", "8888")} ) # Attempt to add new identifiers for device3 with the same # connection that already exists in device1 with pytest.raises( HomeAssistantError, match=f"Identifiers.*already registered.*{device1.id}" ): device_registry.async_update_device( device3.id, new_identifiers={("bridgeid", "0123"), ("bridgeid", "8888")} ) device3_refetched = device_registry.async_get(device3.id) assert device3_refetched.connections == set() assert device3_refetched.identifiers == {("bridgeid", "4567")} device1_refetched = device_registry.async_get(device1.id) assert device1_refetched.connections == set() assert device1_refetched.identifiers == {("bridgeid", "0123")} device2_refetched = device_registry.async_get(device2.id) assert device2_refetched.connections == set() assert device2_refetched.identifiers == {("bridgeid", "0123")} assert device2_refetched.id == device1_refetched.id assert len(device_registry.devices) == 2 # Attempt to implicitly merge identifiers for device3 with the same # connection that already exists in device1 device4 = device_registry.async_get_or_create( config_entry_id=config_entry.entry_id, identifiers={("bridgeid", "4567"), ("bridgeid", "0123")}, ) assert len(device_registry.devices) == 2 assert device4.id in (device1.id, device3.id) device3_refetched = device_registry.async_get(device3.id) device1_refetched = device_registry.async_get(device1.id) assert not device1_refetched.identifiers.isdisjoint(device3_refetched.identifiers) async def test_primary_config_entry( hass: HomeAssistant, device_registry: dr.DeviceRegistry, ) -> None: """Test the primary integration field.""" mock_config_entry_1 = MockConfigEntry(domain="mqtt", title=None) mock_config_entry_1.add_to_hass(hass) mock_config_entry_2 = MockConfigEntry(title=None) mock_config_entry_2.add_to_hass(hass) mock_config_entry_3 = MockConfigEntry(title=None) mock_config_entry_3.add_to_hass(hass) mock_config_entry_4 = MockConfigEntry(domain="matter", title=None) mock_config_entry_4.add_to_hass(hass) # Create device without model name etc, config entry will not be marked primary device = device_registry.async_get_or_create( config_entry_id=mock_config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers=set(), ) assert device.primary_config_entry is None # Set model, mqtt config entry will be promoted to primary device = device_registry.async_get_or_create( config_entry_id=mock_config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, model="model", ) assert device.primary_config_entry == mock_config_entry_1.entry_id # New config entry with model will be promoted to primary device = device_registry.async_get_or_create( config_entry_id=mock_config_entry_2.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, model="model 2", ) assert device.primary_config_entry == mock_config_entry_2.entry_id # New config entry with model will not be promoted to primary device = device_registry.async_get_or_create( config_entry_id=mock_config_entry_3.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, model="model 3", ) assert device.primary_config_entry == mock_config_entry_2.entry_id # New matter config entry with model will not be promoted to primary device = device_registry.async_get_or_create( config_entry_id=mock_config_entry_4.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, model="model 3", ) assert device.primary_config_entry == mock_config_entry_2.entry_id # Remove the primary config entry device = device_registry.async_update_device( device.id, remove_config_entry_id=mock_config_entry_2.entry_id, ) assert device.primary_config_entry is None # Create new device = device_registry.async_get_or_create( config_entry_id=mock_config_entry_1.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers=set(), manufacturer="manufacturer", model="model", ) assert device.primary_config_entry == mock_config_entry_1.entry_id async def test_update_device_no_connections_or_identifiers( hass: HomeAssistant, device_registry: dr.DeviceRegistry, ) -> None: """Test updating a device clearing connections and identifiers.""" mock_config_entry = MockConfigEntry(domain="mqtt", title=None) mock_config_entry.add_to_hass(hass) device = device_registry.async_get_or_create( config_entry_id=mock_config_entry.entry_id, connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")}, identifiers={("bridgeid", "0123")}, ) with pytest.raises(HomeAssistantError): device_registry.async_update_device( device.id, new_connections=set(), new_identifiers=set() )