Migrate old ZHA IasZone sensor state to zigpy cache (#90508)

* Migrate old ZHA IasZone sensor state to zigpy cache

* Use correct type for ZoneStatus

* Test that migration happens

* Test that migration only happens once

* Fix parametrize
This commit is contained in:
TheJulianJES
2023-03-30 20:55:33 -04:00
committed by Paulus Schoutsen
parent 705e68be9e
commit 38aff23be5
2 changed files with 126 additions and 1 deletions
@@ -8,12 +8,15 @@ import zigpy.zcl.clusters.security as security
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import restore_state
from homeassistant.util import dt as dt_util
from .common import (
async_enable_traffic,
async_test_rejoin,
find_entity_id,
send_attributes_report,
update_attribute_cache,
)
from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
@@ -120,3 +123,92 @@ async def test_binary_sensor(
# test rejoin
await async_test_rejoin(hass, zigpy_device, [cluster], reporting)
assert hass.states.get(entity_id).state == STATE_OFF
@pytest.fixture
def core_rs(hass_storage):
"""Core.restore_state fixture."""
def _storage(entity_id, attributes, state):
now = dt_util.utcnow().isoformat()
hass_storage[restore_state.STORAGE_KEY] = {
"version": restore_state.STORAGE_VERSION,
"key": restore_state.STORAGE_KEY,
"data": [
{
"state": {
"entity_id": entity_id,
"state": str(state),
"attributes": attributes,
"last_changed": now,
"last_updated": now,
"context": {
"id": "3c2243ff5f30447eb12e7348cfd5b8ff",
"user_id": None,
},
},
"last_seen": now,
}
],
}
return
return _storage
@pytest.mark.parametrize(
"restored_state",
[
STATE_ON,
STATE_OFF,
],
)
async def test_binary_sensor_migration_not_migrated(
hass: HomeAssistant,
zigpy_device_mock,
core_rs,
zha_device_restored,
restored_state,
) -> None:
"""Test temporary ZHA IasZone binary_sensor migration to zigpy cache."""
entity_id = "binary_sensor.fakemanufacturer_fakemodel_iaszone"
core_rs(entity_id, state=restored_state, attributes={}) # migration sensor state
zigpy_device = zigpy_device_mock(DEVICE_IAS)
zha_device = await zha_device_restored(zigpy_device)
entity_id = await find_entity_id(Platform.BINARY_SENSOR, zha_device, hass)
assert entity_id is not None
assert hass.states.get(entity_id).state == restored_state
# confirm migration extra state attribute was set to True
assert hass.states.get(entity_id).attributes["migrated_to_cache"]
async def test_binary_sensor_migration_already_migrated(
hass: HomeAssistant,
zigpy_device_mock,
core_rs,
zha_device_restored,
) -> None:
"""Test temporary ZHA IasZone binary_sensor migration doesn't migrate multiple times."""
entity_id = "binary_sensor.fakemanufacturer_fakemodel_iaszone"
core_rs(entity_id, state=STATE_OFF, attributes={"migrated_to_cache": True})
zigpy_device = zigpy_device_mock(DEVICE_IAS)
cluster = zigpy_device.endpoints.get(1).ias_zone
cluster.PLUGGED_ATTR_READS = {
"zone_status": security.IasZone.ZoneStatus.Alarm_1,
}
update_attribute_cache(cluster)
zha_device = await zha_device_restored(zigpy_device)
entity_id = await find_entity_id(Platform.BINARY_SENSOR, zha_device, hass)
assert entity_id is not None
assert hass.states.get(entity_id).state == STATE_ON # matches attribute cache
assert hass.states.get(entity_id).attributes["migrated_to_cache"]