Remove legacy migration and yaml from tplink (#62457)

- tplink has been fully migrated to a config flow in previous
  versions.
This commit is contained in:
J. Nick Koston 2021-12-21 04:24:32 -06:00 committed by GitHub
parent a9c45fdcc0
commit e0ef066022
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 8 additions and 689 deletions

View file

@ -7,11 +7,9 @@ from typing import Any
from kasa import SmartDevice, SmartDeviceException
from kasa.discover import Discover
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.config_entries import ConfigEntry, ConfigEntryNotReady
from homeassistant.const import (
CONF_HOST,
@ -20,60 +18,15 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STARTED,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_registry as er,
)
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_DIMMER,
CONF_DISCOVERY,
CONF_LIGHT,
CONF_STRIP,
CONF_SWITCH,
DOMAIN,
PLATFORMS,
)
from .const import DOMAIN, PLATFORMS
from .coordinator import TPLinkDataUpdateCoordinator
from .migration import (
async_migrate_entities_devices,
async_migrate_legacy_entries,
async_migrate_yaml_entries,
)
DISCOVERY_INTERVAL = timedelta(minutes=15)
TPLINK_HOST_SCHEMA = vol.Schema({vol.Required(CONF_HOST): cv.string})
CONFIG_SCHEMA = vol.Schema(
vol.All(
cv.deprecated(DOMAIN),
{
DOMAIN: vol.Schema(
{
vol.Optional(CONF_LIGHT, default=[]): vol.All(
cv.ensure_list, [TPLINK_HOST_SCHEMA]
),
vol.Optional(CONF_SWITCH, default=[]): vol.All(
cv.ensure_list, [TPLINK_HOST_SCHEMA]
),
vol.Optional(CONF_STRIP, default=[]): vol.All(
cv.ensure_list, [TPLINK_HOST_SCHEMA]
),
vol.Optional(CONF_DIMMER, default=[]): vol.All(
cv.ensure_list, [TPLINK_HOST_SCHEMA]
),
vol.Optional(CONF_DISCOVERY, default=True): cv.boolean,
}
)
},
),
extra=vol.ALLOW_EXTRA,
)
@callback
def async_trigger_discovery(
@ -108,30 +61,9 @@ async def async_discover_devices(hass: HomeAssistant) -> dict[str, SmartDevice]:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the TP-Link component."""
conf = config.get(DOMAIN)
hass.data[DOMAIN] = {}
legacy_entry = None
config_entries_by_mac = {}
for entry in hass.config_entries.async_entries(DOMAIN):
if async_entry_is_legacy(entry):
legacy_entry = entry
elif entry.unique_id:
config_entries_by_mac[entry.unique_id] = entry
discovered_devices = await async_discover_devices(hass)
hosts_by_mac = {mac: device.host for mac, device in discovered_devices.items()}
if legacy_entry:
async_migrate_legacy_entries(
hass, hosts_by_mac, config_entries_by_mac, legacy_entry
)
# Migrate the yaml entry that was previously imported
async_migrate_yaml_entries(hass, legacy_entry.data)
if conf is not None:
async_migrate_yaml_entries(hass, conf)
if discovered_devices:
if discovered_devices := await async_discover_devices(hass):
async_trigger_discovery(hass, discovered_devices)
async def _async_discovery(*_: Any) -> None:
@ -146,87 +78,27 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up TPLink from a config entry."""
if async_entry_is_legacy(entry):
return True
legacy_entry: ConfigEntry | None = None
for config_entry in hass.config_entries.async_entries(DOMAIN):
if async_entry_is_legacy(config_entry):
legacy_entry = config_entry
break
if legacy_entry is not None:
await async_migrate_entities_devices(hass, legacy_entry.entry_id, entry)
try:
device: SmartDevice = await Discover.discover_single(entry.data[CONF_HOST])
except SmartDeviceException as ex:
raise ConfigEntryNotReady from ex
if device.is_dimmer:
async_fix_dimmer_unique_id(hass, entry, device)
hass.data[DOMAIN][entry.entry_id] = TPLinkDataUpdateCoordinator(hass, device)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
@callback
def async_fix_dimmer_unique_id(
hass: HomeAssistant, entry: ConfigEntry, device: SmartDevice
) -> None:
"""Migrate the unique id of dimmers back to the legacy one.
Dimmers used to use the switch format since pyHS100 treated them as SmartPlug but
the old code created them as lights
https://github.com/home-assistant/core/blob/2021.9.7/homeassistant/components/tplink/common.py#L86
"""
# This is the unique id before 2021.0/2021.1
original_unique_id = legacy_device_id(device)
# This is the unique id that was used in 2021.0/2021.1 rollout
rollout_unique_id = device.mac.replace(":", "").upper()
entity_registry = er.async_get(hass)
rollout_entity_id = entity_registry.async_get_entity_id(
LIGHT_DOMAIN, DOMAIN, rollout_unique_id
)
original_entry_id = entity_registry.async_get_entity_id(
LIGHT_DOMAIN, DOMAIN, original_unique_id
)
# If they are now using the 2021.0/2021.1 rollout entity id
# and have deleted the original entity id, we want to update that entity id
# so they don't end up with another _2 entity, but only if they deleted
# the original
if rollout_entity_id and not original_entry_id:
entity_registry.async_update_entity(
rollout_entity_id, new_unique_id=original_unique_id
)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
hass_data: dict[str, Any] = hass.data[DOMAIN]
if entry.entry_id not in hass_data:
return True
device: SmartDevice = hass.data[DOMAIN][entry.entry_id].device
device: SmartDevice = hass_data[entry.entry_id].device
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass_data.pop(entry.entry_id)
await device.protocol.close()
return unload_ok
@callback
def async_entry_is_legacy(entry: ConfigEntry) -> bool:
"""Check if a config entry is the legacy shared one."""
return entry.unique_id is None or entry.unique_id == DOMAIN
def legacy_device_id(device: SmartDevice) -> str:
"""Convert the device id so it matches what was used in the original version."""
device_id: str = device.device_id

View file

@ -1,7 +1,6 @@
"""Config flow for TP-Link."""
from __future__ import annotations
import logging
from typing import Any
from kasa import SmartDevice, SmartDeviceException
@ -10,17 +9,15 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import dhcp
from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_MAC, CONF_NAME
from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_MAC
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import DiscoveryInfoType
from . import async_discover_devices, async_entry_is_legacy
from . import async_discover_devices
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for tplink."""
@ -114,9 +111,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self._async_create_entry_from_device(self._discovered_devices[mac])
configured_devices = {
entry.unique_id
for entry in self._async_current_entries()
if not async_entry_is_legacy(entry)
entry.unique_id for entry in self._async_current_entries()
}
self._discovered_devices = await async_discover_devices(self.hass)
devices_name = {
@ -132,18 +127,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data_schema=vol.Schema({vol.Required(CONF_DEVICE): vol.In(devices_name)}),
)
async def async_step_migration(self, migration_input: dict[str, Any]) -> FlowResult:
"""Handle migration from legacy config entry to per device config entry."""
mac = migration_input[CONF_MAC]
await self.async_set_unique_id(dr.format_mac(mac), raise_on_progress=False)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=migration_input[CONF_NAME],
data={
CONF_HOST: migration_input[CONF_HOST],
},
)
@callback
def _async_create_entry_from_device(self, device: SmartDevice) -> FlowResult:
"""Create a config entry from a smart device."""
@ -155,16 +138,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
},
)
async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
"""Handle import step."""
host = user_input[CONF_HOST]
try:
device = await self._async_try_connect(host, raise_on_progress=False)
except SmartDeviceException:
_LOGGER.error("Failed to import %s: cannot connect", host)
return self.async_abort(reason="cannot_connect")
return self._async_create_entry_from_device(device)
async def _async_try_connect(
self, host: str, raise_on_progress: bool = True
) -> SmartDevice:

View file

@ -1,113 +0,0 @@
"""Component to embed TP-Link smart home devices."""
from __future__ import annotations
from datetime import datetime
from types import MappingProxyType
from typing import Any
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_MAC,
CONF_NAME,
EVENT_HOMEASSISTANT_STARTED,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.typing import ConfigType
from .const import CONF_DIMMER, CONF_LIGHT, CONF_STRIP, CONF_SWITCH, DOMAIN
async def async_cleanup_legacy_entry(
hass: HomeAssistant,
legacy_entry_id: str,
) -> None:
"""Cleanup the legacy entry if the migration is successful."""
entity_registry = er.async_get(hass)
if not er.async_entries_for_config_entry(entity_registry, legacy_entry_id):
await hass.config_entries.async_remove(legacy_entry_id)
@callback
def async_migrate_legacy_entries(
hass: HomeAssistant,
hosts_by_mac: dict[str, str],
config_entries_by_mac: dict[str, ConfigEntry],
legacy_entry: ConfigEntry,
) -> None:
"""Migrate the legacy config entries to have an entry per device."""
device_registry = dr.async_get(hass)
for dev_entry in dr.async_entries_for_config_entry(
device_registry, legacy_entry.entry_id
):
for connection_type, mac in dev_entry.connections:
if (
connection_type != dr.CONNECTION_NETWORK_MAC
or mac in config_entries_by_mac
):
continue
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "migration"},
data={
CONF_HOST: hosts_by_mac.get(mac),
CONF_MAC: mac,
CONF_NAME: dev_entry.name or f"TP-Link device {mac}",
},
)
)
async def _async_cleanup_legacy_entry(_now: datetime) -> None:
await async_cleanup_legacy_entry(hass, legacy_entry.entry_id)
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STARTED, _async_cleanup_legacy_entry)
@callback
def async_migrate_yaml_entries(
hass: HomeAssistant, conf: ConfigType | MappingProxyType[str, Any]
) -> None:
"""Migrate yaml to config entries."""
for device_type in (CONF_LIGHT, CONF_SWITCH, CONF_STRIP, CONF_DIMMER):
for device in conf.get(device_type, []):
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_HOST: device[CONF_HOST],
},
)
)
async def async_migrate_entities_devices(
hass: HomeAssistant, legacy_entry_id: str, new_entry: ConfigEntry
) -> None:
"""Move entities and devices to the new config entry."""
migrated_devices = []
device_registry = dr.async_get(hass)
for dev_entry in dr.async_entries_for_config_entry(
device_registry, legacy_entry_id
):
for connection_type, value in dev_entry.connections:
if (
connection_type == dr.CONNECTION_NETWORK_MAC
and value == new_entry.unique_id
):
migrated_devices.append(dev_entry.id)
device_registry.async_update_device(
dev_entry.id, add_config_entry_id=new_entry.entry_id
)
entity_registry = er.async_get(hass)
for reg_entity in er.async_entries_for_config_entry(
entity_registry, legacy_entry_id
):
if reg_entity.device_id in migrated_devices:
entity_registry.async_update_entity(
reg_entity.entity_id, config_entry_id=new_entry.entry_id
)

View file

@ -3,7 +3,7 @@ from unittest.mock import patch
import pytest
from homeassistant import config_entries, setup
from homeassistant import config_entries
from homeassistant.components import dhcp
from homeassistant.components.tplink import DOMAIN
from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_MAC, CONF_NAME
@ -175,52 +175,6 @@ async def test_discovery_no_device(hass: HomeAssistant):
assert result2["reason"] == "no_devices_found"
async def test_import(hass: HomeAssistant):
"""Test import from yaml."""
config = {
CONF_HOST: IP_ADDRESS,
}
# Cannot connect
with _patch_discovery(no_device=True), _patch_single_discovery(no_device=True):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == "abort"
assert result["reason"] == "cannot_connect"
# Success
with _patch_discovery(), _patch_single_discovery(), patch(
f"{MODULE}.async_setup", return_value=True
) as mock_setup, patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == DEFAULT_ENTRY_TITLE
assert result["data"] == {
CONF_HOST: IP_ADDRESS,
}
mock_setup.assert_called_once()
mock_setup_entry.assert_called_once()
# Duplicate
with _patch_discovery(), _patch_single_discovery():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
async def test_manual(hass: HomeAssistant):
"""Test manually setup."""
result = await hass.config_entries.flow.async_init(
@ -406,76 +360,3 @@ async def test_discovered_by_dhcp_or_discovery_failed_to_get_device(hass, source
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_migration_device_online(hass: HomeAssistant):
"""Test migration from single config entry."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
config = {CONF_MAC: MAC_ADDRESS, CONF_NAME: ALIAS, CONF_HOST: IP_ADDRESS}
with _patch_discovery(), _patch_single_discovery(), patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_setup_entry:
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "migration"}, data=config
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == ALIAS
assert result["data"] == {
CONF_HOST: IP_ADDRESS,
}
assert len(mock_setup_entry.mock_calls) == 2
# Duplicate
with _patch_discovery(), _patch_single_discovery():
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "migration"}, data=config
)
await hass.async_block_till_done()
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
async def test_migration_device_offline(hass: HomeAssistant):
"""Test migration from single config entry."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
config = {CONF_MAC: MAC_ADDRESS, CONF_NAME: ALIAS, CONF_HOST: None}
with _patch_discovery(no_device=True), _patch_single_discovery(
no_device=True
), patch(f"{MODULE}.async_setup_entry", return_value=True) as mock_setup_entry:
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "migration"}, data=config
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == ALIAS
new_entry = result["result"]
assert result["data"] == {
CONF_HOST: None,
}
assert len(mock_setup_entry.mock_calls) == 2
# Ensure a manual import updates the missing host
config = {CONF_HOST: IP_ADDRESS}
with _patch_discovery(no_device=True), _patch_single_discovery():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert new_entry.data[CONF_HOST] == IP_ADDRESS

View file

@ -74,37 +74,6 @@ async def test_config_entry_retry(hass):
assert already_migrated_config_entry.state == ConfigEntryState.SETUP_RETRY
async def test_dimmer_switch_unique_id_fix_original_entity_was_deleted(
hass: HomeAssistant, entity_reg: EntityRegistry
):
"""Test that roll out unique id entity id changed to the original unique id."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=MAC_ADDRESS)
config_entry.add_to_hass(hass)
dimmer = _mocked_dimmer()
rollout_unique_id = MAC_ADDRESS.replace(":", "").upper()
original_unique_id = tplink.legacy_device_id(dimmer)
rollout_dimmer_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="light",
unique_id=rollout_unique_id,
original_name="Rollout dimmer",
)
with _patch_discovery(device=dimmer), _patch_single_discovery(device=dimmer):
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
migrated_dimmer_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="light",
unique_id=original_unique_id,
original_name="Migrated dimmer",
)
assert migrated_dimmer_entity_reg.entity_id == rollout_dimmer_entity_reg.entity_id
async def test_dimmer_switch_unique_id_fix_original_entity_still_exists(
hass: HomeAssistant, entity_reg: EntityRegistry
):

