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

Merge conflicting changes (#17761)

Browse Source
This commit is contained in:
kennedyshead 2018-10-28 23:49:55 +01:00 committed by Fabian Affolter
parent b62b3b26f2
commit 3802fec568
1 changed files with 77 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

153
tests/components/fan/test_mqtt.py
View File

@ -1,110 +1,111 @@
"""Test MQTT fans.""" """Test MQTT fans."""
import json import json
import unittest
from homeassistant.setup import setup_component, async_setup_component from homeassistant.setup import async_setup_component
from homeassistant.components import fan from homeassistant.components import fan
from homeassistant.components.mqtt.discovery import async_start from homeassistant.components.mqtt.discovery import async_start
from homeassistant.const import ATTR_ASSUMED_STATE, STATE_UNAVAILABLE from homeassistant.const import ATTR_ASSUMED_STATE, STATE_UNAVAILABLE
from tests.common import ( from tests.common import async_fire_mqtt_message, MockConfigEntry, \
mock_mqtt_component, async_fire_mqtt_message, fire_mqtt_message, async_mock_mqtt_component
get_test_home_assistant, async_mock_mqtt_component, MockConfigEntry)
class TestMqttFan(unittest.TestCase): async def test_fail_setup_if_no_command_topic(hass, mqtt_mock):
"""Test the MQTT fan platform.""" """Test if command fails with command topic."""
assert await async_setup_component(hass, fan.DOMAIN, {
fan.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
}
})
assert hass.states.get('fan.test') is None
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_default_availability_payload(hass, mqtt_mock):
"""Stop everything that was started.""" """Test the availability payload."""
self.hass.stop() assert await async_setup_component(hass, fan.DOMAIN, {
fan.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'availability_topic': 'availability_topic'
}
})
def test_default_availability_payload(self): state = hass.states.get('fan.test')
"""Test the availability payload.""" assert state.state is STATE_UNAVAILABLE
assert setup_component(self.hass, fan.DOMAIN, {
fan.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'availability_topic': 'availability_topic'
}
})
state = self.hass.states.get('fan.test') async_fire_mqtt_message(hass, 'availability_topic', 'online')
assert STATE_UNAVAILABLE == state.state await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'availability_topic', 'online') state = hass.states.get('fan.test')
self.hass.block_till_done() assert state.state is not STATE_UNAVAILABLE
assert not state.attributes.get(ATTR_ASSUMED_STATE)
state = self.hass.states.get('fan.test') async_fire_mqtt_message(hass, 'availability_topic', 'offline')
assert STATE_UNAVAILABLE != 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', 'offline') state = hass.states.get('fan.test')
self.hass.block_till_done() assert state.state is STATE_UNAVAILABLE
state = self.hass.states.get('fan.test') async_fire_mqtt_message(hass, 'state-topic', '1')
assert STATE_UNAVAILABLE == state.state await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'state-topic', '1') state = hass.states.get('fan.test')
self.hass.block_till_done() assert state.state is STATE_UNAVAILABLE
state = self.hass.states.get('fan.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', 'online') state = hass.states.get('fan.test')
self.hass.block_till_done() assert state.state is not STATE_UNAVAILABLE
state = self.hass.states.get('fan.test')
assert STATE_UNAVAILABLE != state.state
def test_custom_availability_payload(self): async def test_custom_availability_payload(hass, mqtt_mock):
"""Test the availability payload.""" """Test the availability payload."""
assert setup_component(self.hass, fan.DOMAIN, { assert await async_setup_component(hass, fan.DOMAIN, {
fan.DOMAIN: { fan.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',
'availability_topic': 'availability_topic', 'availability_topic': 'availability_topic',
'payload_available': 'good', 'payload_available': 'good',
'payload_not_available': 'nogood' 'payload_not_available': 'nogood'
} }
}) })
state = self.hass.states.get('fan.test') state = hass.states.get('fan.test')
assert STATE_UNAVAILABLE == state.state assert state.state is STATE_UNAVAILABLE
fire_mqtt_message(self.hass, 'availability_topic', 'good') async_fire_mqtt_message(hass, 'availability_topic', 'good')
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get('fan.test') state = hass.states.get('fan.test')
assert STATE_UNAVAILABLE != state.state assert state.state is not STATE_UNAVAILABLE
assert not state.attributes.get(ATTR_ASSUMED_STATE) assert not state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'availability_topic', 'nogood') 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('fan.test') state = hass.states.get('fan.test')
assert STATE_UNAVAILABLE == state.state assert state.state is STATE_UNAVAILABLE
fire_mqtt_message(self.hass, 'state-topic', '1') async_fire_mqtt_message(hass, 'state-topic', '1')
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get('fan.test') state = hass.states.get('fan.test')
assert STATE_UNAVAILABLE == state.state assert state.state is STATE_UNAVAILABLE
fire_mqtt_message(self.hass, 'availability_topic', 'good') async_fire_mqtt_message(hass, 'availability_topic', 'good')
self.hass.block_till_done() await hass.async_block_till_done()
state = self.hass.states.get('fan.test') state = hass.states.get('fan.test')
assert STATE_UNAVAILABLE != state.state assert state.state is not STATE_UNAVAILABLE
async def test_discovery_removal_fan(hass, mqtt_mock, caplog): async def test_discovery_removal_fan(hass, mqtt_mock, caplog):