github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Improve type annotations for Airly integration (#49898)

This commit is contained in:
Maciej Bieniek 2021-05-07 16:47:52 +02:00 committed by GitHub
parent 4d0955bae1
commit 6df0190aeb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 268 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -3,6 +3,7 @@
# to enable strict mypy checks.
homeassistant.components
homeassistant.components.airly.*
homeassistant.components.automation.*
homeassistant.components.binary_sensor.*
homeassistant.components.bond.*

63
homeassistant/components/airly/__init__.py
View File

@ -1,15 +1,21 @@
"""The Airly integration."""
from __future__ import annotations
from datetime import timedelta
import logging
from math import ceil
from aiohttp import ClientSession
from aiohttp.client_exceptions import ClientConnectorError
from airly import Airly
from airly.exceptions import AirlyError
import async_timeout
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import async_get_registry
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
@ -30,7 +36,7 @@ PLATFORMS = ["air_quality", "sensor"]
_LOGGER = logging.getLogger(__name__)
def set_update_interval(instances, requests_remaining):
def set_update_interval(instances_count: int, requests_remaining: int) -> timedelta:
"""
Return data update interval.
@ -46,7 +52,7 @@ def set_update_interval(instances, requests_remaining):
interval = timedelta(
minutes=min(
max(
ceil(minutes_to_midnight / requests_remaining * instances),
ceil(minutes_to_midnight / requests_remaining * instances_count),
MIN_UPDATE_INTERVAL,
),
MAX_UPDATE_INTERVAL,
@ -58,19 +64,28 @@ def set_update_interval(instances, requests_remaining):
return interval
async def async_setup_entry(hass, config_entry):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Airly as config entry."""
api_key = config_entry.data[CONF_API_KEY]
latitude = config_entry.data[CONF_LATITUDE]
longitude = config_entry.data[CONF_LONGITUDE]
use_nearest = config_entry.data.get(CONF_USE_NEAREST, False)
api_key = entry.data[CONF_API_KEY]
latitude = entry.data[CONF_LATITUDE]
longitude = entry.data[CONF_LONGITUDE]
use_nearest = entry.data.get(CONF_USE_NEAREST, False)
# For backwards compat, set unique ID
if config_entry.unique_id is None:
if entry.unique_id is None:
hass.config_entries.async_update_entry(
config_entry, unique_id=f"{latitude}-{longitude}"
entry, unique_id=f"{latitude}-{longitude}"
)
# identifiers in device_info should use Tuple[str, str, str] type, but latitude and
# longitude are float, so we convert old device entries to use correct types
device_registry = await async_get_registry(hass)
old_ids = (DOMAIN, latitude, longitude)
device_entry = device_registry.async_get_device({old_ids})
if device_entry and entry.entry_id in device_entry.config_entries:
new_ids = (DOMAIN, str(latitude), str(longitude))
device_registry.async_update_device(device_entry.id, new_identifiers={new_ids})
websession = async_get_clientsession(hass)
update_interval = timedelta(minutes=MIN_UPDATE_INTERVAL)
@ -81,21 +96,19 @@ async def async_setup_entry(hass, config_entry):
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][config_entry.entry_id] = coordinator
hass.data[DOMAIN][entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass, config_entry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(
config_entry, PLATFORMS
)
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(config_entry.entry_id)
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
@ -105,13 +118,13 @@ class AirlyDataUpdateCoordinator(DataUpdateCoordinator):
def __init__(
self,
hass,
session,
api_key,
latitude,
longitude,
update_interval,
use_nearest,
hass: HomeAssistant,
session: ClientSession,
api_key: str,
latitude: float,
longitude: float,
update_interval: timedelta,
use_nearest: bool,
):
"""Initialize."""
self.latitude = latitude
@ -121,9 +134,9 @@ class AirlyDataUpdateCoordinator(DataUpdateCoordinator):
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
async def _async_update_data(self):
async def _async_update_data(self) -> dict[str, str | float | int]:
"""Update data via library."""
data = {}
data: dict[str, str | float | int] = {}
if self.use_nearest:
measurements = self.airly.create_measurements_session_nearest(
self.latitude, self.longitude, max_distance_km=5

71
homeassistant/components/airly/air_quality.py
View File

@ -1,13 +1,22 @@
"""Support for the Airly air_quality service."""
from __future__ import annotations
from typing import Any
from homeassistant.components.air_quality import (
ATTR_AQI,
ATTR_PM_2_5,
ATTR_PM_10,
AirQualityEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import AirlyDataUpdateCoordinator
from .const import (
ATTR_API_ADVICE,
ATTR_API_CAQI,
@ -36,80 +45,73 @@ LABEL_PM_10_PERCENT = f"{ATTR_PM_10}_percent_of_limit"
PARALLEL_UPDATES = 1
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Airly air_quality entity based on a config entry."""
name = config_entry.data[CONF_NAME]
name = entry.data[CONF_NAME]
coordinator = hass.data[DOMAIN][config_entry.entry_id]
coordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities([AirlyAirQuality(coordinator, name)], False)
def round_state(func):
"""Round state."""
def _decorator(self):
res = func(self)
if isinstance(res, float):
return round(res)
return res
return _decorator
class AirlyAirQuality(CoordinatorEntity, AirQualityEntity):
"""Define an Airly air quality."""
def __init__(self, coordinator, name):
coordinator: AirlyDataUpdateCoordinator
def __init__(self, coordinator: AirlyDataUpdateCoordinator, name: str) -> None:
"""Initialize."""
super().__init__(coordinator)
self._name = name
self._icon = "mdi:blur"
@property
def name(self):
def name(self) -> str:
"""Return the name."""
return self._name
@property
def icon(self):
def icon(self) -> str:
"""Return the icon."""
return self._icon
@property
@round_state
def air_quality_index(self):
def air_quality_index(self) -> float | None:
"""Return the air quality index."""
return self.coordinator.data[ATTR_API_CAQI]
return round_state(self.coordinator.data[ATTR_API_CAQI])
@property
@round_state
def particulate_matter_2_5(self):
def particulate_matter_2_5(self) -> float | None:
"""Return the particulate matter 2.5 level."""
return self.coordinator.data.get(ATTR_API_PM25)
return round_state(self.coordinator.data.get(ATTR_API_PM25))
@property
@round_state
def particulate_matter_10(self):
def particulate_matter_10(self) -> float | None:
"""Return the particulate matter 10 level."""
return self.coordinator.data.get(ATTR_API_PM10)
return round_state(self.coordinator.data.get(ATTR_API_PM10))
@property
def attribution(self):
def attribution(self) -> str:
"""Return the attribution."""
return ATTRIBUTION
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return a unique_id for this entity."""
return f"{self.coordinator.latitude}-{self.coordinator.longitude}"
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return {
"identifiers": {
(DOMAIN, self.coordinator.latitude, self.coordinator.longitude)
(
DOMAIN,
str(self.coordinator.latitude),
str(self.coordinator.longitude),
)
},
"name": DEFAULT_NAME,
"manufacturer": MANUFACTURER,
@ -117,7 +119,7 @@ class AirlyAirQuality(CoordinatorEntity, AirQualityEntity):
}
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
attrs = {
LABEL_AQI_DESCRIPTION: self.coordinator.data[ATTR_API_CAQI_DESCRIPTION],
@ -135,3 +137,8 @@ class AirlyAirQuality(CoordinatorEntity, AirQualityEntity):
self.coordinator.data[ATTR_API_PM10_PERCENT]
)
return attrs
def round_state(state: float | None) -> float | None:
"""Round state."""
return round(state) if state else state

18
homeassistant/components/airly/config_flow.py
View File

@ -1,4 +1,9 @@
"""Adds config flow for Airly."""
from __future__ import annotations
from typing import Any
from aiohttp import ClientSession
from airly import Airly
from airly.exceptions import AirlyError
import async_timeout
@ -13,6 +18,7 @@ from homeassistant.const import (
HTTP_NOT_FOUND,
HTTP_UNAUTHORIZED,
)
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
@ -24,7 +30,9 @@ class AirlyFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
errors = {}
use_nearest = False
@ -84,7 +92,13 @@ class AirlyFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
async def test_location(client, api_key, latitude, longitude, use_nearest=False):
async def test_location(
client: ClientSession,
api_key: str,
latitude: float,
longitude: float,
use_nearest: bool = False,
) -> bool:
"""Return true if location is valid."""
airly = Airly(api_key, client)
if use_nearest:

89
homeassistant/components/airly/const.py
View File

@ -1,29 +1,68 @@
"""Constants for Airly integration."""
from __future__ import annotations
ATTR_API_ADVICE = "ADVICE"
ATTR_API_CAQI = "CAQI"
ATTR_API_CAQI_DESCRIPTION = "DESCRIPTION"
ATTR_API_CAQI_LEVEL = "LEVEL"
ATTR_API_HUMIDITY = "HUMIDITY"
ATTR_API_PM1 = "PM1"
ATTR_API_PM10 = "PM10"
ATTR_API_PM10_LIMIT = "PM10_LIMIT"
ATTR_API_PM10_PERCENT = "PM10_PERCENT"
ATTR_API_PM25 = "PM25"
ATTR_API_PM25_LIMIT = "PM25_LIMIT"
ATTR_API_PM25_PERCENT = "PM25_PERCENT"
ATTR_API_PRESSURE = "PRESSURE"
ATTR_API_TEMPERATURE = "TEMPERATURE"
from typing import Final
ATTR_LABEL = "label"
ATTR_UNIT = "unit"
from homeassistant.const import (
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
DEVICE_CLASS_HUMIDITY,
DEVICE_CLASS_PRESSURE,
DEVICE_CLASS_TEMPERATURE,
PERCENTAGE,
PRESSURE_HPA,
TEMP_CELSIUS,
)
ATTRIBUTION = "Data provided by Airly"
CONF_USE_NEAREST = "use_nearest"
DEFAULT_NAME = "Airly"
DOMAIN = "airly"
LABEL_ADVICE = "advice"
MANUFACTURER = "Airly sp. z o.o."
MAX_UPDATE_INTERVAL = 90
MIN_UPDATE_INTERVAL = 5
NO_AIRLY_SENSORS = "There are no Airly sensors in this area yet."
from .model import SensorDescription
ATTR_API_ADVICE: Final = "ADVICE"
ATTR_API_CAQI: Final = "CAQI"
ATTR_API_CAQI_DESCRIPTION: Final = "DESCRIPTION"
ATTR_API_CAQI_LEVEL: Final = "LEVEL"
ATTR_API_HUMIDITY: Final = "HUMIDITY"
ATTR_API_PM1: Final = "PM1"
ATTR_API_PM10: Final = "PM10"
ATTR_API_PM10_LIMIT: Final = "PM10_LIMIT"
ATTR_API_PM10_PERCENT: Final = "PM10_PERCENT"
ATTR_API_PM25: Final = "PM25"
ATTR_API_PM25_LIMIT: Final = "PM25_LIMIT"
ATTR_API_PM25_PERCENT: Final = "PM25_PERCENT"
ATTR_API_PRESSURE: Final = "PRESSURE"
ATTR_API_TEMPERATURE: Final = "TEMPERATURE"
ATTRIBUTION: Final = "Data provided by Airly"
CONF_USE_NEAREST: Final = "use_nearest"
DEFAULT_NAME: Final = "Airly"
DOMAIN: Final = "airly"
LABEL_ADVICE: Final = "advice"
MANUFACTURER: Final = "Airly sp. z o.o."
MAX_UPDATE_INTERVAL: Final = 90
MIN_UPDATE_INTERVAL: Final = 5
NO_AIRLY_SENSORS: Final = "There are no Airly sensors in this area yet."
SENSOR_TYPES: dict[str, SensorDescription] = {
ATTR_API_PM1: {
"device_class": None,
"icon": "mdi:blur",
"label": ATTR_API_PM1,
"unit": CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
},
ATTR_API_HUMIDITY: {
"device_class": DEVICE_CLASS_HUMIDITY,
"icon": None,
"label": ATTR_API_HUMIDITY.capitalize(),
"unit": PERCENTAGE,
},
ATTR_API_PRESSURE: {
"device_class": DEVICE_CLASS_PRESSURE,
"icon": None,
"label": ATTR_API_PRESSURE.capitalize(),
"unit": PRESSURE_HPA,
},
ATTR_API_TEMPERATURE: {
"device_class": DEVICE_CLASS_TEMPERATURE,
"icon": None,
"label": ATTR_API_TEMPERATURE.capitalize(),
"unit": TEMP_CELSIUS,
},
}

13
homeassistant/components/airly/model.py Normal file
View File

@ -0,0 +1,13 @@
"""Type definitions for Airly integration."""
from __future__ import annotations
from typing import TypedDict
class SensorDescription(TypedDict):
"""Sensor description class."""
device_class: str | None
icon: str | None
label: str
unit: str

110
homeassistant/components/airly/sensor.py
View File

@ -1,68 +1,38 @@
"""Support for the Airly sensor service."""
from __future__ import annotations
from typing import Any, cast
from homeassistant.components.sensor import SensorEntity
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_DEVICE_CLASS,
ATTR_ICON,
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
CONF_NAME,
DEVICE_CLASS_HUMIDITY,
DEVICE_CLASS_PRESSURE,
DEVICE_CLASS_TEMPERATURE,
PERCENTAGE,
PRESSURE_HPA,
TEMP_CELSIUS,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import AirlyDataUpdateCoordinator
from .const import (
ATTR_API_HUMIDITY,
ATTR_API_PM1,
ATTR_API_PRESSURE,
ATTR_API_TEMPERATURE,
ATTR_LABEL,
ATTR_UNIT,
ATTRIBUTION,
DEFAULT_NAME,
DOMAIN,
MANUFACTURER,
SENSOR_TYPES,
)
PARALLEL_UPDATES = 1
SENSOR_TYPES = {
ATTR_API_PM1: {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:blur",
ATTR_LABEL: ATTR_API_PM1,
ATTR_UNIT: CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
},
ATTR_API_HUMIDITY: {
ATTR_DEVICE_CLASS: DEVICE_CLASS_HUMIDITY,
ATTR_ICON: None,
ATTR_LABEL: ATTR_API_HUMIDITY.capitalize(),
ATTR_UNIT: PERCENTAGE,
},
ATTR_API_PRESSURE: {
ATTR_DEVICE_CLASS: DEVICE_CLASS_PRESSURE,
ATTR_ICON: None,
ATTR_LABEL: ATTR_API_PRESSURE.capitalize(),
ATTR_UNIT: PRESSURE_HPA,
},
ATTR_API_TEMPERATURE: {
ATTR_DEVICE_CLASS: DEVICE_CLASS_TEMPERATURE,
ATTR_ICON: None,
ATTR_LABEL: ATTR_API_TEMPERATURE.capitalize(),
ATTR_UNIT: TEMP_CELSIUS,
},
}
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Airly sensor entities based on a config entry."""
name = config_entry.data[CONF_NAME]
name = entry.data[CONF_NAME]
coordinator = hass.data[DOMAIN][config_entry.entry_id]
coordinator = hass.data[DOMAIN][entry.entry_id]
sensors = []
for sensor in SENSOR_TYPES:
@ -76,59 +46,63 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class AirlySensor(CoordinatorEntity, SensorEntity):
"""Define an Airly sensor."""
def __init__(self, coordinator, name, kind):
coordinator: AirlyDataUpdateCoordinator
def __init__(
self, coordinator: AirlyDataUpdateCoordinator, name: str, kind: str
) -> None:
"""Initialize."""
super().__init__(coordinator)
self._name = name
self._description = SENSOR_TYPES[kind]
self.kind = kind
self._device_class = None
self._state = None
self._icon = None
self._unit_of_measurement = None
self._attrs = {ATTR_ATTRIBUTION: ATTRIBUTION}
@property
def name(self):
def name(self) -> str:
"""Return the name."""
return f"{self._name} {SENSOR_TYPES[self.kind][ATTR_LABEL]}"
return f"{self._name} {self._description['label']}"
@property
def state(self):
def state(self) -> StateType:
"""Return the state."""
self._state = self.coordinator.data[self.kind]
if self.kind in [ATTR_API_PM1, ATTR_API_PRESSURE]:
self._state = round(self._state)
if self.kind in [ATTR_API_TEMPERATURE, ATTR_API_HUMIDITY]:
self._state = round(self._state, 1)
return self._state
return round(cast(float, self._state))
return round(cast(float, self._state), 1)
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
return self._attrs
@property
def icon(self):
def icon(self) -> str | None:
"""Return the icon."""
self._icon = SENSOR_TYPES[self.kind][ATTR_ICON]
return self._icon
return self._description["icon"]
@property
def device_class(self):
def device_class(self) -> str | None:
"""Return the device_class."""
return SENSOR_TYPES[self.kind][ATTR_DEVICE_CLASS]
return self._description["device_class"]
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return a unique_id for this entity."""
return f"{self.coordinator.latitude}-{self.coordinator.longitude}-{self.kind.lower()}"
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return {
"identifiers": {
(DOMAIN, self.coordinator.latitude, self.coordinator.longitude)
(
DOMAIN,
str(self.coordinator.latitude),
str(self.coordinator.longitude),
)
},
"name": DEFAULT_NAME,
"manufacturer": MANUFACTURER,
@ -136,6 +110,6 @@ class AirlySensor(CoordinatorEntity, SensorEntity):
}
@property
def unit_of_measurement(self):
def unit_of_measurement(self) -> str | None:
"""Return the unit the value is expressed in."""
return SENSOR_TYPES[self.kind][ATTR_UNIT]
return self._description["unit"]

6
homeassistant/components/airly/system_health.py
View File

@ -1,4 +1,8 @@
"""Provide info to system health."""
from __future__ import annotations
from typing import Any
from airly import Airly
from homeassistant.components import system_health
@ -15,7 +19,7 @@ def async_register(
register.async_register_info(system_health_info)
async def system_health_info(hass):
async def system_health_info(hass: HomeAssistant) -> dict[str, Any]:
"""Get info for the info page."""
requests_remaining = list(hass.data[DOMAIN].values())[0].airly.requests_remaining
requests_per_day = list(hass.data[DOMAIN].values())[0].airly.requests_per_day

16
mypy.ini
View File

@ -48,6 +48,19 @@ warn_return_any = true
warn_unreachable = true
warn_unused_ignores = true
[mypy-homeassistant.components.airly.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
strict_equality = true
warn_return_any = true
warn_unreachable = true
warn_unused_ignores = true
[mypy-homeassistant.components.automation.*]
check_untyped_defs = true
disallow_incomplete_defs = true
@ -652,9 +665,6 @@ ignore_errors = true
[mypy-homeassistant.components.aemet.*]
ignore_errors = true
[mypy-homeassistant.components.airly.*]
ignore_errors = true
[mypy-homeassistant.components.alarmdecoder.*]
ignore_errors = true

1
script/hassfest/mypy_config.py
View File

@ -16,7 +16,6 @@ from .model import Config, Integration
IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.adguard.*",
"homeassistant.components.aemet.*",
"homeassistant.components.airly.*",
"homeassistant.components.alarmdecoder.*",
"homeassistant.components.alexa.*",
"homeassistant.components.almond.*",

38
tests/components/airly/test_init.py
View File

@ -13,7 +13,12 @@ from homeassistant.util.dt import utcnow
from . import API_POINT_URL
from tests.common import MockConfigEntry, async_fire_time_changed, load_fixture
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
load_fixture,
mock_device_registry,
)
from tests.components.airly import init_integration
@ -181,3 +186,34 @@ async def test_unload_entry(hass, aioclient_mock):
assert entry.state == ENTRY_STATE_NOT_LOADED
assert not hass.data.get(DOMAIN)
async def test_migrate_device_entry(hass, aioclient_mock):
"""Test device_info identifiers migration."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="Home",
unique_id="123-456",
data={
"api_key": "foo",
"latitude": 123,
"longitude": 456,
"name": "Home",
},
)
aioclient_mock.get(API_POINT_URL, text=load_fixture("airly_valid_station.json"))
config_entry.add_to_hass(hass)
device_reg = mock_device_registry(hass)
device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id, identifiers={(DOMAIN, 123, 456)}
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
migrated_device_entry = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id, identifiers={(DOMAIN, "123", "456")}
)
assert device_entry.id == migrated_device_entry.id