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

Async tests for mqtt switch (#18685)

Browse Source
This commit is contained in:
Adam Mills 2018-11-24 17:08:28 -05:00 committed by GitHub
parent 6f0a3b4b22
commit 00c9ca64c8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 241 additions and 221 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

462
tests/components/switch/test_mqtt.py
View File

@ -1,9 +1,9 @@
"""The tests for the MQTT switch platform.""" """The tests for the MQTT switch platform."""
import json import json
import unittest from asynctest import patch
from unittest.mock import patch import pytest
from homeassistant.setup import setup_component, async_setup_component from homeassistant.setup import async_setup_component
from homeassistant.const import STATE_ON, STATE_OFF, STATE_UNAVAILABLE,\ from homeassistant.const import STATE_ON, STATE_OFF, STATE_UNAVAILABLE,\
ATTR_ASSUMED_STATE ATTR_ASSUMED_STATE
import homeassistant.core as ha import homeassistant.core as ha
@ -11,279 +11,297 @@ from homeassistant.components import switch, mqtt
from homeassistant.components.mqtt.discovery import async_start from homeassistant.components.mqtt.discovery import async_start
from tests.common import ( from tests.common import (
mock_mqtt_component, fire_mqtt_message, get_test_home_assistant, mock_coro, mock_coro, async_mock_mqtt_component, async_fire_mqtt_message,
async_mock_mqtt_component, async_fire_mqtt_message, MockConfigEntry) MockConfigEntry)
from tests.components.switch import common from tests.components.switch import common
class TestSwitchMQTT(unittest.TestCase): @pytest.fixture
"""Test the MQTT switch.""" def mock_publish(hass):
"""Initialize components."""
yield hass.loop.run_until_complete(async_mock_mqtt_component(hass))
def setUp(self): # pylint: disable=invalid-name
"""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 async def test_controlling_state_via_topic(hass, mock_publish):
"""Stop everything that was started.""" """Test the controlling state via topic."""
self.hass.stop() assert await async_setup_component(hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_on': 1,
'payload_off': 0
}
})
def test_controlling_state_via_topic(self): state = hass.states.get('switch.test')
"""Test the controlling state via topic.""" assert STATE_OFF == state.state
assert setup_component(self.hass, switch.DOMAIN, { assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, 'state-topic', '1')
await hass.async_block_till_done()
state = hass.states.get('switch.test')
assert STATE_ON == state.state
async_fire_mqtt_message(hass, 'state-topic', '0')
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('switch.test')
assert STATE_OFF == state.state
async def test_sending_mqtt_commands_and_optimistic(hass, mock_publish):
"""Test the sending MQTT commands in optimistic mode."""
fake_state = ha.State('switch.test', 'on')
with patch('homeassistant.components.switch.mqtt.async_get_last_state',
return_value=mock_coro(fake_state)):
assert await async_setup_component(hass, switch.DOMAIN, {
switch.DOMAIN: { switch.DOMAIN: {
'platform': 'mqtt', 'platform': 'mqtt',
'name': 'test', 'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_on': 1,
'payload_off': 0
}
})
state = self.hass.states.get('switch.test')
assert STATE_OFF == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'state-topic', '1')
self.hass.block_till_done()
state = self.hass.states.get('switch.test')
assert STATE_ON == state.state
fire_mqtt_message(self.hass, 'state-topic', '0')
self.hass.block_till_done()
state = self.hass.states.get('switch.test')
assert STATE_OFF == state.state
def test_sending_mqtt_commands_and_optimistic(self):
"""Test the sending MQTT commands in optimistic mode."""
fake_state = ha.State('switch.test', 'on')
with patch('homeassistant.components.switch.mqtt.async_get_last_state',
return_value=mock_coro(fake_state)):
assert setup_component(self.hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'command_topic': 'command-topic',
'payload_on': 'beer on',
'payload_off': 'beer off',
'qos': '2'
}
})
state = self.hass.states.get('switch.test')
assert STATE_ON == state.state
assert state.attributes.get(ATTR_ASSUMED_STATE)
common.turn_on(self.hass, 'switch.test')
self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with(
'command-topic', 'beer on', 2, False)
self.mock_publish.async_publish.reset_mock()
state = self.hass.states.get('switch.test')
assert STATE_ON == state.state
common.turn_off(self.hass, 'switch.test')
self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with(
'command-topic', 'beer off', 2, False)
state = self.hass.states.get('switch.test')
assert STATE_OFF == state.state
def test_controlling_state_via_topic_and_json_message(self):
"""Test the controlling state via topic and JSON message."""
assert setup_component(self.hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic', 'command_topic': 'command-topic',
'payload_on': 'beer on', 'payload_on': 'beer on',
'payload_off': 'beer off', 'payload_off': 'beer off',
'value_template': '{{ value_json.val }}' 'qos': '2'
} }
}) })
state = self.hass.states.get('switch.test') state = hass.states.get('switch.test')
assert STATE_OFF == state.state assert STATE_ON == state.state
assert state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'state-topic', '{"val":"beer on"}') common.turn_on(hass, 'switch.test')
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get('switch.test') mock_publish.async_publish.assert_called_once_with(
assert STATE_ON == state.state 'command-topic', 'beer on', 2, False)
mock_publish.async_publish.reset_mock()
state = hass.states.get('switch.test')
assert STATE_ON == state.state
fire_mqtt_message(self.hass, 'state-topic', '{"val":"beer off"}') common.turn_off(hass, 'switch.test')
self.hass.block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
state = self.hass.states.get('switch.test') mock_publish.async_publish.assert_called_once_with(
assert STATE_OFF == state.state 'command-topic', 'beer off', 2, False)
state = hass.states.get('switch.test')
assert STATE_OFF == state.state
def test_controlling_availability(self):
"""Test the controlling state via topic."""
assert setup_component(self.hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'availability_topic': 'availability_topic',
'payload_on': 1,
'payload_off': 0,
'payload_available': 1,
'payload_not_available': 0
}
})
state = self.hass.states.get('switch.test') async def test_controlling_state_via_topic_and_json_message(
assert STATE_UNAVAILABLE == state.state hass, mock_publish):
"""Test the controlling state via topic and JSON message."""
assert await async_setup_component(hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_on': 'beer on',
'payload_off': 'beer off',
'value_template': '{{ value_json.val }}'
}
})
fire_mqtt_message(self.hass, 'availability_topic', '1') state = hass.states.get('switch.test')
self.hass.block_till_done() assert STATE_OFF == state.state
state = self.hass.states.get('switch.test') async_fire_mqtt_message(hass, 'state-topic', '{"val":"beer on"}')
assert STATE_OFF == 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, 'availability_topic', '0') state = hass.states.get('switch.test')
self.hass.block_till_done() assert STATE_ON == state.state
state = self.hass.states.get('switch.test') async_fire_mqtt_message(hass, 'state-topic', '{"val":"beer off"}')
assert STATE_UNAVAILABLE == state.state await hass.async_block_till_done()
await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'state-topic', '1') state = hass.states.get('switch.test')
self.hass.block_till_done() assert STATE_OFF == state.state
state = self.hass.states.get('switch.test')
assert STATE_UNAVAILABLE == state.state
fire_mqtt_message(self.hass, 'availability_topic', '1') async def test_controlling_availability(hass, mock_publish):
self.hass.block_till_done() """Test the controlling state via topic."""
assert await async_setup_component(hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'availability_topic': 'availability_topic',
'payload_on': 1,
'payload_off': 0,
'payload_available': 1,
'payload_not_available': 0
}
})
state = self.hass.states.get('switch.test') state = hass.states.get('switch.test')
assert STATE_ON == state.state assert STATE_UNAVAILABLE == state.state
def test_default_availability_payload(self): async_fire_mqtt_message(hass, 'availability_topic', '1')
"""Test the availability payload.""" await hass.async_block_till_done()
assert setup_component(self.hass, switch.DOMAIN, { await hass.async_block_till_done()
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'availability_topic': 'availability_topic',
'payload_on': 1,
'payload_off': 0
}
})
state = self.hass.states.get('switch.test') state = hass.states.get('switch.test')
assert STATE_UNAVAILABLE == state.state assert STATE_OFF == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'availability_topic', 'online') async_fire_mqtt_message(hass, 'availability_topic', '0')
self.hass.block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
state = self.hass.states.get('switch.test') state = hass.states.get('switch.test')
assert STATE_OFF == state.state assert STATE_UNAVAILABLE == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'availability_topic', 'offline') async_fire_mqtt_message(hass, 'state-topic', '1')
self.hass.block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
state = self.hass.states.get('switch.test') state = hass.states.get('switch.test')
assert STATE_UNAVAILABLE == state.state assert STATE_UNAVAILABLE == state.state
fire_mqtt_message(self.hass, 'state-topic', '1') async_fire_mqtt_message(hass, 'availability_topic', '1')
self.hass.block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
state = self.hass.states.get('switch.test') state = hass.states.get('switch.test')
assert STATE_UNAVAILABLE == state.state assert STATE_ON == state.state
fire_mqtt_message(self.hass, 'availability_topic', 'online')
self.hass.block_till_done()
state = self.hass.states.get('switch.test') async def test_default_availability_payload(hass, mock_publish):
assert STATE_ON == state.state """Test the availability payload."""
assert await async_setup_component(hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'availability_topic': 'availability_topic',
'payload_on': 1,
'payload_off': 0
}
})
def test_custom_availability_payload(self): state = hass.states.get('switch.test')
"""Test the availability payload.""" assert STATE_UNAVAILABLE == state.state
assert setup_component(self.hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'availability_topic': 'availability_topic',
'payload_on': 1,
'payload_off': 0,
'payload_available': 'good',
'payload_not_available': 'nogood'
}
})
state = self.hass.states.get('switch.test') async_fire_mqtt_message(hass, 'availability_topic', 'online')
assert STATE_UNAVAILABLE == state.state await hass.async_block_till_done()
await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'availability_topic', 'good') state = hass.states.get('switch.test')
self.hass.block_till_done() assert STATE_OFF == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
state = self.hass.states.get('switch.test') async_fire_mqtt_message(hass, 'availability_topic', 'offline')
assert STATE_OFF == 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, 'availability_topic', 'nogood') state = hass.states.get('switch.test')
self.hass.block_till_done() assert STATE_UNAVAILABLE == state.state
state = self.hass.states.get('switch.test') async_fire_mqtt_message(hass, 'state-topic', '1')
assert STATE_UNAVAILABLE == state.state await hass.async_block_till_done()
await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'state-topic', '1') state = hass.states.get('switch.test')
self.hass.block_till_done() assert STATE_UNAVAILABLE == state.state
state = self.hass.states.get('switch.test') async_fire_mqtt_message(hass, 'availability_topic', 'online')
assert STATE_UNAVAILABLE == state.state await hass.async_block_till_done()
await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'availability_topic', 'good') state = hass.states.get('switch.test')
self.hass.block_till_done() assert STATE_ON == state.state
state = self.hass.states.get('switch.test')
assert STATE_ON == state.state
def test_custom_state_payload(self): async def test_custom_availability_payload(hass, mock_publish):
"""Test the state payload.""" """Test the availability payload."""
assert setup_component(self.hass, switch.DOMAIN, { assert await async_setup_component(hass, switch.DOMAIN, {
switch.DOMAIN: { switch.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_on': 1, 'availability_topic': 'availability_topic',
'payload_off': 0, 'payload_on': 1,
'state_on': "HIGH", 'payload_off': 0,
'state_off': "LOW", 'payload_available': 'good',
} 'payload_not_available': 'nogood'
}) }
})
state = self.hass.states.get('switch.test') state = hass.states.get('switch.test')
assert STATE_OFF == state.state assert STATE_UNAVAILABLE == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'state-topic', 'HIGH') async_fire_mqtt_message(hass, 'availability_topic', 'good')
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get('switch.test') state = hass.states.get('switch.test')
assert STATE_ON == state.state assert STATE_OFF == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'state-topic', 'LOW') 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('switch.test') state = hass.states.get('switch.test')
assert STATE_OFF == state.state assert STATE_UNAVAILABLE == state.state
async_fire_mqtt_message(hass, 'state-topic', '1')
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('switch.test')
assert STATE_UNAVAILABLE == state.state
async_fire_mqtt_message(hass, 'availability_topic', 'good')
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('switch.test')
assert STATE_ON == state.state
async def test_custom_state_payload(hass, mock_publish):
"""Test the state payload."""
assert await async_setup_component(hass, switch.DOMAIN, {
switch.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'payload_on': 1,
'payload_off': 0,
'state_on': "HIGH",
'state_off': "LOW",
}
})
state = hass.states.get('switch.test')
assert STATE_OFF == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, 'state-topic', 'HIGH')
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('switch.test')
assert STATE_ON == state.state
async_fire_mqtt_message(hass, 'state-topic', 'LOW')
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('switch.test')
assert STATE_OFF == state.state
async def test_unique_id(hass): async def test_unique_id(hass):
@ -307,6 +325,7 @@ 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() await hass.async_block_till_done()
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids()) == 2 assert len(hass.states.async_entity_ids()) == 2
# all switches group is 1, unique id created is 1 # all switches group is 1, unique id created is 1
@ -326,6 +345,7 @@ async def test_discovery_removal_switch(hass, mqtt_mock, caplog):
async_fire_mqtt_message(hass, 'homeassistant/switch/bla/config', async_fire_mqtt_message(hass, 'homeassistant/switch/bla/config',
data) data)
await hass.async_block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('switch.beer') state = hass.states.get('switch.beer')
assert state is not None assert state is not None