jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix manual.alarm_control_panel RestoreEntity bugs (#82990)

Browse Source 
* manual: add previous state for TRIGGERED as well

This will be useful to properly restore timers on restart.

* manual: correctly restore timers on startup
This commit is contained in:
Paolo Bonzini 2022-12-19 11:19:24 +01:00 committed by GitHub
parent 20f0abacc0
commit 52594717ca
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 263 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
homeassistant/components/manual/alarm_control_panel.py
View File

@ -358,7 +358,10 @@ class ManualAlarm(alarm.AlarmControlPanelEntity, RestoreEntity):
self._state = state
self._state_ts = dt_util.utcnow()
self.async_schedule_update_ha_state()
self._async_set_state_update_events()
def _async_set_state_update_events(self) -> None:
state = self._state
if state == STATE_ALARM_TRIGGERED:
pending_time = self._pending_time(state)
async_track_point_in_time(
@ -403,6 +406,10 @@ class ManualAlarm(alarm.AlarmControlPanelEntity, RestoreEntity):
ATTR_PREVIOUS_STATE: self._previous_state,
ATTR_NEXT_STATE: self._state,
}
if self.state == STATE_ALARM_TRIGGERED:
return {
ATTR_PREVIOUS_STATE: self._previous_state,
}
return {}
@callback
@ -414,14 +421,14 @@ class ManualAlarm(alarm.AlarmControlPanelEntity, RestoreEntity):
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if state := await self.async_get_last_state():
if (
state.state in (STATE_ALARM_PENDING, STATE_ALARM_ARMING)
and hasattr(state, "attributes")
and state.attributes[ATTR_PREVIOUS_STATE]
):
# If in arming or pending state, we return to the ATTR_PREVIOUS_STATE
self._state = state.attributes[ATTR_PREVIOUS_STATE]
self._state_ts = dt_util.utcnow()
self._state_ts = state.last_updated
if hasattr(state, "attributes") and ATTR_NEXT_STATE in state.attributes:
# If in arming or pending state we record the transition,
# not the current state
self._state = state.attributes[ATTR_NEXT_STATE]
else:
self._state = state.state
self._state_ts = state.last_updated
if hasattr(state, "attributes") and ATTR_PREVIOUS_STATE in state.attributes:
self._previous_state = state.attributes[ATTR_PREVIOUS_STATE]
self._async_set_state_update_events()

256
tests/components/manual/test_alarm_control_panel.py
View File

@ -354,7 +354,9 @@ async def test_trigger_no_pending(hass):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_DISARMED
assert state.state == STATE_ALARM_TRIGGERED
async def test_trigger_with_delay(hass):
@ -398,6 +400,7 @@ async def test_trigger_with_delay(hass):
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
@ -490,6 +493,7 @@ async def test_trigger_with_pending(hass):
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_DISARMED
assert state.state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
@ -546,6 +550,7 @@ async def test_trigger_with_unused_specific_delay(hass):
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
@ -591,6 +596,7 @@ async def test_trigger_with_specific_delay(hass):
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
@ -647,6 +653,7 @@ async def test_trigger_with_pending_and_delay(hass):
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
@ -704,6 +711,7 @@ async def test_trigger_with_pending_and_specific_delay(hass):
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
@ -739,7 +747,9 @@ async def test_trigger_with_specific_pending(hass):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_DISARMED
assert state.state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(
@ -775,7 +785,9 @@ async def test_trigger_with_disarm_after_trigger(hass):
await common.async_alarm_trigger(hass, entity_id=entity_id)
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_DISARMED
assert state.state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(
@ -839,7 +851,9 @@ async def test_trigger_with_unused_zero_specific_trigger_time(hass):
await common.async_alarm_trigger(hass, entity_id=entity_id)
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_DISARMED
assert state.state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(
@ -875,7 +889,9 @@ async def test_trigger_with_specific_trigger_time(hass):
await common.async_alarm_trigger(hass, entity_id=entity_id)
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_DISARMED
assert state.state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(
@ -916,7 +932,9 @@ async def test_trigger_with_no_disarm_after_trigger(hass):
await common.async_alarm_trigger(hass, entity_id=entity_id)
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(
@ -957,7 +975,9 @@ async def test_back_to_back_trigger_with_no_disarm_after_trigger(hass):
await common.async_alarm_trigger(hass, entity_id=entity_id)
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(
@ -971,7 +991,9 @@ async def test_back_to_back_trigger_with_no_disarm_after_trigger(hass):
await common.async_alarm_trigger(hass, entity_id=entity_id)
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(
@ -1060,7 +1082,9 @@ async def test_disarm_during_trigger_with_invalid_code(hass):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_DISARMED
assert state.state == STATE_ALARM_TRIGGERED
async def test_disarm_with_template_code(hass):
@ -1159,6 +1183,7 @@ async def test_arm_away_after_disabled_disarmed(hass):
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["previous_state"] == STATE_ALARM_ARMED_AWAY
assert state.state == STATE_ALARM_TRIGGERED
@ -1198,3 +1223,216 @@ async def test_restore_state(hass, expected_state):
state = hass.states.get("alarm_control_panel.test")
assert state
assert state.state == expected_state
@pytest.mark.parametrize(
"expected_state",
[
(STATE_ALARM_ARMED_AWAY),
(STATE_ALARM_ARMED_CUSTOM_BYPASS),
(STATE_ALARM_ARMED_HOME),
(STATE_ALARM_ARMED_NIGHT),
(STATE_ALARM_ARMED_VACATION),
],
)
async def test_restore_state_arming(hass, expected_state):
"""Ensure ARMING state is restored on startup."""
time = dt_util.utcnow() - timedelta(seconds=15)
entity_id = "alarm_control_panel.test"
attributes = {
"previous_state": STATE_ALARM_DISARMED,
"next_state": expected_state,
}
mock_restore_cache(
hass, (State(entity_id, expected_state, attributes, last_updated=time),)
)
hass.state = CoreState.starting
mock_component(hass, "recorder")
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual",
"name": "test",
"arming_time": 60,
"trigger_time": 0,
"disarm_after_trigger": False,
}
},
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.attributes["previous_state"] == STATE_ALARM_DISARMED
assert state.attributes["next_state"] == expected_state
assert state.state == STATE_ALARM_ARMING
future = time + timedelta(seconds=61)
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == expected_state
@pytest.mark.parametrize(
"previous_state",
[
(STATE_ALARM_ARMED_AWAY),
(STATE_ALARM_ARMED_CUSTOM_BYPASS),
(STATE_ALARM_ARMED_HOME),
(STATE_ALARM_ARMED_NIGHT),
(STATE_ALARM_ARMED_VACATION),
(STATE_ALARM_DISARMED),
],
)
async def test_restore_state_pending(hass, previous_state):
"""Ensure PENDING state is restored on startup."""
time = dt_util.utcnow() - timedelta(seconds=15)
entity_id = "alarm_control_panel.test"
attributes = {
"previous_state": previous_state,
"next_state": STATE_ALARM_TRIGGERED,
}
mock_restore_cache(
hass,
(State(entity_id, STATE_ALARM_TRIGGERED, attributes, last_updated=time),),
)
hass.state = CoreState.starting
mock_component(hass, "recorder")
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual",
"name": "test",
"arming_time": 0,
"delay_time": 60,
"trigger_time": 60,
"disarm_after_trigger": False,
}
},
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.attributes["previous_state"] == previous_state
assert state.attributes["next_state"] == STATE_ALARM_TRIGGERED
assert state.state == STATE_ALARM_PENDING
future = time + timedelta(seconds=61)
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_TRIGGERED
future = time + timedelta(seconds=121)
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == previous_state
@pytest.mark.parametrize(
"previous_state",
[
(STATE_ALARM_ARMED_AWAY),
(STATE_ALARM_ARMED_CUSTOM_BYPASS),
(STATE_ALARM_ARMED_HOME),
(STATE_ALARM_ARMED_NIGHT),
(STATE_ALARM_ARMED_VACATION),
(STATE_ALARM_DISARMED),
],
)
async def test_restore_state_triggered(hass, previous_state):
"""Ensure PENDING state is resolved to TRIGGERED on startup."""
time = dt_util.utcnow() - timedelta(seconds=75)
entity_id = "alarm_control_panel.test"
attributes = {
"previous_state": previous_state,
}
mock_restore_cache(
hass,
(State(entity_id, STATE_ALARM_TRIGGERED, attributes, last_updated=time),),
)
hass.state = CoreState.starting
mock_component(hass, "recorder")
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual",
"name": "test",
"arming_time": 0,
"delay_time": 60,
"trigger_time": 60,
"disarm_after_trigger": False,
}
},
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.attributes["previous_state"] == previous_state
assert "next_state" not in state.attributes
assert state.state == STATE_ALARM_TRIGGERED
future = time + timedelta(seconds=121)
with freeze_time(future):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == previous_state
async def test_restore_state_triggered_long_ago(hass):
"""Ensure TRIGGERED state is resolved on startup."""
time = dt_util.utcnow() - timedelta(seconds=125)
entity_id = "alarm_control_panel.test"
attributes = {
"previous_state": STATE_ALARM_ARMED_AWAY,
}
mock_restore_cache(
hass,
(State(entity_id, STATE_ALARM_TRIGGERED, attributes, last_updated=time),),
)
hass.state = CoreState.starting
mock_component(hass, "recorder")
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual",
"name": "test",
"arming_time": 0,
"delay_time": 60,
"trigger_time": 60,
"disarm_after_trigger": True,
}
},
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_DISARMED