evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Remove legacy zwave migration logic (#72206)

Browse source 
* Remove legacy zwave migration logic

* Update test_migrate.py
This commit is contained in:
Raman Gupta 2022-05-19 18:29:28 -04:00 • committed by GitHub
parent ebc883b43f
commit 68b278d170
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 4 additions and 883 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
homeassistant/components/zwave_js/api.py
View file

@ -66,19 +66,12 @@ from .const import (
DATA_CLIENT,
DOMAIN,
EVENT_DEVICE_ADDED_TO_REGISTRY,
LOGGER,
)
from .helpers import (
async_enable_statistics,
async_get_node_from_device_id,
update_data_collection_preference,
)
from .migrate import (
ZWaveMigrationData,
async_get_migration_data,
async_map_legacy_zwave_values,
async_migrate_legacy_zwave,
)
DATA_UNSUBSCRIBE = "unsubs"
@ -365,7 +358,6 @@ def async_register_api(hass: HomeAssistant) -> None:
)
websocket_api.async_register_command(hass, websocket_subscribe_node_statistics)
websocket_api.async_register_command(hass, websocket_node_ready)
websocket_api.async_register_command(hass, websocket_migrate_zwave)
hass.http.register_view(FirmwareUploadView())
@ -2059,72 +2051,3 @@ async def websocket_subscribe_node_statistics(
},
)
)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(TYPE): "zwave_js/migrate_zwave",
vol.Required(ENTRY_ID): str,
vol.Optional(DRY_RUN, default=True): bool,
}
)
@websocket_api.async_response
@async_get_entry
async def websocket_migrate_zwave(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
entry: ConfigEntry,
client: Client,
) -> None:
"""Migrate Z-Wave device and entity data to Z-Wave JS integration."""
if "zwave" not in hass.config.components:
connection.send_message(
websocket_api.error_message(
msg["id"], "zwave_not_loaded", "Integration zwave is not loaded"
)
)
return
zwave = hass.components.zwave
zwave_config_entries = hass.config_entries.async_entries("zwave")
zwave_config_entry = zwave_config_entries[0] # zwave only has a single config entry
zwave_data: dict[str, ZWaveMigrationData] = await zwave.async_get_migration_data(
hass, zwave_config_entry
)
LOGGER.debug("Migration zwave data: %s", zwave_data)
zwave_js_config_entry = entry
zwave_js_data = await async_get_migration_data(hass, zwave_js_config_entry)
LOGGER.debug("Migration zwave_js data: %s", zwave_js_data)
migration_map = async_map_legacy_zwave_values(zwave_data, zwave_js_data)
zwave_entity_ids = [entry["entity_id"] for entry in zwave_data.values()]
zwave_js_entity_ids = [entry["entity_id"] for entry in zwave_js_data.values()]
migration_device_map = {
zwave_device_id: zwave_js_device_id
for zwave_js_device_id, zwave_device_id in migration_map.device_entries.items()
}
migration_entity_map = {
zwave_entry["entity_id"]: zwave_js_entity_id
for zwave_js_entity_id, zwave_entry in migration_map.entity_entries.items()
}
LOGGER.debug("Migration entity map: %s", migration_entity_map)
if not msg[DRY_RUN]:
await async_migrate_legacy_zwave(
hass, zwave_config_entry, zwave_js_config_entry, migration_map
)
connection.send_result(
msg[ID],
{
"migration_device_map": migration_device_map,
"zwave_entity_ids": zwave_entity_ids,
"zwave_js_entity_ids": zwave_js_entity_ids,
"migration_entity_map": migration_entity_map,
"migrated": not msg[DRY_RUN],
},
)

6
homeassistant/components/zwave_js/entity.py
View file

