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

Async tests for MQTT lock (#17763)

Browse Source
This commit is contained in:
kennedyshead 2018-10-24 22:20:25 +02:00 committed by Paulus Schoutsen
parent 52974ff742
commit c099c259ea
1 changed files with 100 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

244
tests/components/lock/test_mqtt.py
View File

@ -1,181 +1,137 @@
"""The tests for the MQTT lock platform.""" """The tests for the MQTT lock platform."""
import unittest from homeassistant.setup import async_setup_component
from homeassistant.setup import setup_component
from homeassistant.const import ( from homeassistant.const import (
STATE_LOCKED, STATE_UNLOCKED, STATE_UNAVAILABLE, ATTR_ASSUMED_STATE) STATE_LOCKED, STATE_UNLOCKED, STATE_UNAVAILABLE, ATTR_ASSUMED_STATE)
import homeassistant.components.lock as lock import homeassistant.components.lock as lock
from homeassistant.components.mqtt.discovery import async_start from homeassistant.components.mqtt.discovery import async_start
from tests.common import ( from tests.common import async_fire_mqtt_message
mock_mqtt_component, async_fire_mqtt_message, fire_mqtt_message,
get_test_home_assistant)
from tests.components.lock import common
class TestLockMQTT(unittest.TestCase): async def test_controlling_state_via_topic(hass, mqtt_mock):
"""Test the MQTT lock.""" """Test the controlling state via topic."""
assert await async_setup_component(hass, lock.DOMAIN, {
lock.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_lock': 'LOCK',
'payload_unlock': 'UNLOCK'
}
})
def setUp(self): # pylint: disable=invalid-name state = hass.states.get('lock.test')
"""Set up things to be run when tests are started.""" assert state.state is STATE_UNLOCKED
self.hass = get_test_home_assistant() assert not state.attributes.get(ATTR_ASSUMED_STATE)
self.mock_publish = mock_mqtt_component(self.hass)
def tearDown(self): # pylint: disable=invalid-name async_fire_mqtt_message(hass, 'state-topic', 'LOCK')
"""Stop everything that was started.""" await hass.async_block_till_done()
self.hass.stop()
def test_controlling_state_via_topic(self): state = hass.states.get('lock.test')
"""Test the controlling state via topic.""" assert state.state is STATE_LOCKED
assert setup_component(self.hass, lock.DOMAIN, {
lock.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_lock': 'LOCK',
'payload_unlock': 'UNLOCK'
}
})
state = self.hass.states.get('lock.test') async_fire_mqtt_message(hass, 'state-topic', 'UNLOCK')
assert STATE_UNLOCKED == state.state await hass.async_block_till_done()
assert not state.attributes.get(ATTR_ASSUMED_STATE) await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'state-topic', 'LOCK') state = hass.states.get('lock.test')
self.hass.block_till_done() assert state.state is STATE_UNLOCKED
state = self.hass.states.get('lock.test')
assert STATE_LOCKED == state.state
fire_mqtt_message(self.hass, 'state-topic', 'UNLOCK') async def test_controlling_state_via_topic_and_json_message(hass, mqtt_mock):
self.hass.block_till_done() """Test the controlling state via topic and JSON message."""
assert await async_setup_component(hass, lock.DOMAIN, {
lock.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_lock': 'LOCK',
'payload_unlock': 'UNLOCK',
'value_template': '{{ value_json.val }}'
}
})
state = self.hass.states.get('lock.test') state = hass.states.get('lock.test')
assert STATE_UNLOCKED == state.state assert state.state is STATE_UNLOCKED
def test_sending_mqtt_commands_and_optimistic(self): async_fire_mqtt_message(hass, 'state-topic', '{"val":"LOCK"}')
"""Test the sending MQTT commands in optimistic mode.""" await hass.async_block_till_done()
assert setup_component(self.hass, lock.DOMAIN, {
lock.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'command_topic': 'command-topic',
'payload_lock': 'LOCK',
'payload_unlock': 'UNLOCK',
'qos': 2
}
})
state = self.hass.states.get('lock.test') state = hass.states.get('lock.test')
assert STATE_UNLOCKED == state.state assert state.state is STATE_LOCKED
assert state.attributes.get(ATTR_ASSUMED_STATE)
common.lock(self.hass, 'lock.test') async_fire_mqtt_message(hass, 'state-topic', '{"val":"UNLOCK"}')
self.hass.block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
self.mock_publish.async_publish.assert_called_once_with( state = hass.states.get('lock.test')
'command-topic', 'LOCK', 2, False) assert state.state is STATE_UNLOCKED
self.mock_publish.async_publish.reset_mock()
state = self.hass.states.get('lock.test')
assert STATE_LOCKED == state.state
common.unlock(self.hass, 'lock.test')
self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with( async def test_default_availability_payload(hass, mqtt_mock):
'command-topic', 'UNLOCK', 2, False) """Test availability by default payload with defined topic."""
state = self.hass.states.get('lock.test') assert await async_setup_component(hass, lock.DOMAIN, {
assert STATE_UNLOCKED == state.state lock.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_lock': 'LOCK',
'payload_unlock': 'UNLOCK',
'availability_topic': 'availability-topic'
}
})
def test_controlling_state_via_topic_and_json_message(self): state = hass.states.get('lock.test')
"""Test the controlling state via topic and JSON message.""" assert state.state is STATE_UNAVAILABLE
assert setup_component(self.hass, lock.DOMAIN, {
lock.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_lock': 'LOCK',
'payload_unlock': 'UNLOCK',
'value_template': '{{ value_json.val }}'
}
})
state = self.hass.states.get('lock.test') async_fire_mqtt_message(hass, 'availability-topic', 'online')
assert STATE_UNLOCKED == state.state await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'state-topic', '{"val":"LOCK"}') state = hass.states.get('lock.test')
self.hass.block_till_done() assert state.state is not STATE_UNAVAILABLE
state = self.hass.states.get('lock.test') async_fire_mqtt_message(hass, 'availability-topic', 'offline')
assert STATE_LOCKED == state.state await hass.async_block_till_done()
await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'state-topic', '{"val":"UNLOCK"}') state = hass.states.get('lock.test')
self.hass.block_till_done() assert state.state is STATE_UNAVAILABLE
state = self.hass.states.get('lock.test')
assert STATE_UNLOCKED == state.state
def test_default_availability_payload(self): async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic.""" """Test availability by custom payload with defined topic."""
assert setup_component(self.hass, lock.DOMAIN, { assert await async_setup_component(hass, lock.DOMAIN, {
lock.DOMAIN: { lock.DOMAIN: {
'platform': 'mqtt', 'platform': 'mqtt',
'name': 'test', 'name': 'test',
'state_topic': 'state-topic', 'state_topic': 'state-topic',
'command_topic': 'command-topic', 'command_topic': 'command-topic',
'payload_lock': 'LOCK', 'payload_lock': 'LOCK',
'payload_unlock': 'UNLOCK', 'payload_unlock': 'UNLOCK',
'availability_topic': 'availability-topic' 'availability_topic': 'availability-topic',
} 'payload_available': 'good',
}) 'payload_not_available': 'nogood'
}
})
state = self.hass.states.get('lock.test') state = hass.states.get('lock.test')
assert STATE_UNAVAILABLE == state.state assert state.state is STATE_UNAVAILABLE
fire_mqtt_message(self.hass, 'availability-topic', 'online') async_fire_mqtt_message(hass, 'availability-topic', 'good')
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get('lock.test') state = hass.states.get('lock.test')
assert STATE_UNAVAILABLE != state.state assert state.state is not STATE_UNAVAILABLE
fire_mqtt_message(self.hass, 'availability-topic', 'offline') async_fire_mqtt_message(hass, 'availability-topic', 'nogood')
self.hass.block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
state = self.hass.states.get('lock.test') state = hass.states.get('lock.test')
assert STATE_UNAVAILABLE == state.state assert state.state is STATE_UNAVAILABLE
def test_custom_availability_payload(self):
"""Test availability by custom payload with defined topic."""
assert setup_component(self.hass, lock.DOMAIN, {
lock.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_lock': 'LOCK',
'payload_unlock': 'UNLOCK',
'availability_topic': 'availability-topic',
'payload_available': 'good',
'payload_not_available': 'nogood'
}
})
state = self.hass.states.get('lock.test')
assert STATE_UNAVAILABLE == state.state
fire_mqtt_message(self.hass, 'availability-topic', 'good')
self.hass.block_till_done()
state = self.hass.states.get('lock.test')
assert STATE_UNAVAILABLE != state.state
fire_mqtt_message(self.hass, 'availability-topic', 'nogood')
self.hass.block_till_done()
state = self.hass.states.get('lock.test')
assert STATE_UNAVAILABLE == state.state
async def test_discovery_removal_lock(hass, mqtt_mock, caplog): async def test_discovery_removal_lock(hass, mqtt_mock, caplog):