Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make all ARMED states available for manual_mqtt (#84264) 

* manual_mqtt: parametrize test_no_pending

* manual_mqtt: parametrize test_no_pending_when_code_not_req

* manual_mqtt: parametrize test_with_pending

* manual_mqtt: parametrize test_with_invalid_code

* manual_mqtt: parametrize test_with_template_code

* manual_mqtt: parametrize test_with_specific_pending

* manual_mqtt: parametrize test_arm_via_command_topic

* manual_mqtt: remove unnecessary async_block_till_done from tests

* manual_mqtt: bring over a new test from manual

* manual_mqtt: add more states

The manual alarm control panel supports ARMED_CUSTOM_BYPASS and ARMED_VACATION.
Bring them over to the MQTT version.
pull/84500/head
Paolo Bonzini 2022-12-23 08:20:24 +01:00 committed by GitHub
parent 9830cbfd5d
commit 6cfd991e91
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 256 additions and 486 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
homeassistant/components/manual_mqtt/alarm_control_panel.py
View File

@ -21,8 +21,10 @@ from homeassistant.const import (
CONF_PLATFORM,
CONF_TRIGGER_TIME,
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_CUSTOM_BYPASS,
STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_ARMED_VACATION,
STATE_ALARM_DISARMED,
STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED,
@ -46,6 +48,8 @@ CONF_PAYLOAD_DISARM = "payload_disarm"
CONF_PAYLOAD_ARM_HOME = "payload_arm_home"
CONF_PAYLOAD_ARM_AWAY = "payload_arm_away"
CONF_PAYLOAD_ARM_NIGHT = "payload_arm_night"
CONF_PAYLOAD_ARM_VACATION = "payload_arm_vacation"
CONF_PAYLOAD_ARM_CUSTOM_BYPASS = "payload_arm_custom_bypass"
DEFAULT_ALARM_NAME = "HA Alarm"
DEFAULT_DELAY_TIME = datetime.timedelta(seconds=0)
@ -55,6 +59,8 @@ DEFAULT_DISARM_AFTER_TRIGGER = False
DEFAULT_ARM_AWAY = "ARM_AWAY"
DEFAULT_ARM_HOME = "ARM_HOME"
DEFAULT_ARM_NIGHT = "ARM_NIGHT"
DEFAULT_ARM_VACATION = "ARM_VACATION"
DEFAULT_ARM_CUSTOM_BYPASS = "ARM_CUSTOM_BYPASS"
DEFAULT_DISARM = "DISARM"
SUPPORTED_STATES = [
@ -62,6 +68,8 @@ SUPPORTED_STATES = [
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_ARMED_VACATION,
STATE_ALARM_ARMED_CUSTOM_BYPASS,
STATE_ALARM_TRIGGERED,
]
@ -138,6 +146,12 @@ PLATFORM_SCHEMA = vol.Schema(
vol.Optional(STATE_ALARM_ARMED_NIGHT, default={}): _state_schema(
STATE_ALARM_ARMED_NIGHT
),
vol.Optional(STATE_ALARM_ARMED_VACATION, default={}): _state_schema(
STATE_ALARM_ARMED_VACATION
),
vol.Optional(
STATE_ALARM_ARMED_CUSTOM_BYPASS, default={}
): _state_schema(STATE_ALARM_ARMED_CUSTOM_BYPASS),
vol.Optional(STATE_ALARM_DISARMED, default={}): _state_schema(
STATE_ALARM_DISARMED
),
@ -156,6 +170,12 @@ PLATFORM_SCHEMA = vol.Schema(
vol.Optional(
CONF_PAYLOAD_ARM_NIGHT, default=DEFAULT_ARM_NIGHT
): cv.string,
vol.Optional(
CONF_PAYLOAD_ARM_VACATION, default=DEFAULT_ARM_VACATION
): cv.string,
vol.Optional(
CONF_PAYLOAD_ARM_CUSTOM_BYPASS, default=DEFAULT_ARM_CUSTOM_BYPASS
): cv.string,
vol.Optional(CONF_PAYLOAD_DISARM, default=DEFAULT_DISARM): cv.string,
}
),
@ -187,6 +207,8 @@ def setup_platform(
config.get(CONF_PAYLOAD_ARM_HOME),
config.get(CONF_PAYLOAD_ARM_AWAY),
config.get(CONF_PAYLOAD_ARM_NIGHT),
config.get(CONF_PAYLOAD_ARM_VACATION),
config.get(CONF_PAYLOAD_ARM_CUSTOM_BYPASS),
config,
)
]
@ -210,7 +232,9 @@ class ManualMQTTAlarm(alarm.AlarmControlPanelEntity):
AlarmControlPanelEntityFeature.ARM_HOME
| AlarmControlPanelEntityFeature.ARM_AWAY
| AlarmControlPanelEntityFeature.ARM_NIGHT
| AlarmControlPanelEntityFeature.ARM_VACATION
| AlarmControlPanelEntityFeature.TRIGGER
| AlarmControlPanelEntityFeature.ARM_CUSTOM_BYPASS
)
def __init__(
@ -228,6 +252,8 @@ class ManualMQTTAlarm(alarm.AlarmControlPanelEntity):
payload_arm_home,
payload_arm_away,
payload_arm_night,
payload_arm_vacation,
payload_arm_custom_bypass,
config,
):
"""Init the manual MQTT alarm panel."""
@ -264,6 +290,8 @@ class ManualMQTTAlarm(alarm.AlarmControlPanelEntity):
self._payload_arm_home = payload_arm_home
self._payload_arm_away = payload_arm_away
self._payload_arm_night = payload_arm_night
self._payload_arm_vacation = payload_arm_vacation
self._payload_arm_custom_bypass = payload_arm_custom_bypass
@property
def state(self) -> str:
@ -350,6 +378,24 @@ class ManualMQTTAlarm(alarm.AlarmControlPanelEntity):
self._async_update_state(STATE_ALARM_ARMED_NIGHT)
async def async_alarm_arm_vacation(self, code: str | None = None) -> None:
"""Send arm vacation command."""
if self.code_arm_required and not self._async_validate_code(
code, STATE_ALARM_ARMED_VACATION
):
return
self._async_update_state(STATE_ALARM_ARMED_VACATION)
async def async_alarm_arm_custom_bypass(self, code: str | None = None) -> None:
"""Send arm custom bypass command."""
if self.code_arm_required and not self._async_validate_code(
code, STATE_ALARM_ARMED_CUSTOM_BYPASS
):
return
self._async_update_state(STATE_ALARM_ARMED_CUSTOM_BYPASS)
async def async_alarm_trigger(self, code: str | None = None) -> None:
"""
Send alarm trigger command.
@ -434,6 +480,10 @@ class ManualMQTTAlarm(alarm.AlarmControlPanelEntity):
await self.async_alarm_arm_away(self._code)
elif msg.payload == self._payload_arm_night:
await self.async_alarm_arm_night(self._code)
elif msg.payload == self._payload_arm_vacation:
await self.async_alarm_arm_vacation(self._code)
elif msg.payload == self._payload_arm_custom_bypass:
await self.async_alarm_arm_custom_bypass(self._code)
else:
_LOGGER.warning("Received unexpected payload: %s", msg.payload)
return

692
tests/components/manual_mqtt/test_alarm_control_panel.py
View File

@ -3,12 +3,22 @@ from datetime import timedelta
from unittest.mock import patch
from freezegun import freeze_time
import pytest
from homeassistant.components import alarm_control_panel
from homeassistant.const import (
ATTR_CODE,
ATTR_ENTITY_ID,
SERVICE_ALARM_ARM_AWAY,
SERVICE_ALARM_ARM_CUSTOM_BYPASS,
SERVICE_ALARM_ARM_HOME,
SERVICE_ALARM_ARM_NIGHT,
SERVICE_ALARM_ARM_VACATION,
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_CUSTOM_BYPASS,
STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_ARMED_VACATION,
STATE_ALARM_DISARMED,
STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED,
@ -57,8 +67,20 @@ async def test_fail_setup_without_command_topic(hass, mqtt_mock_entry_with_yaml_
)
async def test_arm_home_no_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm home method."""
@pytest.mark.parametrize(
"service,expected_state",
[
(SERVICE_ALARM_ARM_AWAY, STATE_ALARM_ARMED_AWAY),
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, STATE_ALARM_ARMED_CUSTOM_BYPASS),
(SERVICE_ALARM_ARM_HOME, STATE_ALARM_ARMED_HOME),
(SERVICE_ALARM_ARM_NIGHT, STATE_ALARM_ARMED_NIGHT),
(SERVICE_ALARM_ARM_VACATION, STATE_ALARM_ARMED_VACATION),
],
)
async def test_no_pending(
hass, service, expected_state, mqtt_mock_entry_with_yaml_config
):
"""Test arm method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
@ -80,16 +102,30 @@ async def test_arm_home_no_pending(hass, mqtt_mock_entry_with_yaml_config):
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_home(hass, CODE)
await hass.async_block_till_done()
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: CODE},
blocking=True,
)
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME
assert hass.states.get(entity_id).state == expected_state
async def test_arm_home_no_pending_when_code_not_req(
hass, mqtt_mock_entry_with_yaml_config
@pytest.mark.parametrize(
"service,expected_state",
[
(SERVICE_ALARM_ARM_AWAY, STATE_ALARM_ARMED_AWAY),
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, STATE_ALARM_ARMED_CUSTOM_BYPASS),
(SERVICE_ALARM_ARM_HOME, STATE_ALARM_ARMED_HOME),
(SERVICE_ALARM_ARM_NIGHT, STATE_ALARM_ARMED_NIGHT),
(SERVICE_ALARM_ARM_VACATION, STATE_ALARM_ARMED_VACATION),
],
)
async def test_no_pending_when_code_not_req(
hass, service, expected_state, mqtt_mock_entry_with_yaml_config
):
"""Test arm home method."""
"""Test arm method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
@ -112,14 +148,30 @@ async def test_arm_home_no_pending_when_code_not_req(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_home(hass, 0)
await hass.async_block_till_done()
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: CODE},
blocking=True,
)
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME
assert hass.states.get(entity_id).state == expected_state
async def test_arm_home_with_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm home method."""
@pytest.mark.parametrize(
"service,expected_state",
[
(SERVICE_ALARM_ARM_AWAY, STATE_ALARM_ARMED_AWAY),
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, STATE_ALARM_ARMED_CUSTOM_BYPASS),
(SERVICE_ALARM_ARM_HOME, STATE_ALARM_ARMED_HOME),
(SERVICE_ALARM_ARM_NIGHT, STATE_ALARM_ARMED_NIGHT),
(SERVICE_ALARM_ARM_VACATION, STATE_ALARM_ARMED_VACATION),
],
)
async def test_with_pending(
hass, service, expected_state, mqtt_mock_entry_with_yaml_config
):
"""Test arm method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
@ -141,13 +193,17 @@ async def test_arm_home_with_pending(hass, mqtt_mock_entry_with_yaml_config):
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_home(hass, CODE, entity_id)
await hass.async_block_till_done()
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: CODE},
blocking=True,
)
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
state = hass.states.get(entity_id)
assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_HOME
assert state.attributes["post_pending_state"] == expected_state
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(
@ -157,11 +213,34 @@ async def test_arm_home_with_pending(hass, mqtt_mock_entry_with_yaml_config):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME
state = hass.states.get(entity_id)
assert state.state == expected_state
# Do not go to the pending state when updating to the same state
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: CODE},
blocking=True,
)
assert hass.states.get(entity_id).state == expected_state
async def test_arm_home_with_invalid_code(hass, mqtt_mock_entry_with_yaml_config):
"""Attempt to arm home without a valid code."""
@pytest.mark.parametrize(
"service,expected_state",
[
(SERVICE_ALARM_ARM_AWAY, STATE_ALARM_ARMED_AWAY),
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, STATE_ALARM_ARMED_CUSTOM_BYPASS),
(SERVICE_ALARM_ARM_HOME, STATE_ALARM_ARMED_HOME),
(SERVICE_ALARM_ARM_NIGHT, STATE_ALARM_ARMED_NIGHT),
(SERVICE_ALARM_ARM_VACATION, STATE_ALARM_ARMED_VACATION),
],
)
async def test_with_invalid_code(
hass, service, expected_state, mqtt_mock_entry_with_yaml_config
):
"""Attempt to arm without a valid code."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
@ -183,74 +262,29 @@ async def test_arm_home_with_invalid_code(hass, mqtt_mock_entry_with_yaml_config
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_home(hass, f"{CODE}2")
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
async def test_arm_away_no_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm home method."""
assert await async_setup_component(
hass,
await hass.services.async_call(
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"code": CODE,
"pending_time": 0,
"disarm_after_trigger": False,
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: f"{CODE}2"},
blocking=True,
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE, entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
async def test_arm_away_no_pending_when_code_not_req(
hass, mqtt_mock_entry_with_yaml_config
@pytest.mark.parametrize(
"service,expected_state",
[
(SERVICE_ALARM_ARM_AWAY, STATE_ALARM_ARMED_AWAY),
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, STATE_ALARM_ARMED_CUSTOM_BYPASS),
(SERVICE_ALARM_ARM_HOME, STATE_ALARM_ARMED_HOME),
(SERVICE_ALARM_ARM_NIGHT, STATE_ALARM_ARMED_NIGHT),
(SERVICE_ALARM_ARM_VACATION, STATE_ALARM_ARMED_VACATION),
],
)
async def test_with_template_code(
hass, service, expected_state, mqtt_mock_entry_with_yaml_config
):
"""Test arm home method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"code_arm_required": False,
"code": CODE,
"pending_time": 0,
"disarm_after_trigger": False,
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, 0, entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
async def test_arm_home_with_template_code(hass, mqtt_mock_entry_with_yaml_config):
"""Attempt to arm with a template-based code."""
assert await async_setup_component(
hass,
@ -273,117 +307,31 @@ async def test_arm_home_with_template_code(hass, mqtt_mock_entry_with_yaml_confi
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_home(hass, "abc")
await hass.async_block_till_done()
await hass.services.async_call(
alarm_control_panel.DOMAIN,
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: "abc"},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_ARMED_HOME
assert state.state == expected_state
async def test_arm_away_with_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm home method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"code": CODE,
"pending_time": 1,
"disarm_after_trigger": False,
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
state = hass.states.get(entity_id)
assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_AWAY
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
async def test_arm_away_with_invalid_code(hass, mqtt_mock_entry_with_yaml_config):
"""Attempt to arm away without a valid code."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"code": CODE,
"pending_time": 1,
"disarm_after_trigger": False,
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, f"{CODE}2")
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
async def test_arm_night_no_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm night method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"code": CODE,
"pending_time": 0,
"disarm_after_trigger": False,
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_night(hass, CODE, entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
async def test_arm_night_no_pending_when_code_not_req(
hass, mqtt_mock_entry_with_yaml_config
@pytest.mark.parametrize(
"service,expected_state",
[
(SERVICE_ALARM_ARM_AWAY, STATE_ALARM_ARMED_AWAY),
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, STATE_ALARM_ARMED_CUSTOM_BYPASS),
(SERVICE_ALARM_ARM_HOME, STATE_ALARM_ARMED_HOME),
(SERVICE_ALARM_ARM_NIGHT, STATE_ALARM_ARMED_NIGHT),
(SERVICE_ALARM_ARM_VACATION, STATE_ALARM_ARMED_VACATION),
],
)
async def test_with_specific_pending(
hass, service, expected_state, mqtt_mock_entry_with_yaml_config
):
"""Test arm night method."""
"""Test arm method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
@ -391,10 +339,8 @@ async def test_arm_night_no_pending_when_code_not_req(
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"code_arm_required": False,
"code": CODE,
"pending_time": 0,
"disarm_after_trigger": False,
"pending_time": 10,
expected_state: {"pending_time": 2},
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
@ -404,46 +350,16 @@ async def test_arm_night_no_pending_when_code_not_req(
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_night(hass, 0, entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
async def test_arm_night_with_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm night method."""
assert await async_setup_component(
hass,
await hass.services.async_call(
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"code": CODE,
"pending_time": 1,
"disarm_after_trigger": False,
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
service,
{ATTR_ENTITY_ID: "alarm_control_panel.test"},
blocking=True,
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_night(hass, CODE)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
state = hass.states.get(entity_id)
assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_NIGHT
future = dt_util.utcnow() + timedelta(seconds=1)
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
@ -451,42 +367,7 @@ async def test_arm_night_with_pending(hass, mqtt_mock_entry_with_yaml_config):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
# Do not go to the pending state when updating to the same state
await common.async_alarm_arm_night(hass, CODE, entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
async def test_arm_night_with_invalid_code(hass, mqtt_mock_entry_with_yaml_config):
"""Attempt to arm night without a valid code."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"code": CODE,
"pending_time": 1,
"disarm_after_trigger": False,
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_night(hass, f"{CODE}2")
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
assert hass.states.get(entity_id).state == expected_state
async def test_trigger_no_pending(hass, mqtt_mock_entry_with_yaml_config):
@ -552,12 +433,10 @@ async def test_trigger_with_delay(hass, mqtt_mock_entry_with_yaml_config):
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_PENDING
@ -599,7 +478,6 @@ async def test_trigger_zero_trigger_time(hass, mqtt_mock_entry_with_yaml_config)
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_trigger(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
@ -630,7 +508,6 @@ async def test_trigger_zero_trigger_time_with_pending(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_trigger(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
@ -659,7 +536,6 @@ async def test_trigger_with_pending(hass, mqtt_mock_entry_with_yaml_config):
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_trigger(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
@ -713,7 +589,6 @@ async def test_trigger_with_disarm_after_trigger(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
@ -755,7 +630,6 @@ async def test_trigger_with_zero_specific_trigger_time(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
@ -787,7 +661,6 @@ async def test_trigger_with_unused_zero_specific_trigger_time(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
@ -828,7 +701,6 @@ async def test_trigger_with_specific_trigger_time(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
@ -869,12 +741,10 @@ async def test_back_to_back_trigger_with_no_disarm_after_trigger(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE, entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
@ -889,7 +759,6 @@ async def test_back_to_back_trigger_with_no_disarm_after_trigger(
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
@ -927,12 +796,10 @@ async def test_disarm_while_pending_trigger(hass, mqtt_mock_entry_with_yaml_conf
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_trigger(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
await common.async_alarm_disarm(hass, entity_id=entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
@ -977,12 +844,10 @@ async def test_disarm_during_trigger_with_invalid_code(
)
await common.async_alarm_trigger(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
await common.async_alarm_disarm(hass, entity_id=entity_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
@ -1025,12 +890,10 @@ async def test_trigger_with_unused_specific_delay(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_PENDING
@ -1074,12 +937,10 @@ async def test_trigger_with_specific_delay(hass, mqtt_mock_entry_with_yaml_confi
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_PENDING
@ -1123,12 +984,10 @@ async def test_trigger_with_pending_and_delay(hass, mqtt_mock_entry_with_yaml_co
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_PENDING
@ -1187,12 +1046,10 @@ async def test_trigger_with_pending_and_specific_delay(
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_PENDING
@ -1222,116 +1079,6 @@ async def test_trigger_with_pending_and_specific_delay(
assert state.state == STATE_ALARM_TRIGGERED
async def test_armed_home_with_specific_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm home method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"pending_time": 10,
"armed_home": {"pending_time": 2},
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
await common.async_alarm_arm_home(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME
async def test_armed_away_with_specific_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm home method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"pending_time": 10,
"armed_away": {"pending_time": 2},
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
await common.async_alarm_arm_away(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
async def test_armed_night_with_specific_pending(
hass, mqtt_mock_entry_with_yaml_config
):
"""Test arm home method."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"pending_time": 10,
"armed_night": {"pending_time": 2},
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
await common.async_alarm_arm_night(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
future = dt_util.utcnow() + timedelta(seconds=2)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
async def test_trigger_with_specific_pending(hass, mqtt_mock_entry_with_yaml_config):
"""Test arm home method."""
assert await async_setup_component(
@ -1355,7 +1102,6 @@ async def test_trigger_with_specific_pending(hass, mqtt_mock_entry_with_yaml_con
entity_id = "alarm_control_panel.test"
await common.async_alarm_trigger(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
@ -1380,6 +1126,51 @@ async def test_trigger_with_specific_pending(hass, mqtt_mock_entry_with_yaml_con
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
async def test_trigger_with_no_disarm_after_trigger(
hass, mqtt_mock_entry_with_yaml_config
):
"""Test disarm after trigger."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
"alarm_control_panel": {
"platform": "manual_mqtt",
"name": "test",
"trigger_time": 5,
"pending_time": 0,
"delay_time": 0,
"disarm_after_trigger": False,
"command_topic": "alarm/command",
"state_topic": "alarm/state",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE, entity_id)
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
assert hass.states.get(entity_id).state == STATE_ALARM_TRIGGERED
future = dt_util.utcnow() + timedelta(seconds=5)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
async def test_arm_away_after_disabled_disarmed(hass, mqtt_mock_entry_with_yaml_config):
"""Test pending state with and without zero trigger time."""
assert await async_setup_component(
@ -1407,7 +1198,6 @@ async def test_arm_away_after_disabled_disarmed(hass, mqtt_mock_entry_with_yaml_
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_away(hass, CODE)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_PENDING
@ -1415,7 +1205,6 @@ async def test_arm_away_after_disabled_disarmed(hass, mqtt_mock_entry_with_yaml_
assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_PENDING
@ -1431,7 +1220,6 @@ async def test_arm_away_after_disabled_disarmed(hass, mqtt_mock_entry_with_yaml_
assert state.state == STATE_ALARM_ARMED_AWAY
await common.async_alarm_trigger(hass, entity_id=entity_id)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_PENDING
@ -1471,26 +1259,36 @@ async def test_disarm_with_template_code(hass, mqtt_mock_entry_with_yaml_config)
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
await common.async_alarm_arm_home(hass, "def")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_ARMED_HOME
await common.async_alarm_disarm(hass, "def")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_ARMED_HOME
await common.async_alarm_disarm(hass, "abc")
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == STATE_ALARM_DISARMED
async def test_arm_home_via_command_topic(hass, mqtt_mock_entry_with_yaml_config):
"""Test arming home via command topic."""
@pytest.mark.parametrize(
"config,expected_state",
[
("payload_arm_away", STATE_ALARM_ARMED_AWAY),
("payload_arm_custom_bypass", STATE_ALARM_ARMED_CUSTOM_BYPASS),
("payload_arm_home", STATE_ALARM_ARMED_HOME),
("payload_arm_night", STATE_ALARM_ARMED_NIGHT),
("payload_arm_vacation", STATE_ALARM_ARMED_VACATION),
],
)
async def test_arm_via_command_topic(
hass, config, expected_state, mqtt_mock_entry_with_yaml_config
):
"""Test arming via command topic."""
command = config[8:].upper()
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
@ -1501,7 +1299,7 @@ async def test_arm_home_via_command_topic(hass, mqtt_mock_entry_with_yaml_config
"pending_time": 1,
"state_topic": "alarm/state",
"command_topic": "alarm/command",
"payload_arm_home": "ARM_HOME",
config: command,
}
},
)
@ -1511,8 +1309,8 @@ async def test_arm_home_via_command_topic(hass, mqtt_mock_entry_with_yaml_config
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
# Fire the arm command via MQTT; ensure state changes to pending
async_fire_mqtt_message(hass, "alarm/command", "ARM_HOME")
# Fire the arm command via MQTT; ensure state changes to arming
async_fire_mqtt_message(hass, "alarm/command", command)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
@ -1525,85 +1323,7 @@ async def test_arm_home_via_command_topic(hass, mqtt_mock_entry_with_yaml_config
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_HOME
async def test_arm_away_via_command_topic(hass, mqtt_mock_entry_with_yaml_config):
"""Test arming away via command topic."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
alarm_control_panel.DOMAIN: {
"platform": "manual_mqtt",
"name": "test",
"pending_time": 1,
"state_topic": "alarm/state",
"command_topic": "alarm/command",
"payload_arm_away": "ARM_AWAY",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
# Fire the arm command via MQTT; ensure state changes to pending
async_fire_mqtt_message(hass, "alarm/command", "ARM_AWAY")
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
# Fast-forward a little bit
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_AWAY
async def test_arm_night_via_command_topic(hass, mqtt_mock_entry_with_yaml_config):
"""Test arming night via command topic."""
assert await async_setup_component(
hass,
alarm_control_panel.DOMAIN,
{
alarm_control_panel.DOMAIN: {
"platform": "manual_mqtt",
"name": "test",
"pending_time": 1,
"state_topic": "alarm/state",
"command_topic": "alarm/command",
"payload_arm_night": "ARM_NIGHT",
}
},
)
await hass.async_block_till_done()
entity_id = "alarm_control_panel.test"
assert hass.states.get(entity_id).state == STATE_ALARM_DISARMED
# Fire the arm command via MQTT; ensure state changes to pending
async_fire_mqtt_message(hass, "alarm/command", "ARM_NIGHT")
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_PENDING
# Fast-forward a little bit
future = dt_util.utcnow() + timedelta(seconds=1)
with patch(
("homeassistant.components.manual_mqtt.alarm_control_panel.dt_util.utcnow"),
return_value=future,
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ALARM_ARMED_NIGHT
assert hass.states.get(entity_id).state == expected_state
async def test_disarm_pending_via_command_topic(hass, mqtt_mock_entry_with_yaml_config):