Deduplicate code in MQTT alarm_control_panel tests (#55149)
This commit is contained in:
parent
abfba1f455
commit
2c997586eb
1 changed files with 96 additions and 287 deletions
|
@ -10,6 +10,14 @@ from homeassistant.components.mqtt.alarm_control_panel import (
|
|||
MQTT_ALARM_ATTRIBUTES_BLOCKED,
|
||||
)
|
||||
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,
|
||||
SERVICE_ALARM_DISARM,
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_CUSTOM_BYPASS,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
|
@ -153,8 +161,19 @@ async def test_ignore_update_state_if_unknown_via_state_topic(hass, mqtt_mock):
|
|||
assert hass.states.get(entity_id).state == STATE_UNKNOWN
|
||||
|
||||
|
||||
async def test_arm_home_publishes_mqtt(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages while armed."""
|
||||
@pytest.mark.parametrize(
|
||||
"service,payload",
|
||||
[
|
||||
(SERVICE_ALARM_ARM_HOME, "ARM_HOME"),
|
||||
(SERVICE_ALARM_ARM_AWAY, "ARM_AWAY"),
|
||||
(SERVICE_ALARM_ARM_NIGHT, "ARM_NIGHT"),
|
||||
(SERVICE_ALARM_ARM_VACATION, "ARM_VACATION"),
|
||||
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, "ARM_CUSTOM_BYPASS"),
|
||||
(SERVICE_ALARM_DISARM, "DISARM"),
|
||||
],
|
||||
)
|
||||
async def test_publish_mqtt_no_code(hass, mqtt_mock, service, payload):
|
||||
"""Test publishing of MQTT messages when no code is configured."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
|
@ -162,194 +181,85 @@ async def test_arm_home_publishes_mqtt(hass, mqtt_mock):
|
|||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_home(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_HOME", 0, False
|
||||
await hass.services.async_call(
|
||||
alarm_control_panel.DOMAIN,
|
||||
service,
|
||||
{ATTR_ENTITY_ID: "alarm_control_panel.test"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
mqtt_mock.async_publish.assert_called_once_with("alarm/command", payload, 0, False)
|
||||
|
||||
async def test_arm_home_not_publishes_mqtt_with_invalid_code_when_req(hass, mqtt_mock):
|
||||
"""Test not publishing of MQTT messages with invalid.
|
||||
|
||||
When code_arm_required = True
|
||||
"""
|
||||
@pytest.mark.parametrize(
|
||||
"service,payload",
|
||||
[
|
||||
(SERVICE_ALARM_ARM_HOME, "ARM_HOME"),
|
||||
(SERVICE_ALARM_ARM_AWAY, "ARM_AWAY"),
|
||||
(SERVICE_ALARM_ARM_NIGHT, "ARM_NIGHT"),
|
||||
(SERVICE_ALARM_ARM_VACATION, "ARM_VACATION"),
|
||||
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, "ARM_CUSTOM_BYPASS"),
|
||||
(SERVICE_ALARM_DISARM, "DISARM"),
|
||||
],
|
||||
)
|
||||
async def test_publish_mqtt_with_code(hass, mqtt_mock, service, payload):
|
||||
"""Test publishing of MQTT messages when code is configured."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG_CODE,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
call_count = mqtt_mock.async_publish.call_count
|
||||
await common.async_alarm_arm_home(hass, "abcd")
|
||||
|
||||
# No code provided, should not publish
|
||||
await hass.services.async_call(
|
||||
alarm_control_panel.DOMAIN,
|
||||
service,
|
||||
{ATTR_ENTITY_ID: "alarm_control_panel.test"},
|
||||
blocking=True,
|
||||
)
|
||||
assert mqtt_mock.async_publish.call_count == call_count
|
||||
|
||||
|
||||
async def test_arm_home_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages.
|
||||
|
||||
When code_arm_required = False
|
||||
"""
|
||||
config = copy.deepcopy(DEFAULT_CONFIG_CODE)
|
||||
config[alarm_control_panel.DOMAIN]["code_arm_required"] = False
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
# Wrong code provided, should not publish
|
||||
await hass.services.async_call(
|
||||
alarm_control_panel.DOMAIN,
|
||||
config,
|
||||
service,
|
||||
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: "abcd"},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_home(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_HOME", 0, False
|
||||
)
|
||||
|
||||
|
||||
async def test_arm_away_publishes_mqtt(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages while armed."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_away(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_AWAY", 0, False
|
||||
)
|
||||
|
||||
|
||||
async def test_arm_away_not_publishes_mqtt_with_invalid_code_when_req(hass, mqtt_mock):
|
||||
"""Test not publishing of MQTT messages with invalid code.
|
||||
|
||||
When code_arm_required = True
|
||||
"""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG_CODE,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
call_count = mqtt_mock.async_publish.call_count
|
||||
await common.async_alarm_arm_away(hass, "abcd")
|
||||
assert mqtt_mock.async_publish.call_count == call_count
|
||||
|
||||
|
||||
async def test_arm_away_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages.
|
||||
|
||||
When code_arm_required = False
|
||||
"""
|
||||
config = copy.deepcopy(DEFAULT_CONFIG_CODE)
|
||||
config[alarm_control_panel.DOMAIN]["code_arm_required"] = False
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
# Correct code provided, should publish
|
||||
await hass.services.async_call(
|
||||
alarm_control_panel.DOMAIN,
|
||||
config,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_away(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_AWAY", 0, False
|
||||
service,
|
||||
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: "0123"},
|
||||
blocking=True,
|
||||
)
|
||||
mqtt_mock.async_publish.assert_called_once_with("alarm/command", payload, 0, False)
|
||||
|
||||
|
||||
async def test_arm_night_publishes_mqtt(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages while armed."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_night(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_NIGHT", 0, False
|
||||
)
|
||||
|
||||
|
||||
async def test_arm_night_not_publishes_mqtt_with_invalid_code_when_req(hass, mqtt_mock):
|
||||
"""Test not publishing of MQTT messages with invalid code.
|
||||
|
||||
When code_arm_required = True
|
||||
"""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG_CODE,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
call_count = mqtt_mock.async_publish.call_count
|
||||
await common.async_alarm_arm_night(hass, "abcd")
|
||||
assert mqtt_mock.async_publish.call_count == call_count
|
||||
|
||||
|
||||
async def test_arm_night_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages.
|
||||
|
||||
When code_arm_required = False
|
||||
"""
|
||||
config = copy.deepcopy(DEFAULT_CONFIG_CODE)
|
||||
config[alarm_control_panel.DOMAIN]["code_arm_required"] = False
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
config,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_night(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_NIGHT", 0, False
|
||||
)
|
||||
|
||||
|
||||
async def test_arm_vacation_publishes_mqtt(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages while armed."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_vacation(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_VACATION", 0, False
|
||||
)
|
||||
|
||||
|
||||
async def test_arm_vacation_not_publishes_mqtt_with_invalid_code_when_req(
|
||||
hass, mqtt_mock
|
||||
@pytest.mark.parametrize(
|
||||
"service,payload,disable_code",
|
||||
[
|
||||
(SERVICE_ALARM_ARM_HOME, "ARM_HOME", "code_arm_required"),
|
||||
(SERVICE_ALARM_ARM_AWAY, "ARM_AWAY", "code_arm_required"),
|
||||
(SERVICE_ALARM_ARM_NIGHT, "ARM_NIGHT", "code_arm_required"),
|
||||
(SERVICE_ALARM_ARM_VACATION, "ARM_VACATION", "code_arm_required"),
|
||||
(SERVICE_ALARM_ARM_CUSTOM_BYPASS, "ARM_CUSTOM_BYPASS", "code_arm_required"),
|
||||
(SERVICE_ALARM_DISARM, "DISARM", "code_disarm_required"),
|
||||
],
|
||||
)
|
||||
async def test_publish_mqtt_with_code_required_false(
|
||||
hass, mqtt_mock, service, payload, disable_code
|
||||
):
|
||||
"""Test not publishing of MQTT messages with invalid code.
|
||||
"""Test publishing of MQTT messages when code is configured.
|
||||
|
||||
When code_arm_required = True
|
||||
"""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG_CODE,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
call_count = mqtt_mock.async_publish.call_count
|
||||
await common.async_alarm_arm_vacation(hass, "abcd")
|
||||
assert mqtt_mock.async_publish.call_count == call_count
|
||||
|
||||
|
||||
async def test_arm_vacation_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages.
|
||||
|
||||
When code_arm_required = False
|
||||
code_arm_required = False / code_disarm_required = false
|
||||
"""
|
||||
config = copy.deepcopy(DEFAULT_CONFIG_CODE)
|
||||
config[alarm_control_panel.DOMAIN]["code_arm_required"] = False
|
||||
config[alarm_control_panel.DOMAIN][disable_code] = False
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
|
@ -357,100 +267,35 @@ async def test_arm_vacation_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
|||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_vacation(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_VACATION", 0, False
|
||||
)
|
||||
|
||||
|
||||
async def test_arm_custom_bypass_publishes_mqtt(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages while armed."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
# No code provided, should publish
|
||||
await hass.services.async_call(
|
||||
alarm_control_panel.DOMAIN,
|
||||
{
|
||||
alarm_control_panel.DOMAIN: {
|
||||
"platform": "mqtt",
|
||||
"name": "test",
|
||||
"state_topic": "alarm/state",
|
||||
"command_topic": "alarm/command",
|
||||
}
|
||||
},
|
||||
service,
|
||||
{ATTR_ENTITY_ID: "alarm_control_panel.test"},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
mqtt_mock.async_publish.assert_called_once_with("alarm/command", payload, 0, False)
|
||||
mqtt_mock.reset_mock()
|
||||
|
||||
await common.async_alarm_arm_custom_bypass(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_CUSTOM_BYPASS", 0, False
|
||||
)
|
||||
|
||||
|
||||
async def test_arm_custom_bypass_not_publishes_mqtt_with_invalid_code_when_req(
|
||||
hass, mqtt_mock
|
||||
):
|
||||
"""Test not publishing of MQTT messages with invalid code.
|
||||
|
||||
When code_arm_required = True
|
||||
"""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
# Wrong code provided, should publish
|
||||
await hass.services.async_call(
|
||||
alarm_control_panel.DOMAIN,
|
||||
{
|
||||
alarm_control_panel.DOMAIN: {
|
||||
"platform": "mqtt",
|
||||
"name": "test",
|
||||
"state_topic": "alarm/state",
|
||||
"command_topic": "alarm/command",
|
||||
"code": "1234",
|
||||
"code_arm_required": True,
|
||||
}
|
||||
},
|
||||
service,
|
||||
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: "abcd"},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
mqtt_mock.async_publish.assert_called_once_with("alarm/command", payload, 0, False)
|
||||
mqtt_mock.reset_mock()
|
||||
|
||||
call_count = mqtt_mock.async_publish.call_count
|
||||
await common.async_alarm_arm_custom_bypass(hass, "abcd")
|
||||
assert mqtt_mock.async_publish.call_count == call_count
|
||||
|
||||
|
||||
async def test_arm_custom_bypass_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages.
|
||||
|
||||
When code_arm_required = False
|
||||
"""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
# Correct code provided, should publish
|
||||
await hass.services.async_call(
|
||||
alarm_control_panel.DOMAIN,
|
||||
{
|
||||
alarm_control_panel.DOMAIN: {
|
||||
"platform": "mqtt",
|
||||
"name": "test",
|
||||
"state_topic": "alarm/state",
|
||||
"command_topic": "alarm/command",
|
||||
"code": "1234",
|
||||
"code_arm_required": False,
|
||||
}
|
||||
},
|
||||
service,
|
||||
{ATTR_ENTITY_ID: "alarm_control_panel.test", ATTR_CODE: "0123"},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_arm_custom_bypass(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with(
|
||||
"alarm/command", "ARM_CUSTOM_BYPASS", 0, False
|
||||
)
|
||||
|
||||
|
||||
async def test_disarm_publishes_mqtt(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages while disarmed."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_disarm(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with("alarm/command", "DISARM", 0, False)
|
||||
mqtt_mock.async_publish.assert_called_once_with("alarm/command", payload, 0, False)
|
||||
mqtt_mock.reset_mock()
|
||||
|
||||
|
||||
async def test_disarm_publishes_mqtt_with_template(hass, mqtt_mock):
|
||||
|
@ -476,42 +321,6 @@ async def test_disarm_publishes_mqtt_with_template(hass, mqtt_mock):
|
|||
)
|
||||
|
||||
|
||||
async def test_disarm_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||
"""Test publishing of MQTT messages while disarmed.
|
||||
|
||||
When code_disarm_required = False
|
||||
"""
|
||||
config = copy.deepcopy(DEFAULT_CONFIG_CODE)
|
||||
config[alarm_control_panel.DOMAIN]["code"] = "1234"
|
||||
config[alarm_control_panel.DOMAIN]["code_disarm_required"] = False
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
config,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await common.async_alarm_disarm(hass)
|
||||
mqtt_mock.async_publish.assert_called_once_with("alarm/command", "DISARM", 0, False)
|
||||
|
||||
|
||||
async def test_disarm_not_publishes_mqtt_with_invalid_code_when_req(hass, mqtt_mock):
|
||||
"""Test not publishing of MQTT messages with invalid code.
|
||||
|
||||
When code_disarm_required = True
|
||||
"""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
alarm_control_panel.DOMAIN,
|
||||
DEFAULT_CONFIG_CODE,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
call_count = mqtt_mock.async_publish.call_count
|
||||
await common.async_alarm_disarm(hass, "abcd")
|
||||
assert mqtt_mock.async_publish.call_count == call_count
|
||||
|
||||
|
||||
async def test_update_state_via_state_topic_template(hass, mqtt_mock):
|
||||
"""Test updating with template_value via state topic."""
|
||||
assert await async_setup_component(
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue