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

Drop unnecessary block_till_done, improve tests (#23247)

Browse Source
This commit is contained in:
Erik Montnemery 2019-04-19 21:26:56 +02:00 committed by Jason Hu
parent 887e1cd8e3
commit 0e429cca33
1 changed files with 205 additions and 239 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

444
tests/components/mqtt/test_binary_sensor.py
View File

@ -1,285 +1,269 @@
"""The tests for the MQTT binary sensor platform.""" """The tests for the MQTT binary sensor platform."""
from datetime import timedelta from datetime import timedelta
import json import json
import unittest from unittest.mock import ANY
from unittest.mock import ANY, Mock
from homeassistant.components import binary_sensor, mqtt from homeassistant.components import binary_sensor, mqtt
from homeassistant.components.mqtt.discovery import async_start from homeassistant.components.mqtt.discovery import async_start
from homeassistant.const import ( from homeassistant.const import (
EVENT_STATE_CHANGED, STATE_OFF, STATE_ON, STATE_UNAVAILABLE) EVENT_STATE_CHANGED, STATE_OFF, STATE_ON, STATE_UNAVAILABLE)
import homeassistant.core as ha import homeassistant.core as ha
from homeassistant.setup import async_setup_component, setup_component from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from tests.common import ( from tests.common import (
MockConfigEntry, async_fire_mqtt_message, async_mock_mqtt_component, MockConfigEntry, async_fire_mqtt_message, async_fire_time_changed,
fire_mqtt_message, fire_time_changed, get_test_home_assistant, async_mock_mqtt_component, mock_registry)
mock_component, mock_mqtt_component, mock_registry)
class TestSensorMQTT(unittest.TestCase): async def test_setting_sensor_value_via_mqtt_message(hass, mqtt_mock):
"""Test the MQTT sensor.""" """Test the setting of the value via MQTT."""
assert await async_setup_component(hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF',
}
})
def setUp(self): # pylint: disable=invalid-name state = hass.states.get('binary_sensor.test')
"""Set up things to be run when tests are started.""" assert STATE_OFF == state.state
self.hass = get_test_home_assistant()
self.hass.config_entries._async_schedule_save = Mock()
mock_mqtt_component(self.hass)
def tearDown(self): # pylint: disable=invalid-name async_fire_mqtt_message(hass, 'test-topic', 'ON')
"""Stop everything that was started.""" state = hass.states.get('binary_sensor.test')
self.hass.stop() assert STATE_ON == state.state
def test_setting_sensor_value_via_mqtt_message(self): async_fire_mqtt_message(hass, 'test-topic', 'OFF')
"""Test the setting of the value via MQTT.""" state = hass.states.get('binary_sensor.test')
assert setup_component(self.hass, binary_sensor.DOMAIN, { assert STATE_OFF == state.state
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF',
}
})
state = self.hass.states.get('binary_sensor.test')
assert STATE_OFF == state.state
fire_mqtt_message(self.hass, 'test-topic', 'ON') async def test_setting_sensor_value_via_mqtt_message_and_template(
self.hass.block_till_done() hass, mqtt_mock):
state = self.hass.states.get('binary_sensor.test') """Test the setting of the value via MQTT."""
assert STATE_ON == state.state assert await async_setup_component(hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF',
'value_template': '{%if is_state(entity_id,\"on\")-%}OFF'
'{%-else-%}ON{%-endif%}'
}
})
fire_mqtt_message(self.hass, 'test-topic', 'OFF') state = hass.states.get('binary_sensor.test')
self.hass.block_till_done() assert STATE_OFF == state.state
state = self.hass.states.get('binary_sensor.test')
assert STATE_OFF == state.state
def test_setting_sensor_value_via_mqtt_message_and_template(self): async_fire_mqtt_message(hass, 'test-topic', '')
"""Test the setting of the value via MQTT.""" state = hass.states.get('binary_sensor.test')
assert setup_component(self.hass, binary_sensor.DOMAIN, { assert STATE_ON == state.state
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF',
'value_template': '{%if is_state(entity_id,\"on\")-%}OFF'
'{%-else-%}ON{%-endif%}'
}
})
state = self.hass.states.get('binary_sensor.test') async_fire_mqtt_message(hass, 'test-topic', '')
assert STATE_OFF == state.state state = hass.states.get('binary_sensor.test')
assert STATE_OFF == state.state
fire_mqtt_message(self.hass, 'test-topic', '')
self.hass.block_till_done()
state = self.hass.states.get('binary_sensor.test')
assert STATE_ON == state.state
fire_mqtt_message(self.hass, 'test-topic', '') async def test_valid_device_class(hass, mqtt_mock):
self.hass.block_till_done() """Test the setting of a valid sensor class."""
state = self.hass.states.get('binary_sensor.test') assert await async_setup_component(hass, binary_sensor.DOMAIN, {
assert STATE_OFF == state.state binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'device_class': 'motion',
'state_topic': 'test-topic',
}
})
def test_valid_device_class(self): state = hass.states.get('binary_sensor.test')
"""Test the setting of a valid sensor class.""" assert 'motion' == state.attributes.get('device_class')
assert setup_component(self.hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'device_class': 'motion',
'state_topic': 'test-topic',
}
})
state = self.hass.states.get('binary_sensor.test')
assert 'motion' == state.attributes.get('device_class')
def test_invalid_device_class(self): async def test_invalid_device_class(hass, mqtt_mock):
"""Test the setting of an invalid sensor class.""" """Test the setting of an invalid sensor class."""
assert setup_component(self.hass, binary_sensor.DOMAIN, { assert await async_setup_component(hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: { binary_sensor.DOMAIN: {
'platform': 'mqtt', 'platform': 'mqtt',
'name': 'test', 'name': 'test',
'device_class': 'abc123', 'device_class': 'abc123',
'state_topic': 'test-topic', 'state_topic': 'test-topic',
} }
}) })
state = self.hass.states.get('binary_sensor.test') state = hass.states.get('binary_sensor.test')
assert state is None assert state is None
def test_availability_without_topic(self):
"""Test availability without defined availability topic."""
assert setup_component(self.hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
}
})
state = self.hass.states.get('binary_sensor.test') async def test_availability_without_topic(hass, mqtt_mock):
assert STATE_UNAVAILABLE != state.state """Test availability without defined availability topic."""
assert await async_setup_component(hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
}
})
def test_availability_by_defaults(self): state = hass.states.get('binary_sensor.test')
"""Test availability by defaults with defined topic.""" assert STATE_UNAVAILABLE != state.state
assert setup_component(self.hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'availability_topic': 'availability-topic'
}
})
state = self.hass.states.get('binary_sensor.test')
assert STATE_UNAVAILABLE == state.state
fire_mqtt_message(self.hass, 'availability-topic', 'online') async def test_availability_by_defaults(hass, mqtt_mock):
self.hass.block_till_done() """Test availability by defaults with defined topic."""
assert await async_setup_component(hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'availability_topic': 'availability-topic'
}
})
state = self.hass.states.get('binary_sensor.test') state = hass.states.get('binary_sensor.test')
assert STATE_UNAVAILABLE != state.state assert STATE_UNAVAILABLE == state.state
fire_mqtt_message(self.hass, 'availability-topic', 'offline') async_fire_mqtt_message(hass, 'availability-topic', 'online')
self.hass.block_till_done()
state = self.hass.states.get('binary_sensor.test') state = hass.states.get('binary_sensor.test')
assert STATE_UNAVAILABLE == state.state assert STATE_UNAVAILABLE != state.state
def test_availability_by_custom_payload(self): async_fire_mqtt_message(hass, 'availability-topic', 'offline')
"""Test availability by custom payload with defined topic."""
assert setup_component(self.hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'availability_topic': 'availability-topic',
'payload_available': 'good',
'payload_not_available': 'nogood'
}
})
state = self.hass.states.get('binary_sensor.test') state = hass.states.get('binary_sensor.test')
assert STATE_UNAVAILABLE == state.state assert STATE_UNAVAILABLE == state.state
fire_mqtt_message(self.hass, 'availability-topic', 'good')
self.hass.block_till_done()
state = self.hass.states.get('binary_sensor.test') async def test_availability_by_custom_payload(hass, mqtt_mock):
assert STATE_UNAVAILABLE != state.state """Test availability by custom payload with defined topic."""
assert await async_setup_component(hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'availability_topic': 'availability-topic',
'payload_available': 'good',
'payload_not_available': 'nogood'
}
})
fire_mqtt_message(self.hass, 'availability-topic', 'nogood') state = hass.states.get('binary_sensor.test')
self.hass.block_till_done() assert STATE_UNAVAILABLE == state.state
state = self.hass.states.get('binary_sensor.test') async_fire_mqtt_message(hass, 'availability-topic', 'good')
assert STATE_UNAVAILABLE == state.state
def test_force_update_disabled(self): state = hass.states.get('binary_sensor.test')
"""Test force update option.""" assert STATE_UNAVAILABLE != state.state
mock_component(self.hass, 'mqtt')
assert setup_component(self.hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF'
}
})
events = [] async_fire_mqtt_message(hass, 'availability-topic', 'nogood')
@ha.callback state = hass.states.get('binary_sensor.test')
def callback(event): assert STATE_UNAVAILABLE == state.state
"""Verify event got called."""
events.append(event)
self.hass.bus.listen(EVENT_STATE_CHANGED, callback)
fire_mqtt_message(self.hass, 'test-topic', 'ON') async def test_force_update_disabled(hass, mqtt_mock):
self.hass.block_till_done() """Test force update option."""
assert 1 == len(events) assert await async_setup_component(hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF'
}
})
fire_mqtt_message(self.hass, 'test-topic', 'ON') events = []
self.hass.block_till_done()
assert 1 == len(events)
def test_force_update_enabled(self): @ha.callback
"""Test force update option.""" def callback(event):
mock_component(self.hass, 'mqtt') """Verify event got called."""
assert setup_component(self.hass, binary_sensor.DOMAIN, { events.append(event)
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF',
'force_update': True
}
})
events = [] hass.bus.async_listen(EVENT_STATE_CHANGED, callback)
@ha.callback async_fire_mqtt_message(hass, 'test-topic', 'ON')
def callback(event): await hass.async_block_till_done()
"""Verify event got called.""" assert 1 == len(events)
events.append(event)
self.hass.bus.listen(EVENT_STATE_CHANGED, callback) async_fire_mqtt_message(hass, 'test-topic', 'ON')
await hass.async_block_till_done()
assert 1 == len(events)
fire_mqtt_message(self.hass, 'test-topic', 'ON')
self.hass.block_till_done()
assert 1 == len(events)
fire_mqtt_message(self.hass, 'test-topic', 'ON') async def test_force_update_enabled(hass, mqtt_mock):
self.hass.block_till_done() """Test force update option."""
assert 2 == len(events) assert await async_setup_component(hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF',
'force_update': True
}
})
def test_off_delay(self): events = []
"""Test off_delay option."""
mock_component(self.hass, 'mqtt')
assert setup_component(self.hass, binary_sensor.DOMAIN, {
binary_sensor.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF',
'off_delay': 30,
'force_update': True
}
})
events = [] @ha.callback
def callback(event):
"""Verify event got called."""
events.append(event)
@ha.callback hass.bus.async_listen(EVENT_STATE_CHANGED, callback)
def callback(event):
"""Verify event got called."""
events.append(event)
self.hass.bus.listen(EVENT_STATE_CHANGED, callback) async_fire_mqtt_message(hass, 'test-topic', 'ON')
await hass.async_block_till_done()
assert 1 == len(events)
fire_mqtt_message(self.hass, 'test-topic', 'ON') async_fire_mqtt_message(hass, 'test-topic', 'ON')
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get('binary_sensor.test') assert 2 == len(events)
assert STATE_ON == state.state
assert 1 == len(events)
fire_mqtt_message(self.hass, 'test-topic', 'ON')
self.hass.block_till_done()
state = self.hass.states.get('binary_sensor.test')
assert STATE_ON == state.state
assert 2 == len(events)
fire_time_changed(self.hass, dt_util.utcnow() + timedelta(seconds=30)) async def test_off_delay(hass, mqtt_mock):
self.hass.block_till_done() """Test off_delay option."""
state = self.hass.states.get('binary_sensor.test') assert await async_setup_component(hass, binary_sensor.DOMAIN, {
assert STATE_OFF == state.state binary_sensor.DOMAIN: {
assert 3 == len(events) 'platform': 'mqtt',
'name': 'test',
'state_topic': 'test-topic',
'payload_on': 'ON',
'payload_off': 'OFF',
'off_delay': 30,
'force_update': True
}
})
events = []
@ha.callback
def callback(event):
"""Verify event got called."""
events.append(event)
hass.bus.async_listen(EVENT_STATE_CHANGED, callback)
async_fire_mqtt_message(hass, 'test-topic', 'ON')
await hass.async_block_till_done()
state = hass.states.get('binary_sensor.test')
assert STATE_ON == state.state
assert 1 == len(events)
async_fire_mqtt_message(hass, 'test-topic', 'ON')
await hass.async_block_till_done()
state = hass.states.get('binary_sensor.test')
assert STATE_ON == state.state
assert 2 == len(events)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=30))
await hass.async_block_till_done()
state = hass.states.get('binary_sensor.test')
assert STATE_OFF == state.state
assert 3 == len(events)
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock): async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
@ -294,7 +278,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('binary_sensor.test') state = hass.states.get('binary_sensor.test')
assert '100' == state.attributes.get('val') assert '100' == state.attributes.get('val')
@ -312,7 +295,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('binary_sensor.test') state = hass.states.get('binary_sensor.test')
assert state.attributes.get('val') is None assert state.attributes.get('val') is None
@ -331,7 +313,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('binary_sensor.test') state = hass.states.get('binary_sensor.test')
assert state.attributes.get('val') is None assert state.attributes.get('val') is None
@ -356,8 +337,6 @@ async def test_discovery_update_attr(hass, mqtt_mock, caplog):
data1) 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('binary_sensor.beer') state = hass.states.get('binary_sensor.beer')
assert '100' == state.attributes.get('val') assert '100' == state.attributes.get('val')
@ -365,19 +344,14 @@ async def test_discovery_update_attr(hass, mqtt_mock, caplog):
async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config', async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config',
data2) 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('binary_sensor.beer') state = hass.states.get('binary_sensor.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('binary_sensor.beer') state = hass.states.get('binary_sensor.beer')
assert '75' == state.attributes.get('val') assert '75' == state.attributes.get('val')
@ -399,7 +373,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_all()) == 1 assert len(hass.states.async_all()) == 1
@ -421,7 +394,6 @@ async def test_discovery_removal_binary_sensor(hass, mqtt_mock, caplog):
async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config', async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config',
'') '')
await hass.async_block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('binary_sensor.beer') state = hass.states.get('binary_sensor.beer')
assert state is None assert state is None
@ -449,7 +421,6 @@ async def test_discovery_update_binary_sensor(hass, mqtt_mock, caplog):
async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config', async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/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('binary_sensor.beer') state = hass.states.get('binary_sensor.beer')
assert state is not None assert state is not None
assert state.name == 'Milk' assert state.name == 'Milk'
@ -482,7 +453,6 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config', async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/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('binary_sensor.milk') state = hass.states.get('binary_sensor.milk')
assert state is not None assert state is not None
@ -517,7 +487,6 @@ async def test_entity_device_info_with_identifier(hass, mqtt_mock):
async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config', async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config',
data) 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
@ -557,7 +526,6 @@ async def test_entity_device_info_update(hass, mqtt_mock):
async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config', async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config',
data) 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
@ -568,7 +536,6 @@ async def test_entity_device_info_update(hass, mqtt_mock):
async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config', async_fire_mqtt_message(hass, 'homeassistant/binary_sensor/bla/config',
data) 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
@ -599,7 +566,6 @@ async def test_entity_id_update(hass, mqtt_mock):
registry.async_update_entity( registry.async_update_entity(
'binary_sensor.beer', new_entity_id='binary_sensor.milk') 'binary_sensor.beer', new_entity_id='binary_sensor.milk')
await hass.async_block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('binary_sensor.beer') state = hass.states.get('binary_sensor.beer')
assert state is None assert state is None