View file

@ -1,263 +0,0 @@
"""Test the tplink config flow."""
from homeassistant import setup
from homeassistant.components.tplink import CONF_DISCOVERY, CONF_SWITCH, DOMAIN
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STARTED
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.entity_registry import EntityRegistry
from . import ALIAS, IP_ADDRESS, MAC_ADDRESS, _patch_discovery, _patch_single_discovery
from tests.common import MockConfigEntry
async def test_migration_device_online_end_to_end(
hass: HomeAssistant, device_reg: DeviceRegistry, entity_reg: EntityRegistry
):
"""Test migration from single config entry."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
device = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, MAC_ADDRESS)},
name=ALIAS,
)
switch_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="switch",
unique_id=MAC_ADDRESS,
original_name=ALIAS,
device_id=device.id,
)
light_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="light",
unique_id=dr.format_mac(MAC_ADDRESS),
original_name=ALIAS,
device_id=device.id,
)
power_sensor_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="sensor",
unique_id=f"{MAC_ADDRESS}_sensor",
original_name=ALIAS,
device_id=device.id,
)
with _patch_discovery(), _patch_single_discovery():
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
migrated_entry = None
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.unique_id == DOMAIN:
migrated_entry = entry
break
assert migrated_entry is not None
assert device.config_entries == {migrated_entry.entry_id}
assert light_entity_reg.config_entry_id == migrated_entry.entry_id
assert switch_entity_reg.config_entry_id == migrated_entry.entry_id
assert power_sensor_entity_reg.config_entry_id == migrated_entry.entry_id
assert er.async_entries_for_config_entry(entity_reg, config_entry) == []
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
legacy_entry = None
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.unique_id == DOMAIN:
legacy_entry = entry
break
assert legacy_entry is None
async def test_migration_device_online_end_to_end_after_downgrade(
hass: HomeAssistant, device_reg: DeviceRegistry, entity_reg: EntityRegistry
):
"""Test migration from single config entry can happen again after a downgrade."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
already_migrated_config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=MAC_ADDRESS
)
already_migrated_config_entry.add_to_hass(hass)
device = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, MAC_ADDRESS)},
name=ALIAS,
)
light_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="light",
unique_id=MAC_ADDRESS,
original_name=ALIAS,
device_id=device.id,
)
power_sensor_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="sensor",
unique_id=f"{MAC_ADDRESS}_sensor",
original_name=ALIAS,
device_id=device.id,
)
with _patch_discovery(), _patch_single_discovery():
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
assert device.config_entries == {config_entry.entry_id}
assert light_entity_reg.config_entry_id == config_entry.entry_id
assert power_sensor_entity_reg.config_entry_id == config_entry.entry_id
assert er.async_entries_for_config_entry(entity_reg, config_entry) == []
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
legacy_entry = None
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.unique_id == DOMAIN:
legacy_entry = entry
break
assert legacy_entry is None
async def test_migration_device_online_end_to_end_ignores_other_devices(
hass: HomeAssistant, device_reg: DeviceRegistry, entity_reg: EntityRegistry
):
"""Test migration from single config entry."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
other_domain_config_entry = MockConfigEntry(
domain="other_domain", data={}, unique_id="other_domain"
)
other_domain_config_entry.add_to_hass(hass)
device = device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, MAC_ADDRESS)},
name=ALIAS,
)
other_device = device_reg.async_get_or_create(
config_entry_id=other_domain_config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "556655665566")},
name=ALIAS,
)
light_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="light",
unique_id=MAC_ADDRESS,
original_name=ALIAS,
device_id=device.id,
)
power_sensor_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="sensor",
unique_id=f"{MAC_ADDRESS}_sensor",
original_name=ALIAS,
device_id=device.id,
)
ignored_entity_reg = entity_reg.async_get_or_create(
config_entry=other_domain_config_entry,
platform=DOMAIN,
domain="sensor",
unique_id="00:00:00:00:00:00_sensor",
original_name=ALIAS,
device_id=device.id,
)
garbage_entity_reg = entity_reg.async_get_or_create(
config_entry=config_entry,
platform=DOMAIN,
domain="sensor",
unique_id="garbage",
original_name=ALIAS,
device_id=other_device.id,
)
with _patch_discovery(), _patch_single_discovery():
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
migrated_entry = None
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.unique_id == DOMAIN:
migrated_entry = entry
break
assert migrated_entry is not None
assert device.config_entries == {migrated_entry.entry_id}
assert light_entity_reg.config_entry_id == migrated_entry.entry_id
assert power_sensor_entity_reg.config_entry_id == migrated_entry.entry_id
assert ignored_entity_reg.config_entry_id == other_domain_config_entry.entry_id
assert garbage_entity_reg.config_entry_id == config_entry.entry_id
assert er.async_entries_for_config_entry(entity_reg, config_entry) == []
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
legacy_entry = None
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.unique_id == DOMAIN:
legacy_entry = entry
break
assert legacy_entry is not None
async def test_migrate_from_yaml(hass: HomeAssistant):
"""Test migrate from yaml."""
config = {
DOMAIN: {
CONF_DISCOVERY: False,
CONF_SWITCH: [{CONF_HOST: IP_ADDRESS}],
}
}
with _patch_discovery(), _patch_single_discovery():
await setup.async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
migrated_entry = None
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.unique_id == MAC_ADDRESS:
migrated_entry = entry
break
assert migrated_entry is not None
assert migrated_entry.data[CONF_HOST] == IP_ADDRESS
async def test_migrate_from_legacy_entry(hass: HomeAssistant):
"""Test migrate from legacy entry that was already imported from yaml."""
data = {
CONF_DISCOVERY: False,
CONF_SWITCH: [{CONF_HOST: IP_ADDRESS}],
}
config_entry = MockConfigEntry(domain=DOMAIN, data=data, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_single_discovery():
await setup.async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
migrated_entry = None
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.unique_id == MAC_ADDRESS:
migrated_entry = entry
break
assert migrated_entry is not None
assert migrated_entry.data[CONF_HOST] == IP_ADDRESS