mirror of
https://github.com/home-assistant/core.git
synced 2025-07-25 06:07:17 +00:00
Drop unnecessary block_till_done, improve tests (#23246)
This commit is contained in:
Erik Montnemery
Jason Hu
parent
c899e2a662
commit
887e1cd8e3
@ -1,6 +1,5 @@
|
|||||||
"""The tests the MQTT alarm control panel component."""
|
"""The tests the MQTT alarm control panel component."""
|
||||||
import json
|
import json
|
||||||
import unittest
|
|
||||||
from unittest.mock import ANY
|
from unittest.mock import ANY
|
||||||
|
|
||||||
from homeassistant.components import alarm_control_panel, mqtt
|
from homeassistant.components import alarm_control_panel, mqtt
|
||||||
@ -9,398 +8,408 @@ from homeassistant.const import (
|
|||||||
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT,
|
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, STATE_ALARM_ARMED_NIGHT,
|
||||||
STATE_ALARM_DISARMED, STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED,
|
STATE_ALARM_DISARMED, STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED,
|
||||||
STATE_UNAVAILABLE, STATE_UNKNOWN)
|
STATE_UNAVAILABLE, STATE_UNKNOWN)
|
||||||
from homeassistant.setup import setup_component
|
|
||||||
|
|
||||||
from tests.common import (
|
from tests.common import (
|
||||||
MockConfigEntry, assert_setup_component, async_fire_mqtt_message,
|
MockConfigEntry, assert_setup_component, async_fire_mqtt_message,
|
||||||
async_mock_mqtt_component, async_setup_component, fire_mqtt_message,
|
async_mock_mqtt_component, async_setup_component, mock_registry)
|
||||||
get_test_home_assistant, mock_mqtt_component, mock_registry)
|
|
||||||
from tests.components.alarm_control_panel import common
|
from tests.components.alarm_control_panel import common
|
||||||
|
|
||||||
CODE = 'HELLO_CODE'
|
CODE = 'HELLO_CODE'
|
||||||
|
|
||||||
|
|
||||||
class TestAlarmControlPanelMQTT(unittest.TestCase):
|
async def test_fail_setup_without_state_topic(hass, mqtt_mock):
|
||||||
"""Test the manual alarm module."""
|
"""Test for failing with no state topic."""
|
||||||
|
with assert_setup_component(0) as config:
|
||||||
# pylint: disable=invalid-name
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
"""Set up things to be run when tests are started."""
|
|
||||||
self.hass = get_test_home_assistant()
|
|
||||||
self.mock_publish = mock_mqtt_component(self.hass)
|
|
||||||
|
|
||||||
def tearDown(self): # pylint: disable=invalid-name
|
|
||||||
"""Stop down stuff we started."""
|
|
||||||
self.hass.stop()
|
|
||||||
|
|
||||||
def test_fail_setup_without_state_topic(self):
|
|
||||||
"""Test for failing with no state topic."""
|
|
||||||
with assert_setup_component(0) as config:
|
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
|
||||||
alarm_control_panel.DOMAIN: {
|
|
||||||
'platform': 'mqtt',
|
|
||||||
'command_topic': 'alarm/command'
|
|
||||||
}
|
|
||||||
})
|
|
||||||
assert not config[alarm_control_panel.DOMAIN]
|
|
||||||
|
|
||||||
def test_fail_setup_without_command_topic(self):
|
|
||||||
"""Test failing with no command topic."""
|
|
||||||
with assert_setup_component(0):
|
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
|
||||||
alarm_control_panel.DOMAIN: {
|
|
||||||
'platform': 'mqtt',
|
|
||||||
'state_topic': 'alarm/state'
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
def test_update_state_via_state_topic(self):
|
|
||||||
"""Test updating with via state topic."""
|
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
|
||||||
alarm_control_panel.DOMAIN: {
|
alarm_control_panel.DOMAIN: {
|
||||||
'platform': 'mqtt',
|
'platform': 'mqtt',
|
||||||
'name': 'test',
|
'command_topic': 'alarm/command'
|
||||||
'state_topic': 'alarm/state',
|
}
|
||||||
'command_topic': 'alarm/command',
|
})
|
||||||
|
assert not config[alarm_control_panel.DOMAIN]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_fail_setup_without_command_topic(hass, mqtt_mock):
|
||||||
|
"""Test failing with no command topic."""
|
||||||
|
with assert_setup_component(0):
|
||||||
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'state_topic': 'alarm/state'
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
entity_id = 'alarm_control_panel.test'
|
|
||||||
|
|
||||||
assert STATE_UNKNOWN == \
|
async def test_update_state_via_state_topic(hass, mqtt_mock):
|
||||||
self.hass.states.get(entity_id).state
|
"""Test updating with via state topic."""
|
||||||
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
for state in (STATE_ALARM_DISARMED, STATE_ALARM_ARMED_HOME,
|
entity_id = 'alarm_control_panel.test'
|
||||||
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_NIGHT,
|
|
||||||
STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED):
|
|
||||||
fire_mqtt_message(self.hass, 'alarm/state', state)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
assert state == self.hass.states.get(entity_id).state
|
|
||||||
|
|
||||||
def test_ignore_update_state_if_unknown_via_state_topic(self):
|
assert STATE_UNKNOWN == \
|
||||||
"""Test ignoring updates via state topic."""
|
hass.states.get(entity_id).state
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
|
||||||
alarm_control_panel.DOMAIN: {
|
|
||||||
'platform': 'mqtt',
|
|
||||||
'name': 'test',
|
|
||||||
'state_topic': 'alarm/state',
|
|
||||||
'command_topic': 'alarm/command',
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
entity_id = 'alarm_control_panel.test'
|
for state in (STATE_ALARM_DISARMED, STATE_ALARM_ARMED_HOME,
|
||||||
|
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_NIGHT,
|
||||||
|
STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED):
|
||||||
|
async_fire_mqtt_message(hass, 'alarm/state', state)
|
||||||
|
assert state == hass.states.get(entity_id).state
|
||||||
|
|
||||||
assert STATE_UNKNOWN == \
|
|
||||||
self.hass.states.get(entity_id).state
|
|
||||||
|
|
||||||
fire_mqtt_message(self.hass, 'alarm/state', 'unsupported state')
|
async def test_ignore_update_state_if_unknown_via_state_topic(hass, mqtt_mock):
|
||||||
self.hass.block_till_done()
|
"""Test ignoring updates via state topic."""
|
||||||
assert STATE_UNKNOWN == self.hass.states.get(entity_id).state
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
def test_arm_home_publishes_mqtt(self):
|
entity_id = 'alarm_control_panel.test'
|
||||||
"""Test publishing of MQTT messages while armed."""
|
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
|
||||||
alarm_control_panel.DOMAIN: {
|
|
||||||
'platform': 'mqtt',
|
|
||||||
'name': 'test',
|
|
||||||
'state_topic': 'alarm/state',
|
|
||||||
'command_topic': 'alarm/command',
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
common.alarm_arm_home(self.hass)
|
assert STATE_UNKNOWN == \
|
||||||
self.hass.block_till_done()
|
hass.states.get(entity_id).state
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
|
||||||
'alarm/command', 'ARM_HOME', 0, False)
|
|
||||||
|
|
||||||
def test_arm_home_not_publishes_mqtt_with_invalid_code_when_req(self):
|
async_fire_mqtt_message(hass, 'alarm/state', 'unsupported state')
|
||||||
"""Test not publishing of MQTT messages with invalid.
|
assert STATE_UNKNOWN == hass.states.get(entity_id).state
|
||||||
|
|
||||||
When code_arm_required = True
|
|
||||||
"""
|
|
||||||
assert setup_component(self.hass, 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
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
call_count = self.mock_publish.call_count
|
async def test_arm_home_publishes_mqtt(hass, mqtt_mock):
|
||||||
common.alarm_arm_home(self.hass, 'abcd')
|
"""Test publishing of MQTT messages while armed."""
|
||||||
self.hass.block_till_done()
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
assert call_count == self.mock_publish.call_count
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
def test_arm_home_publishes_mqtt_when_code_not_req(self):
|
common.async_alarm_arm_home(hass)
|
||||||
"""Test publishing of MQTT messages.
|
await hass.async_block_till_done()
|
||||||
|
mqtt_mock.async_publish.assert_called_once_with(
|
||||||
|
'alarm/command', 'ARM_HOME', 0, False)
|
||||||
|
|
||||||
When code_arm_required = False
|
|
||||||
"""
|
|
||||||
assert setup_component(self.hass, 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
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
common.alarm_arm_home(self.hass)
|
async def test_arm_home_not_publishes_mqtt_with_invalid_code_when_req(
|
||||||
self.hass.block_till_done()
|
hass, mqtt_mock):
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
"""Test not publishing of MQTT messages with invalid.
|
||||||
'alarm/command', 'ARM_HOME', 0, False)
|
|
||||||
|
|
||||||
def test_arm_away_publishes_mqtt(self):
|
When code_arm_required = True
|
||||||
"""Test publishing of MQTT messages while armed."""
|
"""
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
alarm_control_panel.DOMAIN: {
|
alarm_control_panel.DOMAIN: {
|
||||||
'platform': 'mqtt',
|
'platform': 'mqtt',
|
||||||
'name': 'test',
|
'name': 'test',
|
||||||
'state_topic': 'alarm/state',
|
'state_topic': 'alarm/state',
|
||||||
'command_topic': 'alarm/command',
|
'command_topic': 'alarm/command',
|
||||||
}
|
'code': '1234',
|
||||||
})
|
'code_arm_required': True
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
common.alarm_arm_away(self.hass)
|
call_count = mqtt_mock.async_publish.call_count
|
||||||
self.hass.block_till_done()
|
common.async_alarm_arm_home(hass, 'abcd')
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
await hass.async_block_till_done()
|
||||||
'alarm/command', 'ARM_AWAY', 0, False)
|
assert call_count == mqtt_mock.async_publish.call_count
|
||||||
|
|
||||||
def test_arm_away_not_publishes_mqtt_with_invalid_code_when_req(self):
|
|
||||||
"""Test not publishing of MQTT messages with invalid code.
|
|
||||||
|
|
||||||
When code_arm_required = True
|
async def test_arm_home_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||||
"""
|
"""Test publishing of MQTT messages.
|
||||||
assert setup_component(self.hass, 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
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
call_count = self.mock_publish.call_count
|
When code_arm_required = False
|
||||||
common.alarm_arm_away(self.hass, 'abcd')
|
"""
|
||||||
self.hass.block_till_done()
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
assert call_count == self.mock_publish.call_count
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
'code': '1234',
|
||||||
|
'code_arm_required': False
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
def test_arm_away_publishes_mqtt_when_code_not_req(self):
|
common.async_alarm_arm_home(hass)
|
||||||
"""Test publishing of MQTT messages.
|
await hass.async_block_till_done()
|
||||||
|
mqtt_mock.async_publish.assert_called_once_with(
|
||||||
|
'alarm/command', 'ARM_HOME', 0, False)
|
||||||
|
|
||||||
When code_arm_required = False
|
|
||||||
"""
|
|
||||||
assert setup_component(self.hass, 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
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
common.alarm_arm_away(self.hass)
|
async def test_arm_away_publishes_mqtt(hass, mqtt_mock):
|
||||||
self.hass.block_till_done()
|
"""Test publishing of MQTT messages while armed."""
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
'alarm/command', 'ARM_AWAY', 0, False)
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
def test_arm_night_publishes_mqtt(self):
|
common.async_alarm_arm_away(hass)
|
||||||
"""Test publishing of MQTT messages while armed."""
|
await hass.async_block_till_done()
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
mqtt_mock.async_publish.assert_called_once_with(
|
||||||
alarm_control_panel.DOMAIN: {
|
'alarm/command', 'ARM_AWAY', 0, False)
|
||||||
'platform': 'mqtt',
|
|
||||||
'name': 'test',
|
|
||||||
'state_topic': 'alarm/state',
|
|
||||||
'command_topic': 'alarm/command',
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
common.alarm_arm_night(self.hass)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
|
||||||
'alarm/command', 'ARM_NIGHT', 0, False)
|
|
||||||
|
|
||||||
def test_arm_night_not_publishes_mqtt_with_invalid_code_when_req(self):
|
async def test_arm_away_not_publishes_mqtt_with_invalid_code_when_req(
|
||||||
"""Test not publishing of MQTT messages with invalid code.
|
hass, mqtt_mock):
|
||||||
|
"""Test not publishing of MQTT messages with invalid code.
|
||||||
|
|
||||||
When code_arm_required = True
|
When code_arm_required = True
|
||||||
"""
|
"""
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
alarm_control_panel.DOMAIN: {
|
alarm_control_panel.DOMAIN: {
|
||||||
'platform': 'mqtt',
|
'platform': 'mqtt',
|
||||||
'name': 'test',
|
'name': 'test',
|
||||||
'state_topic': 'alarm/state',
|
'state_topic': 'alarm/state',
|
||||||
'command_topic': 'alarm/command',
|
'command_topic': 'alarm/command',
|
||||||
'code': '1234',
|
'code': '1234',
|
||||||
'code_arm_required': True
|
'code_arm_required': True
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
call_count = self.mock_publish.call_count
|
call_count = mqtt_mock.async_publish.call_count
|
||||||
common.alarm_arm_night(self.hass, 'abcd')
|
common.async_alarm_arm_away(hass, 'abcd')
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert call_count == self.mock_publish.call_count
|
assert call_count == mqtt_mock.async_publish.call_count
|
||||||
|
|
||||||
def test_arm_night_publishes_mqtt_when_code_not_req(self):
|
|
||||||
"""Test publishing of MQTT messages.
|
|
||||||
|
|
||||||
When code_arm_required = False
|
async def test_arm_away_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||||
"""
|
"""Test publishing of MQTT messages.
|
||||||
assert setup_component(self.hass, 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
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
common.alarm_arm_night(self.hass)
|
When code_arm_required = False
|
||||||
self.hass.block_till_done()
|
"""
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
'alarm/command', 'ARM_NIGHT', 0, False)
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
'code': '1234',
|
||||||
|
'code_arm_required': False
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
def test_disarm_publishes_mqtt(self):
|
common.async_alarm_arm_away(hass)
|
||||||
"""Test publishing of MQTT messages while disarmed."""
|
await hass.async_block_till_done()
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
mqtt_mock.async_publish.assert_called_once_with(
|
||||||
alarm_control_panel.DOMAIN: {
|
'alarm/command', 'ARM_AWAY', 0, False)
|
||||||
'platform': 'mqtt',
|
|
||||||
'name': 'test',
|
|
||||||
'state_topic': 'alarm/state',
|
|
||||||
'command_topic': 'alarm/command',
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
common.alarm_disarm(self.hass)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
|
||||||
'alarm/command', 'DISARM', 0, False)
|
|
||||||
|
|
||||||
def test_disarm_publishes_mqtt_with_template(self):
|
async def test_arm_night_publishes_mqtt(hass, mqtt_mock):
|
||||||
"""Test publishing of MQTT messages while disarmed.
|
"""Test publishing of MQTT messages while armed."""
|
||||||
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
When command_template set to output json
|
common.async_alarm_arm_night(hass)
|
||||||
"""
|
await hass.async_block_till_done()
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
mqtt_mock.async_publish.assert_called_once_with(
|
||||||
alarm_control_panel.DOMAIN: {
|
'alarm/command', 'ARM_NIGHT', 0, False)
|
||||||
'platform': 'mqtt',
|
|
||||||
'name': 'test',
|
|
||||||
'state_topic': 'alarm/state',
|
|
||||||
'command_topic': 'alarm/command',
|
|
||||||
'code': '1234',
|
|
||||||
'command_template': '{\"action\":\"{{ action }}\",'
|
|
||||||
'\"code\":\"{{ code }}\"}',
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
common.alarm_disarm(self.hass, 1234)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
|
||||||
'alarm/command', '{\"action\":\"DISARM\",\"code\":\"1234\"}',
|
|
||||||
0,
|
|
||||||
False)
|
|
||||||
|
|
||||||
def test_disarm_publishes_mqtt_when_code_not_req(self):
|
async def test_arm_night_not_publishes_mqtt_with_invalid_code_when_req(
|
||||||
"""Test publishing of MQTT messages while disarmed.
|
hass, mqtt_mock):
|
||||||
|
"""Test not publishing of MQTT messages with invalid code.
|
||||||
|
|
||||||
When code_disarm_required = False
|
When code_arm_required = True
|
||||||
"""
|
"""
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
alarm_control_panel.DOMAIN: {
|
alarm_control_panel.DOMAIN: {
|
||||||
'platform': 'mqtt',
|
'platform': 'mqtt',
|
||||||
'name': 'test',
|
'name': 'test',
|
||||||
'state_topic': 'alarm/state',
|
'state_topic': 'alarm/state',
|
||||||
'command_topic': 'alarm/command',
|
'command_topic': 'alarm/command',
|
||||||
'code': '1234',
|
'code': '1234',
|
||||||
'code_disarm_required': False
|
'code_arm_required': True
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
common.alarm_disarm(self.hass)
|
call_count = mqtt_mock.async_publish.call_count
|
||||||
self.hass.block_till_done()
|
common.async_alarm_arm_night(hass, 'abcd')
|
||||||
self.mock_publish.async_publish.assert_called_once_with(
|
await hass.async_block_till_done()
|
||||||
'alarm/command', 'DISARM', 0, False)
|
assert call_count == mqtt_mock.async_publish.call_count
|
||||||
|
|
||||||
def test_disarm_not_publishes_mqtt_with_invalid_code_when_req(self):
|
|
||||||
"""Test not publishing of MQTT messages with invalid code.
|
|
||||||
|
|
||||||
When code_disarm_required = True
|
async def test_arm_night_publishes_mqtt_when_code_not_req(hass, mqtt_mock):
|
||||||
"""
|
"""Test publishing of MQTT messages.
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
|
||||||
alarm_control_panel.DOMAIN: {
|
|
||||||
'platform': 'mqtt',
|
|
||||||
'name': 'test',
|
|
||||||
'state_topic': 'alarm/state',
|
|
||||||
'command_topic': 'alarm/command',
|
|
||||||
'code': '1234',
|
|
||||||
'code_disarm_required': True
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
call_count = self.mock_publish.call_count
|
When code_arm_required = False
|
||||||
common.alarm_disarm(self.hass, 'abcd')
|
"""
|
||||||
self.hass.block_till_done()
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
assert call_count == self.mock_publish.call_count
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
'code': '1234',
|
||||||
|
'code_arm_required': False
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
def test_default_availability_payload(self):
|
common.async_alarm_arm_night(hass)
|
||||||
"""Test availability by default payload with defined topic."""
|
await hass.async_block_till_done()
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
mqtt_mock.async_publish.assert_called_once_with(
|
||||||
alarm_control_panel.DOMAIN: {
|
'alarm/command', 'ARM_NIGHT', 0, False)
|
||||||
'platform': 'mqtt',
|
|
||||||
'name': 'test',
|
|
||||||
'state_topic': 'alarm/state',
|
|
||||||
'command_topic': 'alarm/command',
|
|
||||||
'code': '1234',
|
|
||||||
'availability_topic': 'availability-topic'
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
state = self.hass.states.get('alarm_control_panel.test')
|
|
||||||
assert STATE_UNAVAILABLE == state.state
|
|
||||||
|
|
||||||
fire_mqtt_message(self.hass, 'availability-topic', 'online')
|
async def test_disarm_publishes_mqtt(hass, mqtt_mock):
|
||||||
self.hass.block_till_done()
|
"""Test publishing of MQTT messages while disarmed."""
|
||||||
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
state = self.hass.states.get('alarm_control_panel.test')
|
common.async_alarm_disarm(hass)
|
||||||
assert STATE_UNAVAILABLE != state.state
|
await hass.async_block_till_done()
|
||||||
|
mqtt_mock.async_publish.assert_called_once_with(
|
||||||
|
'alarm/command', 'DISARM', 0, False)
|
||||||
|
|
||||||
fire_mqtt_message(self.hass, 'availability-topic', 'offline')
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
state = self.hass.states.get('alarm_control_panel.test')
|
async def test_disarm_publishes_mqtt_with_template(hass, mqtt_mock):
|
||||||
assert STATE_UNAVAILABLE == state.state
|
"""Test publishing of MQTT messages while disarmed.
|
||||||
|
|
||||||
def test_custom_availability_payload(self):
|
When command_template set to output json
|
||||||
"""Test availability by custom payload with defined topic."""
|
"""
|
||||||
assert setup_component(self.hass, alarm_control_panel.DOMAIN, {
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
alarm_control_panel.DOMAIN: {
|
alarm_control_panel.DOMAIN: {
|
||||||
'platform': 'mqtt',
|
'platform': 'mqtt',
|
||||||
'name': 'test',
|
'name': 'test',
|
||||||
'state_topic': 'alarm/state',
|
'state_topic': 'alarm/state',
|
||||||
'command_topic': 'alarm/command',
|
'command_topic': 'alarm/command',
|
||||||
'code': '1234',
|
'code': '1234',
|
||||||
'availability_topic': 'availability-topic',
|
'command_template': '{\"action\":\"{{ action }}\",'
|
||||||
'payload_available': 'good',
|
'\"code\":\"{{ code }}\"}',
|
||||||
'payload_not_available': 'nogood'
|
}
|
||||||
}
|
})
|
||||||
})
|
|
||||||
|
|
||||||
state = self.hass.states.get('alarm_control_panel.test')
|
common.async_alarm_disarm(hass, 1234)
|
||||||
assert STATE_UNAVAILABLE == state.state
|
await hass.async_block_till_done()
|
||||||
|
mqtt_mock.async_publish.assert_called_once_with(
|
||||||
|
'alarm/command', '{\"action\":\"DISARM\",\"code\":\"1234\"}',
|
||||||
|
0,
|
||||||
|
False)
|
||||||
|
|
||||||
fire_mqtt_message(self.hass, 'availability-topic', 'good')
|
|
||||||
|
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
|
||||||
|
"""
|
||||||
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
'code': '1234',
|
||||||
|
'code_disarm_required': False
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
common.async_alarm_disarm(hass)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
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, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
'code': '1234',
|
||||||
|
'code_disarm_required': True
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
call_count = mqtt_mock.async_publish.call_count
|
||||||
|
common.async_alarm_disarm(hass, 'abcd')
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
assert call_count == mqtt_mock.async_publish.call_count
|
||||||
|
|
||||||
|
|
||||||
|
async def test_default_availability_payload(hass, mqtt_mock):
|
||||||
|
"""Test availability by default payload with defined topic."""
|
||||||
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
'code': '1234',
|
||||||
|
'availability_topic': 'availability-topic'
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
|
assert STATE_UNAVAILABLE == state.state
|
||||||
|
|
||||||
|
async_fire_mqtt_message(hass, 'availability-topic', 'online')
|
||||||
|
|
||||||
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
|
assert STATE_UNAVAILABLE != state.state
|
||||||
|
|
||||||
|
async_fire_mqtt_message(hass, 'availability-topic', 'offline')
|
||||||
|
|
||||||
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
|
assert STATE_UNAVAILABLE == state.state
|
||||||
|
|
||||||
|
|
||||||
|
async def test_custom_availability_payload(hass, mqtt_mock):
|
||||||
|
"""Test availability by custom payload with defined topic."""
|
||||||
|
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, {
|
||||||
|
alarm_control_panel.DOMAIN: {
|
||||||
|
'platform': 'mqtt',
|
||||||
|
'name': 'test',
|
||||||
|
'state_topic': 'alarm/state',
|
||||||
|
'command_topic': 'alarm/command',
|
||||||
|
'code': '1234',
|
||||||
|
'availability_topic': 'availability-topic',
|
||||||
|
'payload_available': 'good',
|
||||||
|
'payload_not_available': 'nogood'
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
|
assert STATE_UNAVAILABLE == state.state
|
||||||
|
|
||||||
|
async_fire_mqtt_message(hass, 'availability-topic', 'good')
|
||||||
|
|
||||||
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
|
assert STATE_UNAVAILABLE != state.state
|
||||||
|
|
||||||
|
async_fire_mqtt_message(hass, 'availability-topic', 'nogood')
|
||||||
|
|
||||||
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
|
assert STATE_UNAVAILABLE == state.state
|
||||||
|
|
||||||
|
|
||||||
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
|
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
|
||||||
@ -416,7 +425,6 @@ async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
|
|||||||
})
|
})
|
||||||
|
|
||||||
async_fire_mqtt_message(hass, 'attr-topic', '{ "val": "100" }')
|
async_fire_mqtt_message(hass, 'attr-topic', '{ "val": "100" }')
|
||||||
await hass.async_block_till_done()
|
|
||||||
state = hass.states.get('alarm_control_panel.test')
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
|
|
||||||
assert '100' == state.attributes.get('val')
|
assert '100' == state.attributes.get('val')
|
||||||
@ -443,7 +451,6 @@ async def test_update_state_via_state_topic_template(hass, mqtt_mock):
|
|||||||
assert STATE_UNKNOWN == state.state
|
assert STATE_UNKNOWN == state.state
|
||||||
|
|
||||||
async_fire_mqtt_message(hass, 'test-topic', '100')
|
async_fire_mqtt_message(hass, 'test-topic', '100')
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
state = hass.states.get('alarm_control_panel.test')
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
assert STATE_ALARM_ARMED_AWAY == state.state
|
assert STATE_ALARM_ARMED_AWAY == state.state
|
||||||
@ -462,7 +469,6 @@ async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
|
|||||||
})
|
})
|
||||||
|
|
||||||
async_fire_mqtt_message(hass, 'attr-topic', '[ "list", "of", "things"]')
|
async_fire_mqtt_message(hass, 'attr-topic', '[ "list", "of", "things"]')
|
||||||
await hass.async_block_till_done()
|
|
||||||
state = hass.states.get('alarm_control_panel.test')
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
|
|
||||||
assert state.attributes.get('val') is None
|
assert state.attributes.get('val') is None
|
||||||
@ -482,7 +488,6 @@ async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
|
|||||||
})
|
})
|
||||||
|
|
||||||
async_fire_mqtt_message(hass, 'attr-topic', 'This is not JSON')
|
async_fire_mqtt_message(hass, 'attr-topic', 'This is not JSON')
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
state = hass.states.get('alarm_control_panel.test')
|
state = hass.states.get('alarm_control_panel.test')
|
||||||
assert state.attributes.get('val') is None
|
assert state.attributes.get('val') is None
|
||||||
@ -507,8 +512,6 @@ async def test_discovery_update_attr(hass, mqtt_mock, caplog):
|
|||||||
hass, 'homeassistant/alarm_control_panel/bla/config', data1)
|
hass, 'homeassistant/alarm_control_panel/bla/config', data1)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
async_fire_mqtt_message(hass, 'attr-topic1', '{ "val": "100" }')
|
async_fire_mqtt_message(hass, 'attr-topic1', '{ "val": "100" }')
|
||||||
await hass.async_block_till_done()
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
state = hass.states.get('alarm_control_panel.beer')
|
state = hass.states.get('alarm_control_panel.beer')
|
||||||
assert '100' == state.attributes.get('val')
|
assert '100' == state.attributes.get('val')
|
||||||
|
|
||||||
@ -516,19 +519,14 @@ async def test_discovery_update_attr(hass, mqtt_mock, caplog):
|
|||||||
async_fire_mqtt_message(
|
async_fire_mqtt_message(
|
||||||
hass, 'homeassistant/alarm_control_panel/bla/config', data2)
|
hass, 'homeassistant/alarm_control_panel/bla/config', data2)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
# Verify we are no longer subscribing to the old topic
|
# Verify we are no longer subscribing to the old topic
|
||||||
async_fire_mqtt_message(hass, 'attr-topic1', '{ "val": "50" }')
|
async_fire_mqtt_message(hass, 'attr-topic1', '{ "val": "50" }')
|
||||||
await hass.async_block_till_done()
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
state = hass.states.get('alarm_control_panel.beer')
|
state = hass.states.get('alarm_control_panel.beer')
|
||||||
assert '100' == state.attributes.get('val')
|
assert '100' == state.attributes.get('val')
|
||||||
|
|
||||||
# Verify we are subscribing to the new topic
|
# Verify we are subscribing to the new topic
|
||||||
async_fire_mqtt_message(hass, 'attr-topic2', '{ "val": "75" }')
|
async_fire_mqtt_message(hass, 'attr-topic2', '{ "val": "75" }')
|
||||||
await hass.async_block_till_done()
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
state = hass.states.get('alarm_control_panel.beer')
|
state = hass.states.get('alarm_control_panel.beer')
|
||||||
assert '75' == state.attributes.get('val')
|
assert '75' == state.attributes.get('val')
|
||||||
|
|
||||||
@ -552,7 +550,6 @@ async def test_unique_id(hass):
|
|||||||
}]
|
}]
|
||||||
})
|
})
|
||||||
async_fire_mqtt_message(hass, 'test-topic', 'payload')
|
async_fire_mqtt_message(hass, 'test-topic', 'payload')
|
||||||
await hass.async_block_till_done()
|
|
||||||
assert len(hass.states.async_entity_ids(alarm_control_panel.DOMAIN)) == 1
|
assert len(hass.states.async_entity_ids(alarm_control_panel.DOMAIN)) == 1
|
||||||
|
|
||||||
|
|
||||||
@ -580,7 +577,6 @@ async def test_discovery_removal_alarm(hass, mqtt_mock, caplog):
|
|||||||
'homeassistant/alarm_control_panel/bla/config',
|
'homeassistant/alarm_control_panel/bla/config',
|
||||||
'')
|
'')
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
state = hass.states.get('alarm_control_panel.beer')
|
state = hass.states.get('alarm_control_panel.beer')
|
||||||
assert state is None
|
assert state is None
|
||||||
@ -615,7 +611,6 @@ async def test_discovery_update_alarm(hass, mqtt_mock, caplog):
|
|||||||
'homeassistant/alarm_control_panel/bla/config',
|
'homeassistant/alarm_control_panel/bla/config',
|
||||||
data2)
|
data2)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
state = hass.states.get('alarm_control_panel.beer')
|
state = hass.states.get('alarm_control_panel.beer')
|
||||||
assert state is not None
|
assert state is not None
|
||||||
@ -651,7 +646,6 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
|
|||||||
'homeassistant/alarm_control_panel/bla/config',
|
'homeassistant/alarm_control_panel/bla/config',
|
||||||
data2)
|
data2)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
state = hass.states.get('alarm_control_panel.milk')
|
state = hass.states.get('alarm_control_panel.milk')
|
||||||
assert state is not None
|
assert state is not None
|
||||||
@ -687,7 +681,6 @@ async def test_entity_device_info_with_identifier(hass, mqtt_mock):
|
|||||||
async_fire_mqtt_message(
|
async_fire_mqtt_message(
|
||||||
hass, 'homeassistant/alarm_control_panel/bla/config', data)
|
hass, 'homeassistant/alarm_control_panel/bla/config', data)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
device = registry.async_get_device({('mqtt', 'helloworld')}, set())
|
device = registry.async_get_device({('mqtt', 'helloworld')}, set())
|
||||||
assert device is not None
|
assert device is not None
|
||||||
@ -728,7 +721,6 @@ async def test_entity_device_info_update(hass, mqtt_mock):
|
|||||||
async_fire_mqtt_message(
|
async_fire_mqtt_message(
|
||||||
hass, 'homeassistant/alarm_control_panel/bla/config', data)
|
hass, 'homeassistant/alarm_control_panel/bla/config', data)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
device = registry.async_get_device({('mqtt', 'helloworld')}, set())
|
device = registry.async_get_device({('mqtt', 'helloworld')}, set())
|
||||||
assert device is not None
|
assert device is not None
|
||||||
@ -739,7 +731,6 @@ async def test_entity_device_info_update(hass, mqtt_mock):
|
|||||||
async_fire_mqtt_message(
|
async_fire_mqtt_message(
|
||||||
hass, 'homeassistant/alarm_control_panel/bla/config', data)
|
hass, 'homeassistant/alarm_control_panel/bla/config', data)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
device = registry.async_get_device({('mqtt', 'helloworld')}, set())
|
device = registry.async_get_device({('mqtt', 'helloworld')}, set())
|
||||||
assert device is not None
|
assert device is not None
|
||||||
@ -771,7 +762,6 @@ async def test_entity_id_update(hass, mqtt_mock):
|
|||||||
registry.async_update_entity(
|
registry.async_update_entity(
|
||||||
'alarm_control_panel.beer', new_entity_id='alarm_control_panel.milk')
|
'alarm_control_panel.beer', new_entity_id='alarm_control_panel.milk')
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
state = hass.states.get('alarm_control_panel.beer')
|
state = hass.states.get('alarm_control_panel.beer')
|
||||||
assert state is None
|
assert state is None
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user