@ -15,7 +15,6 @@ from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import DOMAIN
from .discovery import ZwaveDiscoveryInfo
from .helpers import get_device_id, get_unique_id
from .migrate import async_add_migration_entity_value
LOGGER = logging.getLogger(__name__)
@ -117,11 +116,6 @@ class ZWaveBaseEntity(Entity):
)
)
# Add legacy Z-Wave migration data.
await async_add_migration_entity_value(
self.hass, self.config_entry, self.entity_id, self.info
)
def generate_name(
self,
include_value_name: bool = False,

338
homeassistant/components/zwave_js/migrate.py
View file

@ -1,357 +1,27 @@
"""Functions used to migrate unique IDs for Z-Wave JS entities."""
from __future__ import annotations
from dataclasses import dataclass, field
from dataclasses import dataclass
import logging
from typing import TypedDict, cast
from zwave_js_server.client import Client as ZwaveClient
from zwave_js_server.model.value import Value as ZwaveValue
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import LIGHT_LUX, STATE_UNAVAILABLE
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import (
DeviceEntry,
async_get as async_get_device_registry,
)
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.helpers.entity_registry import (
EntityRegistry,
RegistryEntry,
async_entries_for_device,
async_get as async_get_entity_registry,
)
from homeassistant.helpers.singleton import singleton
from homeassistant.helpers.storage import Store
from .const import DOMAIN
from .discovery import ZwaveDiscoveryInfo
from .helpers import get_device_id, get_unique_id
from .helpers import get_unique_id
_LOGGER = logging.getLogger(__name__)
LEGACY_ZWAVE_MIGRATION = f"{DOMAIN}_legacy_zwave_migration"
MIGRATED = "migrated"
STORAGE_WRITE_DELAY = 30
STORAGE_KEY = f"{DOMAIN}.legacy_zwave_migration"
STORAGE_VERSION = 1
NOTIFICATION_CC_LABEL_TO_PROPERTY_NAME = {
"Smoke": "Smoke Alarm",
"Carbon Monoxide": "CO Alarm",
"Carbon Dioxide": "CO2 Alarm",
"Heat": "Heat Alarm",
"Flood": "Water Alarm",
"Access Control": "Access Control",
"Burglar": "Home Security",
"Power Management": "Power Management",
"System": "System",
"Emergency": "Siren",
"Clock": "Clock",
"Appliance": "Appliance",
"HomeHealth": "Home Health",
}
SENSOR_MULTILEVEL_CC_LABEL_TO_PROPERTY_NAME = {
"Temperature": "Air temperature",
"General": "General purpose",
"Luminance": "Illuminance",
"Power": "Power",
"Relative Humidity": "Humidity",
"Velocity": "Velocity",
"Direction": "Direction",
"Atmospheric Pressure": "Atmospheric pressure",
"Barometric Pressure": "Barometric pressure",
"Solar Radiation": "Solar radiation",
"Dew Point": "Dew point",
"Rain Rate": "Rain rate",
"Tide Level": "Tide level",
"Weight": "Weight",
"Voltage": "Voltage",
"Current": "Current",
"CO2 Level": "Carbon dioxide (CO₂) level",
"Air Flow": "Air flow",
"Tank Capacity": "Tank capacity",
"Distance": "Distance",
"Angle Position": "Angle position",
"Rotation": "Rotation",
"Water Temperature": "Water temperature",
"Soil Temperature": "Soil temperature",
"Seismic Intensity": "Seismic Intensity",
"Seismic Magnitude": "Seismic magnitude",
"Ultraviolet": "Ultraviolet",
"Electrical Resistivity": "Electrical resistivity",
"Electrical Conductivity": "Electrical conductivity",
"Loudness": "Loudness",
"Moisture": "Moisture",
}
CC_ID_LABEL_TO_PROPERTY = {
49: SENSOR_MULTILEVEL_CC_LABEL_TO_PROPERTY_NAME,
113: NOTIFICATION_CC_LABEL_TO_PROPERTY_NAME,
}
UNIT_LEGACY_MIGRATION_MAP = {LIGHT_LUX: "Lux"}
class ZWaveMigrationData(TypedDict):
"""Represent the Z-Wave migration data dict."""
node_id: int
node_instance: int
command_class: int
command_class_label: str
value_index: int
device_id: str
domain: str
entity_id: str
unique_id: str
unit_of_measurement: str | None
class ZWaveJSMigrationData(TypedDict):
"""Represent the Z-Wave JS migration data dict."""
node_id: int
endpoint_index: int
command_class: int
value_property_name: str
value_property_key_name: str | None
value_id: str
device_id: str
domain: str
entity_id: str
unique_id: str
unit_of_measurement: str | None
@dataclass
class LegacyZWaveMappedData:
"""Represent the mapped data between Z-Wave and Z-Wave JS."""
entity_entries: dict[str, ZWaveMigrationData] = field(default_factory=dict)
device_entries: dict[str, str] = field(default_factory=dict)
async def async_add_migration_entity_value(
hass: HomeAssistant,
config_entry: ConfigEntry,
entity_id: str,
discovery_info: ZwaveDiscoveryInfo,
) -> None:
"""Add Z-Wave JS entity value for legacy Z-Wave migration."""
migration_handler: LegacyZWaveMigration = await get_legacy_zwave_migration(hass)
migration_handler.add_entity_value(config_entry, entity_id, discovery_info)
async def async_get_migration_data(
hass: HomeAssistant, config_entry: ConfigEntry
) -> dict[str, ZWaveJSMigrationData]:
"""Return Z-Wave JS migration data."""
migration_handler: LegacyZWaveMigration = await get_legacy_zwave_migration(hass)
return await migration_handler.get_data(config_entry)
@singleton(LEGACY_ZWAVE_MIGRATION)
async def get_legacy_zwave_migration(hass: HomeAssistant) -> LegacyZWaveMigration:
"""Return legacy Z-Wave migration handler."""
migration_handler = LegacyZWaveMigration(hass)
await migration_handler.load_data()
return migration_handler
class LegacyZWaveMigration:
"""Handle the migration from zwave to zwave_js."""
def __init__(self, hass: HomeAssistant) -> None:
"""Set up migration instance."""
self._hass = hass
self._store = Store(hass, STORAGE_VERSION, STORAGE_KEY)
self._data: dict[str, dict[str, ZWaveJSMigrationData]] = {}
async def load_data(self) -> None:
"""Load Z-Wave JS migration data."""
stored = cast(dict, await self._store.async_load())
if stored:
self._data = stored
@callback
def save_data(
self, config_entry_id: str, entity_id: str, data: ZWaveJSMigrationData
) -> None:
"""Save Z-Wave JS migration data."""
if config_entry_id not in self._data:
self._data[config_entry_id] = {}
self._data[config_entry_id][entity_id] = data
self._store.async_delay_save(self._data_to_save, STORAGE_WRITE_DELAY)
@callback
def _data_to_save(self) -> dict[str, dict[str, ZWaveJSMigrationData]]:
"""Return data to save."""
return self._data
@callback
def add_entity_value(
self,
config_entry: ConfigEntry,
entity_id: str,
discovery_info: ZwaveDiscoveryInfo,
) -> None:
"""Add info for one entity and Z-Wave JS value."""
ent_reg = async_get_entity_registry(self._hass)
dev_reg = async_get_device_registry(self._hass)
node = discovery_info.node
primary_value = discovery_info.primary_value
entity_entry = ent_reg.async_get(entity_id)
assert entity_entry
device_identifier = get_device_id(node.client, node)
device_entry = dev_reg.async_get_device({device_identifier}, set())
assert device_entry
# Normalize unit of measurement.
if unit := entity_entry.unit_of_measurement:
_unit = UNIT_LEGACY_MIGRATION_MAP.get(unit, unit)
unit = _unit.lower()
if unit == "":
unit = None
data: ZWaveJSMigrationData = {
"node_id": node.node_id,
"endpoint_index": node.index,
"command_class": primary_value.command_class,
"value_property_name": primary_value.property_name,
"value_property_key_name": primary_value.property_key_name,
"value_id": primary_value.value_id,
"device_id": device_entry.id,
"domain": entity_entry.domain,
"entity_id": entity_id,
"unique_id": entity_entry.unique_id,
"unit_of_measurement": unit,
}
self.save_data(config_entry.entry_id, entity_id, data)
async def get_data(
self, config_entry: ConfigEntry
) -> dict[str, ZWaveJSMigrationData]:
"""Return Z-Wave JS migration data for a config entry."""
await self.load_data()
data = self._data.get(config_entry.entry_id)
return data or {}
@callback
def async_map_legacy_zwave_values(
zwave_data: dict[str, ZWaveMigrationData],
zwave_js_data: dict[str, ZWaveJSMigrationData],
) -> LegacyZWaveMappedData:
"""Map Z-Wave node values onto Z-Wave JS node values."""
migration_map = LegacyZWaveMappedData()
zwave_proc_data: dict[
tuple[int, int, int, str, str | None, str | None],
ZWaveMigrationData | None,
] = {}
zwave_js_proc_data: dict[
tuple[int, int, int, str, str | None, str | None],
ZWaveJSMigrationData | None,
] = {}
for zwave_item in zwave_data.values():
zwave_js_property_name = CC_ID_LABEL_TO_PROPERTY.get(
zwave_item["command_class"], {}
).get(zwave_item["command_class_label"])
item_id = (
zwave_item["node_id"],
zwave_item["command_class"],
zwave_item["node_instance"] - 1,
zwave_item["domain"],
zwave_item["unit_of_measurement"],
zwave_js_property_name,
)
# Filter out duplicates that are not resolvable.
if item_id in zwave_proc_data:
zwave_proc_data[item_id] = None
continue
zwave_proc_data[item_id] = zwave_item
for zwave_js_item in zwave_js_data.values():
# Only identify with property name if there is a command class label map.
if zwave_js_item["command_class"] in CC_ID_LABEL_TO_PROPERTY:
zwave_js_property_name = zwave_js_item["value_property_name"]
else:
zwave_js_property_name = None
item_id = (
zwave_js_item["node_id"],
zwave_js_item["command_class"],
zwave_js_item["endpoint_index"],
zwave_js_item["domain"],
zwave_js_item["unit_of_measurement"],
zwave_js_property_name,
)
# Filter out duplicates that are not resolvable.
if item_id in zwave_js_proc_data:
zwave_js_proc_data[item_id] = None
continue
zwave_js_proc_data[item_id] = zwave_js_item
for item_id, zwave_entry in zwave_proc_data.items():
zwave_js_entry = zwave_js_proc_data.pop(item_id, None)
if zwave_entry is None or zwave_js_entry is None:
continue
migration_map.entity_entries[zwave_js_entry["entity_id"]] = zwave_entry
migration_map.device_entries[zwave_js_entry["device_id"]] = zwave_entry[
"device_id"
]
return migration_map
async def async_migrate_legacy_zwave(
hass: HomeAssistant,
zwave_config_entry: ConfigEntry,
zwave_js_config_entry: ConfigEntry,
migration_map: LegacyZWaveMappedData,
) -> None:
"""Perform Z-Wave to Z-Wave JS migration."""
dev_reg = async_get_device_registry(hass)
for zwave_js_device_id, zwave_device_id in migration_map.device_entries.items():
zwave_device_entry = dev_reg.async_get(zwave_device_id)
if not zwave_device_entry:
continue
dev_reg.async_update_device(
zwave_js_device_id,
area_id=zwave_device_entry.area_id,
name_by_user=zwave_device_entry.name_by_user,
)
ent_reg = async_get_entity_registry(hass)
for zwave_js_entity_id, zwave_entry in migration_map.entity_entries.items():
zwave_entity_id = zwave_entry["entity_id"]
if not (entity_entry := ent_reg.async_get(zwave_entity_id)):
continue
ent_reg.async_remove(zwave_entity_id)
ent_reg.async_update_entity(
zwave_js_entity_id,
new_entity_id=entity_entry.entity_id,
name=entity_entry.name,
icon=entity_entry.icon,
)
await hass.config_entries.async_remove(zwave_config_entry.entry_id)
updates = {
**zwave_js_config_entry.data,
MIGRATED: True,
}
hass.config_entries.async_update_entry(zwave_js_config_entry, data=updates)
@dataclass
class ValueID:

466
tests/components/zwave_js/test_migrate.py
View file

@ -1,481 +1,15 @@
"""Test the Z-Wave JS migration module."""
import copy
from unittest.mock import patch
import pytest
from zwave_js_server.model.node import Node
from homeassistant.components.zwave_js.api import ENTRY_ID, ID, TYPE
from homeassistant.components.zwave_js.const import DOMAIN
from homeassistant.components.zwave_js.helpers import get_device_id
from homeassistant.const import LIGHT_LUX
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .common import AIR_TEMPERATURE_SENSOR, NOTIFICATION_MOTION_BINARY_SENSOR
from tests.common import MockConfigEntry, mock_device_registry, mock_registry
# Switch device
ZWAVE_SWITCH_DEVICE_ID = "zwave_switch_device_id"
ZWAVE_SWITCH_DEVICE_NAME = "Z-Wave Switch Device"
ZWAVE_SWITCH_DEVICE_AREA = "Z-Wave Switch Area"
ZWAVE_SWITCH_ENTITY = "switch.zwave_switch_node"
ZWAVE_SWITCH_UNIQUE_ID = "102-6789"
ZWAVE_SWITCH_NAME = "Z-Wave Switch"
ZWAVE_SWITCH_ICON = "mdi:zwave-test-switch"
ZWAVE_POWER_ENTITY = "sensor.zwave_power"
ZWAVE_POWER_UNIQUE_ID = "102-5678"
ZWAVE_POWER_NAME = "Z-Wave Power"
ZWAVE_POWER_ICON = "mdi:zwave-test-power"
# Multisensor device
ZWAVE_MULTISENSOR_DEVICE_ID = "zwave_multisensor_device_id"
ZWAVE_MULTISENSOR_DEVICE_NAME = "Z-Wave Multisensor Device"
ZWAVE_MULTISENSOR_DEVICE_AREA = "Z-Wave Multisensor Area"
ZWAVE_SOURCE_NODE_ENTITY = "sensor.zwave_source_node"
ZWAVE_SOURCE_NODE_UNIQUE_ID = "52-4321"
ZWAVE_LUMINANCE_ENTITY = "sensor.zwave_luminance"
ZWAVE_LUMINANCE_UNIQUE_ID = "52-6543"
ZWAVE_LUMINANCE_NAME = "Z-Wave Luminance"
ZWAVE_LUMINANCE_ICON = "mdi:zwave-test-luminance"
ZWAVE_BATTERY_ENTITY = "sensor.zwave_battery_level"
ZWAVE_BATTERY_UNIQUE_ID = "52-1234"
ZWAVE_BATTERY_NAME = "Z-Wave Battery Level"
ZWAVE_BATTERY_ICON = "mdi:zwave-test-battery"
ZWAVE_TAMPERING_ENTITY = "sensor.zwave_tampering"
ZWAVE_TAMPERING_UNIQUE_ID = "52-3456"
ZWAVE_TAMPERING_NAME = "Z-Wave Tampering"
ZWAVE_TAMPERING_ICON = "mdi:zwave-test-tampering"
@pytest.fixture(name="zwave_migration_data")
def zwave_migration_data_fixture(hass):
"""Return mock zwave migration data."""
zwave_switch_device = dr.DeviceEntry(
id=ZWAVE_SWITCH_DEVICE_ID,
name_by_user=ZWAVE_SWITCH_DEVICE_NAME,
area_id=ZWAVE_SWITCH_DEVICE_AREA,
)
zwave_switch_entry = er.RegistryEntry(
entity_id=ZWAVE_SWITCH_ENTITY,
unique_id=ZWAVE_SWITCH_UNIQUE_ID,
platform="zwave",
name=ZWAVE_SWITCH_NAME,
icon=ZWAVE_SWITCH_ICON,
)
zwave_multisensor_device = dr.DeviceEntry(
id=ZWAVE_MULTISENSOR_DEVICE_ID,
name_by_user=ZWAVE_MULTISENSOR_DEVICE_NAME,
area_id=ZWAVE_MULTISENSOR_DEVICE_AREA,
)
zwave_source_node_entry = er.RegistryEntry(
entity_id=ZWAVE_SOURCE_NODE_ENTITY,
unique_id=ZWAVE_SOURCE_NODE_UNIQUE_ID,
platform="zwave",
name="Z-Wave Source Node",
)
zwave_luminance_entry = er.RegistryEntry(
entity_id=ZWAVE_LUMINANCE_ENTITY,
unique_id=ZWAVE_LUMINANCE_UNIQUE_ID,
platform="zwave",
name=ZWAVE_LUMINANCE_NAME,
icon=ZWAVE_LUMINANCE_ICON,
unit_of_measurement="lux",
)
zwave_battery_entry = er.RegistryEntry(
entity_id=ZWAVE_BATTERY_ENTITY,
unique_id=ZWAVE_BATTERY_UNIQUE_ID,
platform="zwave",
name=ZWAVE_BATTERY_NAME,
icon=ZWAVE_BATTERY_ICON,
unit_of_measurement="%",
)
zwave_power_entry = er.RegistryEntry(
entity_id=ZWAVE_POWER_ENTITY,
unique_id=ZWAVE_POWER_UNIQUE_ID,
platform="zwave",
name=ZWAVE_POWER_NAME,
icon=ZWAVE_POWER_ICON,
unit_of_measurement="W",
)
zwave_tampering_entry = er.RegistryEntry(
entity_id=ZWAVE_TAMPERING_ENTITY,
unique_id=ZWAVE_TAMPERING_UNIQUE_ID,
platform="zwave",
name=ZWAVE_TAMPERING_NAME,
icon=ZWAVE_TAMPERING_ICON,
unit_of_measurement="", # Test empty string unit normalization.
)
zwave_migration_data = {
ZWAVE_SWITCH_ENTITY: {
"node_id": 102,
"node_instance": 1,
"command_class": 37,
"command_class_label": "",
"value_index": 1,
"device_id": zwave_switch_device.id,
"domain": zwave_switch_entry.domain,
"entity_id": zwave_switch_entry.entity_id,
"unique_id": ZWAVE_SWITCH_UNIQUE_ID,
"unit_of_measurement": zwave_switch_entry.unit_of_measurement,
},
ZWAVE_POWER_ENTITY: {
"node_id": 102,
"node_instance": 1,
"command_class": 50,
"command_class_label": "Power",
"value_index": 8,
"device_id": zwave_switch_device.id,
"domain": zwave_power_entry.domain,
"entity_id": zwave_power_entry.entity_id,
"unique_id": ZWAVE_POWER_UNIQUE_ID,
"unit_of_measurement": zwave_power_entry.unit_of_measurement,
},
ZWAVE_SOURCE_NODE_ENTITY: {
"node_id": 52,
"node_instance": 1,
"command_class": 113,
"command_class_label": "SourceNodeId",
"value_index": 1,
"device_id": zwave_multisensor_device.id,
"domain": zwave_source_node_entry.domain,
"entity_id": zwave_source_node_entry.entity_id,
"unique_id": ZWAVE_SOURCE_NODE_UNIQUE_ID,
"unit_of_measurement": zwave_source_node_entry.unit_of_measurement,
},
ZWAVE_LUMINANCE_ENTITY: {
"node_id": 52,
"node_instance": 1,
"command_class": 49,
"command_class_label": "Luminance",
"value_index": 3,
"device_id": zwave_multisensor_device.id,
"domain": zwave_luminance_entry.domain,
"entity_id": zwave_luminance_entry.entity_id,
"unique_id": ZWAVE_LUMINANCE_UNIQUE_ID,
"unit_of_measurement": zwave_luminance_entry.unit_of_measurement,
},
ZWAVE_BATTERY_ENTITY: {
"node_id": 52,
"node_instance": 1,
"command_class": 128,
"command_class_label": "Battery Level",
"value_index": 0,
"device_id": zwave_multisensor_device.id,
"domain": zwave_battery_entry.domain,
"entity_id": zwave_battery_entry.entity_id,
"unique_id": ZWAVE_BATTERY_UNIQUE_ID,
"unit_of_measurement": zwave_battery_entry.unit_of_measurement,
},
ZWAVE_TAMPERING_ENTITY: {
"node_id": 52,
"node_instance": 1,
"command_class": 113,
"command_class_label": "Burglar",
"value_index": 10,
"device_id": zwave_multisensor_device.id,
"domain": zwave_tampering_entry.domain,
"entity_id": zwave_tampering_entry.entity_id,
"unique_id": ZWAVE_TAMPERING_UNIQUE_ID,
"unit_of_measurement": zwave_tampering_entry.unit_of_measurement,
},
}
mock_device_registry(
hass,
{
zwave_switch_device.id: zwave_switch_device,
zwave_multisensor_device.id: zwave_multisensor_device,
},
)
mock_registry(
hass,
{
ZWAVE_SWITCH_ENTITY: zwave_switch_entry,
ZWAVE_SOURCE_NODE_ENTITY: zwave_source_node_entry,
ZWAVE_LUMINANCE_ENTITY: zwave_luminance_entry,
ZWAVE_BATTERY_ENTITY: zwave_battery_entry,
ZWAVE_POWER_ENTITY: zwave_power_entry,
ZWAVE_TAMPERING_ENTITY: zwave_tampering_entry,
},
)
return zwave_migration_data
@pytest.fixture(name="zwave_integration")
def zwave_integration_fixture(hass, zwave_migration_data):
"""Mock the zwave integration."""
hass.config.components.add("zwave")
zwave_config_entry = MockConfigEntry(domain="zwave", data={"usb_path": "/dev/test"})
zwave_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.zwave.async_get_migration_data",
return_value=zwave_migration_data,
):
yield zwave_config_entry
@pytest.mark.skip(reason="The old zwave integration has been removed.")
async def test_migrate_zwave(
hass,
zwave_integration,
aeon_smart_switch_6,
multisensor_6,
integration,
hass_ws_client,
):
"""Test the Z-Wave to Z-Wave JS migration websocket api."""
entry = integration
client = await hass_ws_client(hass)
assert hass.config_entries.async_entries("zwave")
await client.send_json(
{
ID: 5,
TYPE: "zwave_js/migrate_zwave",
ENTRY_ID: entry.entry_id,
"dry_run": False,
}
)
msg = await client.receive_json()
result = msg["result"]
migration_entity_map = {
ZWAVE_SWITCH_ENTITY: "switch.smart_switch_6",
ZWAVE_LUMINANCE_ENTITY: "sensor.multisensor_6_illuminance",
ZWAVE_BATTERY_ENTITY: "sensor.multisensor_6_battery_level",
}
assert result["zwave_entity_ids"] == [
ZWAVE_SWITCH_ENTITY,
ZWAVE_POWER_ENTITY,
ZWAVE_SOURCE_NODE_ENTITY,
ZWAVE_LUMINANCE_ENTITY,
ZWAVE_BATTERY_ENTITY,
ZWAVE_TAMPERING_ENTITY,
]
expected_zwave_js_entities = [
"switch.smart_switch_6",
"sensor.multisensor_6_air_temperature",
"sensor.multisensor_6_illuminance",
"sensor.multisensor_6_humidity",
"sensor.multisensor_6_ultraviolet",
"binary_sensor.multisensor_6_home_security_tampering_product_cover_removed",
"binary_sensor.multisensor_6_home_security_motion_detection",
"sensor.multisensor_6_battery_level",
"binary_sensor.multisensor_6_low_battery_level",
"light.smart_switch_6",
"sensor.smart_switch_6_electric_consumed_kwh",
"sensor.smart_switch_6_electric_consumed_w",
"sensor.smart_switch_6_electric_consumed_v",
"sensor.smart_switch_6_electric_consumed_a",
]
# Assert that both lists have the same items without checking order
assert not set(result["zwave_js_entity_ids"]) ^ set(expected_zwave_js_entities)
assert result["migration_entity_map"] == migration_entity_map
assert result["migrated"] is True
dev_reg = dr.async_get(hass)
ent_reg = er.async_get(hass)
# check the device registry migration
# check that the migrated entries have correct attributes
multisensor_device_entry = dev_reg.async_get_device(
identifiers={("zwave_js", "3245146787-52")}, connections=set()
)
assert multisensor_device_entry
assert multisensor_device_entry.name_by_user == ZWAVE_MULTISENSOR_DEVICE_NAME
assert multisensor_device_entry.area_id == ZWAVE_MULTISENSOR_DEVICE_AREA
switch_device_entry = dev_reg.async_get_device(
identifiers={("zwave_js", "3245146787-102")}, connections=set()
)
assert switch_device_entry
assert switch_device_entry.name_by_user == ZWAVE_SWITCH_DEVICE_NAME
assert switch_device_entry.area_id == ZWAVE_SWITCH_DEVICE_AREA
migration_device_map = {
ZWAVE_SWITCH_DEVICE_ID: switch_device_entry.id,
ZWAVE_MULTISENSOR_DEVICE_ID: multisensor_device_entry.id,
}
assert result["migration_device_map"] == migration_device_map
# check the entity registry migration
# this should have been migrated and no longer present under that id
assert not ent_reg.async_is_registered("sensor.multisensor_6_battery_level")
assert not ent_reg.async_is_registered("sensor.multisensor_6_illuminance")
# these should not have been migrated and is still in the registry
assert ent_reg.async_is_registered(ZWAVE_SOURCE_NODE_ENTITY)
source_entry = ent_reg.async_get(ZWAVE_SOURCE_NODE_ENTITY)
assert source_entry.unique_id == ZWAVE_SOURCE_NODE_UNIQUE_ID
assert ent_reg.async_is_registered(ZWAVE_POWER_ENTITY)
source_entry = ent_reg.async_get(ZWAVE_POWER_ENTITY)
assert source_entry.unique_id == ZWAVE_POWER_UNIQUE_ID
assert ent_reg.async_is_registered(ZWAVE_TAMPERING_ENTITY)
tampering_entry = ent_reg.async_get(ZWAVE_TAMPERING_ENTITY)
assert tampering_entry.unique_id == ZWAVE_TAMPERING_UNIQUE_ID
assert ent_reg.async_is_registered("sensor.smart_switch_6_electric_consumed_w")
# this is the new entity_ids of the zwave_js entities
assert ent_reg.async_is_registered(ZWAVE_SWITCH_ENTITY)
assert ent_reg.async_is_registered(ZWAVE_BATTERY_ENTITY)
assert ent_reg.async_is_registered(ZWAVE_LUMINANCE_ENTITY)
# check that the migrated entries have correct attributes
switch_entry = ent_reg.async_get(ZWAVE_SWITCH_ENTITY)
assert switch_entry
assert switch_entry.unique_id == "3245146787.102-37-0-currentValue"
assert switch_entry.name == ZWAVE_SWITCH_NAME
assert switch_entry.icon == ZWAVE_SWITCH_ICON
battery_entry = ent_reg.async_get(ZWAVE_BATTERY_ENTITY)
assert battery_entry
assert battery_entry.unique_id == "3245146787.52-128-0-level"
assert battery_entry.name == ZWAVE_BATTERY_NAME
assert battery_entry.icon == ZWAVE_BATTERY_ICON
luminance_entry = ent_reg.async_get(ZWAVE_LUMINANCE_ENTITY)
assert luminance_entry
assert luminance_entry.unique_id == "3245146787.52-49-0-Illuminance"
assert luminance_entry.name == ZWAVE_LUMINANCE_NAME
assert luminance_entry.icon == ZWAVE_LUMINANCE_ICON
assert luminance_entry.unit_of_measurement == LIGHT_LUX
# check that the zwave config entry has been removed
assert not hass.config_entries.async_entries("zwave")
# Check that the zwave integration fails entry setup after migration
zwave_config_entry = MockConfigEntry(domain="zwave")
zwave_config_entry.add_to_hass(hass)
assert not await hass.config_entries.async_setup(zwave_config_entry.entry_id)
@pytest.mark.skip(reason="The old zwave integration has been removed.")
async def test_migrate_zwave_dry_run(
hass,
zwave_integration,
aeon_smart_switch_6,
multisensor_6,
integration,
hass_ws_client,
):
"""Test the zwave to zwave_js migration websocket api dry run."""
entry = integration
client = await hass_ws_client(hass)
await client.send_json(
{ID: 5, TYPE: "zwave_js/migrate_zwave", ENTRY_ID: entry.entry_id}
)
msg = await client.receive_json()
result = msg["result"]
migration_entity_map = {
ZWAVE_SWITCH_ENTITY: "switch.smart_switch_6",
ZWAVE_BATTERY_ENTITY: "sensor.multisensor_6_battery_level",
}
assert result["zwave_entity_ids"] == [
ZWAVE_SWITCH_ENTITY,
ZWAVE_POWER_ENTITY,
ZWAVE_SOURCE_NODE_ENTITY,
ZWAVE_BATTERY_ENTITY,
ZWAVE_TAMPERING_ENTITY,
]
expected_zwave_js_entities = [
"switch.smart_switch_6",
"sensor.multisensor_6_air_temperature",
"sensor.multisensor_6_illuminance",
"sensor.multisensor_6_humidity",
"sensor.multisensor_6_ultraviolet",
"binary_sensor.multisensor_6_home_security_tampering_product_cover_removed",
"binary_sensor.multisensor_6_home_security_motion_detection",
"sensor.multisensor_6_battery_level",
"binary_sensor.multisensor_6_low_battery_level",
"light.smart_switch_6",
"sensor.smart_switch_6_electric_consumed_kwh",
"sensor.smart_switch_6_electric_consumed_w",
"sensor.smart_switch_6_electric_consumed_v",
"sensor.smart_switch_6_electric_consumed_a",
]
# Assert that both lists have the same items without checking order
assert not set(result["zwave_js_entity_ids"]) ^ set(expected_zwave_js_entities)
assert result["migration_entity_map"] == migration_entity_map
dev_reg = dr.async_get(hass)
multisensor_device_entry = dev_reg.async_get_device(
identifiers={("zwave_js", "3245146787-52")}, connections=set()
)
assert multisensor_device_entry
assert multisensor_device_entry.name_by_user is None
assert multisensor_device_entry.area_id is None
switch_device_entry = dev_reg.async_get_device(
identifiers={("zwave_js", "3245146787-102")}, connections=set()
)
assert switch_device_entry
assert switch_device_entry.name_by_user is None
assert switch_device_entry.area_id is None
migration_device_map = {
ZWAVE_SWITCH_DEVICE_ID: switch_device_entry.id,
ZWAVE_MULTISENSOR_DEVICE_ID: multisensor_device_entry.id,
}
assert result["migration_device_map"] == migration_device_map
assert result["migrated"] is False
ent_reg = er.async_get(hass)
# no real migration should have been done
assert ent_reg.async_is_registered("switch.smart_switch_6")
assert ent_reg.async_is_registered("sensor.multisensor_6_battery_level")
assert ent_reg.async_is_registered("sensor.smart_switch_6_electric_consumed_w")
assert ent_reg.async_is_registered(ZWAVE_SOURCE_NODE_ENTITY)
source_entry = ent_reg.async_get(ZWAVE_SOURCE_NODE_ENTITY)
assert source_entry
assert source_entry.unique_id == ZWAVE_SOURCE_NODE_UNIQUE_ID
assert ent_reg.async_is_registered(ZWAVE_BATTERY_ENTITY)
battery_entry = ent_reg.async_get(ZWAVE_BATTERY_ENTITY)
assert battery_entry
assert battery_entry.unique_id == ZWAVE_BATTERY_UNIQUE_ID
assert ent_reg.async_is_registered(ZWAVE_POWER_ENTITY)
power_entry = ent_reg.async_get(ZWAVE_POWER_ENTITY)
assert power_entry
assert power_entry.unique_id == ZWAVE_POWER_UNIQUE_ID
# check that the zwave config entry has not been removed
assert hass.config_entries.async_entries("zwave")
# Check that the zwave integration can be setup after dry run
zwave_config_entry = zwave_integration
with patch("openzwave.option.ZWaveOption"), patch("openzwave.network.ZWaveNetwork"):
assert await hass.config_entries.async_setup(zwave_config_entry.entry_id)
async def test_migrate_zwave_not_setup(
hass, aeon_smart_switch_6, multisensor_6, integration, hass_ws_client
):
"""Test the zwave to zwave_js migration websocket without zwave setup."""
entry = integration
client = await hass_ws_client(hass)
await client.send_json(
{ID: 5, TYPE: "zwave_js/migrate_zwave", ENTRY_ID: entry.entry_id}
)
msg = await client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_not_loaded"
assert msg["error"]["message"] == "Integration zwave is not loaded"
async def test_unique_id_migration_dupes(
hass, multisensor_6_state, client, integration