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

Testing async in MQTT_json lights (#17768)

Browse Source
This commit is contained in:
kennedyshead 2018-10-25 09:54:53 +02:00 committed by Paulus Schoutsen
parent 5024a80d61
commit 8bebfba21a
1 changed files with 406 additions and 550 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

956
tests/components/light/test_mqtt_json.py
View File

@ -87,12 +87,9 @@ light:
brightness: true brightness: true
brightness_scale: 99 brightness_scale: 99
""" """
import json
import unittest
from unittest.mock import patch from unittest.mock import patch
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from homeassistant.const import ( from homeassistant.const import (
STATE_ON, STATE_OFF, STATE_UNAVAILABLE, ATTR_ASSUMED_STATE, STATE_ON, STATE_OFF, STATE_UNAVAILABLE, ATTR_ASSUMED_STATE,
ATTR_SUPPORTED_FEATURES) ATTR_SUPPORTED_FEATURES)
@ -100,570 +97,429 @@ import homeassistant.components.light as light
from homeassistant.components.mqtt.discovery import async_start from homeassistant.components.mqtt.discovery import async_start
import homeassistant.core as ha import homeassistant.core as ha
from tests.common import ( from tests.common import mock_coro, async_fire_mqtt_message
get_test_home_assistant, mock_mqtt_component, fire_mqtt_message,
assert_setup_component, mock_coro, async_fire_mqtt_message)
from tests.components.light import common
class TestLightMQTTJSON(unittest.TestCase): async def test_fail_setup_if_no_command_topic(hass, mqtt_mock):
"""Test the MQTT JSON light.""" """Test if setup fails with no command topic."""
assert await async_setup_component(hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
}
})
assert hass.states.get('light.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_no_color_brightness_color_temp_white_val_if_no_topics(
"""Stop everything that was started.""" hass, mqtt_mock):
self.hass.stop() """Test for no RGB, brightness, color temp, effect, white val or XY."""
assert await async_setup_component(hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set',
}
})
def test_fail_setup_if_no_command_topic(self): state = hass.states.get('light.test')
"""Test if setup fails with no command topic.""" assert STATE_OFF == state.state
with assert_setup_component(0, light.DOMAIN): assert 40 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
assert setup_component(self.hass, light.DOMAIN, { assert state.attributes.get('rgb_color') is None
light.DOMAIN: { assert state.attributes.get('brightness') is None
'platform': 'mqtt_json', assert state.attributes.get('color_temp') is None
'name': 'test', assert state.attributes.get('effect') is None
} assert state.attributes.get('white_value') is None
}) assert state.attributes.get('xy_color') is None
assert self.hass.states.get('light.test') is None assert state.attributes.get('hs_color') is None
def test_no_color_brightness_color_temp_white_val_if_no_topics(self): async_fire_mqtt_message(hass, 'test_light_rgb', '{"state":"ON"}')
"""Test for no RGB, brightness, color temp, effect, white val or XY.""" await hass.async_block_till_done()
assert setup_component(self.hass, light.DOMAIN, {
state = hass.states.get('light.test')
assert STATE_ON == state.state
assert state.attributes.get('rgb_color') is None
assert state.attributes.get('brightness') is None
assert state.attributes.get('color_temp') is None
assert state.attributes.get('effect') is None
assert state.attributes.get('white_value') is None
assert state.attributes.get('xy_color') is None
assert state.attributes.get('hs_color') is None
async def test_controlling_state_via_topic(hass, mqtt_mock):
"""Test the controlling of the state via topic."""
assert await async_setup_component(hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set',
'brightness': True,
'color_temp': True,
'effect': True,
'rgb': True,
'white_value': True,
'xy': True,
'hs': True,
'qos': '0'
}
})
state = hass.states.get('light.test')
assert STATE_OFF == state.state
assert 191 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
assert state.attributes.get('rgb_color') is None
assert state.attributes.get('brightness') is None
assert state.attributes.get('color_temp') is None
assert state.attributes.get('effect') is None
assert state.attributes.get('white_value') is None
assert state.attributes.get('xy_color') is None
assert state.attributes.get('hs_color') is None
assert not state.attributes.get(ATTR_ASSUMED_STATE)
# Turn on the light, full white
async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON",'
'"color":{"r":255,"g":255,"b":255},'
'"brightness":255,'
'"color_temp":155,'
'"effect":"colorloop",'
'"white_value":150}')
await hass.async_block_till_done()
state = hass.states.get('light.test')
assert STATE_ON == state.state
assert (255, 255, 255) == state.attributes.get('rgb_color')
assert 255 == state.attributes.get('brightness')
assert 155 == state.attributes.get('color_temp')
assert 'colorloop' == state.attributes.get('effect')
assert 150 == state.attributes.get('white_value')
assert (0.323, 0.329) == state.attributes.get('xy_color')
assert (0.0, 0.0) == state.attributes.get('hs_color')
# Turn the light off
async_fire_mqtt_message(hass, 'test_light_rgb', '{"state":"OFF"}')
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get('light.test')
assert STATE_OFF == state.state
async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON", "brightness":100}')
await hass.async_block_till_done()
await hass.async_block_till_done()
light_state = hass.states.get('light.test')
assert 100 == \
light_state.attributes['brightness']
async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON", '
'"color":{"r":125,"g":125,"b":125}}')
await hass.async_block_till_done()
light_state = hass.states.get('light.test')
assert (255, 255, 255) == \
light_state.attributes.get('rgb_color')
async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON", "color":{"x":0.135,"y":0.135}}')
await hass.async_block_till_done()
light_state = hass.states.get('light.test')
assert (0.141, 0.14) == \
light_state.attributes.get('xy_color')
async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON", "color":{"h":180,"s":50}}')
await hass.async_block_till_done()
await hass.async_block_till_done()
light_state = hass.states.get('light.test')
assert (180.0, 50.0) == \
light_state.attributes.get('hs_color')
async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON", "color_temp":155}')
await hass.async_block_till_done()
await hass.async_block_till_done()
light_state = hass.states.get('light.test')
assert 155 == light_state.attributes.get('color_temp')
async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON", "effect":"colorloop"}')
await hass.async_block_till_done()
await hass.async_block_till_done()
light_state = hass.states.get('light.test')
assert 'colorloop' == light_state.attributes.get('effect')
async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON", "white_value":155}')
await hass.async_block_till_done()
await hass.async_block_till_done()
light_state = hass.states.get('light.test')
assert 155 == light_state.attributes.get('white_value')
async def test_sending_mqtt_commands_and_optimistic(hass, mqtt_mock):
"""Test the sending of command in optimistic mode."""
fake_state = ha.State('light.test', 'on', {'brightness': 95,
'hs_color': [100, 100],
'effect': 'random',
'color_temp': 100,
'white_value': 50})
with patch('homeassistant.components.light.mqtt_json'
'.async_get_last_state',
return_value=mock_coro(fake_state)):
assert await async_setup_component(hass, light.DOMAIN, {
light.DOMAIN: { light.DOMAIN: {
'platform': 'mqtt_json', 'platform': 'mqtt_json',
'name': 'test', 'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set',
}
})
state = self.hass.states.get('light.test')
assert STATE_OFF == state.state
assert 40 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
assert state.attributes.get('rgb_color') is None
assert state.attributes.get('brightness') is None
assert state.attributes.get('color_temp') is None
assert state.attributes.get('effect') is None
assert state.attributes.get('white_value') is None
assert state.attributes.get('xy_color') is None
assert state.attributes.get('hs_color') is None
fire_mqtt_message(self.hass, 'test_light_rgb', '{"state":"ON"}')
self.hass.block_till_done()
state = self.hass.states.get('light.test')
assert STATE_ON == state.state
assert state.attributes.get('rgb_color') is None
assert state.attributes.get('brightness') is None
assert state.attributes.get('color_temp') is None
assert state.attributes.get('effect') is None
assert state.attributes.get('white_value') is None
assert state.attributes.get('xy_color') is None
assert state.attributes.get('hs_color') is None
def test_controlling_state_via_topic(self):
"""Test the controlling of the state via topic."""
assert setup_component(self.hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set', 'command_topic': 'test_light_rgb/set',
'brightness': True, 'brightness': True,
'color_temp': True, 'color_temp': True,
'effect': True, 'effect': True,
'rgb': True, 'rgb': True,
'white_value': True, 'white_value': True,
'xy': True, 'qos': 2
'hs': True,
'qos': '0'
} }
}) })
state = self.hass.states.get('light.test') state = hass.states.get('light.test')
assert STATE_OFF == state.state assert STATE_ON == state.state
assert 191 == state.attributes.get(ATTR_SUPPORTED_FEATURES) assert 95 == state.attributes.get('brightness')
assert state.attributes.get('rgb_color') is None assert (100, 100) == state.attributes.get('hs_color')
assert state.attributes.get('brightness') is None assert 'random' == state.attributes.get('effect')
assert state.attributes.get('color_temp') is None assert 100 == state.attributes.get('color_temp')
assert state.attributes.get('effect') is None assert 50 == state.attributes.get('white_value')
assert state.attributes.get('white_value') is None assert 191 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
assert state.attributes.get('xy_color') is None assert state.attributes.get(ATTR_ASSUMED_STATE)
assert state.attributes.get('hs_color') is None
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async def test_sending_hs_color(hass, mqtt_mock):
# Turn on the light, full white """Test light.turn_on with hs color sends hs color parameters."""
fire_mqtt_message(self.hass, 'test_light_rgb', assert await async_setup_component(hass, light.DOMAIN, {
'{"state":"ON",' light.DOMAIN: {
'"color":{"r":255,"g":255,"b":255},' 'platform': 'mqtt_json',
'"brightness":255,' 'name': 'test',
'"color_temp":155,' 'command_topic': 'test_light_rgb/set',
'"effect":"colorloop",' 'hs': True,
'"white_value":150}') }
self.hass.block_till_done() })
state = self.hass.states.get('light.test') state = hass.states.get('light.test')
assert STATE_ON == state.state assert STATE_OFF == state.state
assert (255, 255, 255) == state.attributes.get('rgb_color')
assert 255 == state.attributes.get('brightness')
assert 155 == state.attributes.get('color_temp') async def test_flash_short_and_long(hass, mqtt_mock):
assert 'colorloop' == state.attributes.get('effect') """Test for flash length being sent when included."""
assert 150 == state.attributes.get('white_value') assert await async_setup_component(hass, light.DOMAIN, {
assert (0.323, 0.329) == state.attributes.get('xy_color') light.DOMAIN: {
assert (0.0, 0.0) == state.attributes.get('hs_color') 'platform': 'mqtt_json',
'name': 'test',
# Turn the light off 'state_topic': 'test_light_rgb',
fire_mqtt_message(self.hass, 'test_light_rgb', '{"state":"OFF"}') 'command_topic': 'test_light_rgb/set',
self.hass.block_till_done() 'flash_time_short': 5,
'flash_time_long': 15,
state = self.hass.states.get('light.test') 'qos': 0
assert STATE_OFF == state.state }
})
fire_mqtt_message(self.hass, 'test_light_rgb',
'{"state":"ON",' state = hass.states.get('light.test')
'"brightness":100}') assert STATE_OFF == state.state
self.hass.block_till_done() assert 40 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
light_state = self.hass.states.get('light.test')
self.hass.block_till_done() async def test_transition(hass, mqtt_mock):
assert 100 == \ """Test for transition time being sent when included."""
light_state.attributes['brightness'] assert await async_setup_component(hass, light.DOMAIN, {
light.DOMAIN: {
fire_mqtt_message(self.hass, 'test_light_rgb', 'platform': 'mqtt_json',
'{"state":"ON",' 'name': 'test',
'"color":{"r":125,"g":125,"b":125}}') 'state_topic': 'test_light_rgb',
self.hass.block_till_done() 'command_topic': 'test_light_rgb/set',
'qos': 0
light_state = self.hass.states.get('light.test') }
assert (255, 255, 255) == \ })
light_state.attributes.get('rgb_color')
state = hass.states.get('light.test')
fire_mqtt_message(self.hass, 'test_light_rgb', assert STATE_OFF == state.state
'{"state":"ON",' assert 40 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
'"color":{"x":0.135,"y":0.135}}')
self.hass.block_till_done()
async def test_brightness_scale(hass, mqtt_mock):
light_state = self.hass.states.get('light.test') """Test for brightness scaling."""
assert (0.141, 0.14) == \ assert await async_setup_component(hass, light.DOMAIN, {
light_state.attributes.get('xy_color') light.DOMAIN: {
'platform': 'mqtt_json',
fire_mqtt_message(self.hass, 'test_light_rgb', 'name': 'test',
'{"state":"ON",' 'state_topic': 'test_light_bright_scale',
'"color":{"h":180,"s":50}}') 'command_topic': 'test_light_bright_scale/set',
self.hass.block_till_done() 'brightness': True,
'brightness_scale': 99
light_state = self.hass.states.get('light.test') }
assert (180.0, 50.0) == \ })
light_state.attributes.get('hs_color')
state = hass.states.get('light.test')
fire_mqtt_message(self.hass, 'test_light_rgb', assert STATE_OFF == state.state
'{"state":"ON",' assert state.attributes.get('brightness') is None
'"color_temp":155}') assert not state.attributes.get(ATTR_ASSUMED_STATE)
self.hass.block_till_done()
# Turn on the light
light_state = self.hass.states.get('light.test') async_fire_mqtt_message(hass, 'test_light_bright_scale', '{"state":"ON"}')
assert 155 == light_state.attributes.get('color_temp') await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'test_light_rgb', state = hass.states.get('light.test')
'{"state":"ON",' assert STATE_ON == state.state
'"effect":"colorloop"}') assert 255 == state.attributes.get('brightness')
self.hass.block_till_done()
# Turn on the light with brightness
light_state = self.hass.states.get('light.test') async_fire_mqtt_message(hass, 'test_light_bright_scale',
assert 'colorloop' == light_state.attributes.get('effect') '{"state":"ON", "brightness": 99}')
await hass.async_block_till_done()
fire_mqtt_message(self.hass, 'test_light_rgb',
'{"state":"ON",' state = hass.states.get('light.test')
'"white_value":155}') assert STATE_ON == state.state
self.hass.block_till_done() assert 255 == state.attributes.get('brightness')
light_state = self.hass.states.get('light.test')
assert 155 == light_state.attributes.get('white_value') async def test_invalid_color_brightness_and_white_values(hass, mqtt_mock):
"""Test that invalid color/brightness/white values are ignored."""
def test_sending_mqtt_commands_and_optimistic(self): assert await async_setup_component(hass, light.DOMAIN, {
"""Test the sending of command in optimistic mode.""" light.DOMAIN: {
fake_state = ha.State('light.test', 'on', {'brightness': 95, 'platform': 'mqtt_json',
'hs_color': [100, 100], 'name': 'test',
'effect': 'random', 'state_topic': 'test_light_rgb',
'color_temp': 100, 'command_topic': 'test_light_rgb/set',
'white_value': 50}) 'brightness': True,
'rgb': True,
with patch('homeassistant.components.light.mqtt_json' 'white_value': True,
'.async_get_last_state', 'qos': '0'
return_value=mock_coro(fake_state)): }
assert setup_component(self.hass, light.DOMAIN, { })
light.DOMAIN: {
'platform': 'mqtt_json', state = hass.states.get('light.test')
'name': 'test', assert STATE_OFF == state.state
'command_topic': 'test_light_rgb/set', assert 185 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
'brightness': True, assert state.attributes.get('rgb_color') is None
'color_temp': True, assert state.attributes.get('brightness') is None
'effect': True, assert state.attributes.get('white_value') is None
'rgb': True, assert not state.attributes.get(ATTR_ASSUMED_STATE)
'white_value': True,
'qos': 2 # Turn on the light
} async_fire_mqtt_message(hass, 'test_light_rgb',
}) '{"state":"ON",'
'"color":{"r":255,"g":255,"b":255},'
state = self.hass.states.get('light.test') '"brightness": 255,'
assert STATE_ON == state.state '"white_value": 255}')
assert 95 == state.attributes.get('brightness') await hass.async_block_till_done()
assert (100, 100) == state.attributes.get('hs_color')
assert 'random' == state.attributes.get('effect') state = hass.states.get('light.test')
assert 100 == state.attributes.get('color_temp') assert STATE_ON == state.state
assert 50 == state.attributes.get('white_value') assert (255, 255, 255) == state.attributes.get('rgb_color')
assert 191 == state.attributes.get(ATTR_SUPPORTED_FEATURES) assert 255 == state.attributes.get('brightness')
assert state.attributes.get(ATTR_ASSUMED_STATE) assert 255 == state.attributes.get('white_value')
common.turn_on(self.hass, 'light.test') # Bad color values
self.hass.block_till_done() async_fire_mqtt_message(hass, 'test_light_rgb',
'{"state":"ON",'
self.mock_publish.async_publish.assert_called_once_with( '"color":{"r":"bad","g":"val","b":"test"}}')
'test_light_rgb/set', '{"state": "ON"}', 2, False) await hass.async_block_till_done()
self.mock_publish.async_publish.reset_mock()
state = self.hass.states.get('light.test') # Color should not have changed
assert STATE_ON == state.state state = hass.states.get('light.test')
assert STATE_ON == state.state
common.turn_off(self.hass, 'light.test') assert (255, 255, 255) == state.attributes.get('rgb_color')
self.hass.block_till_done()
# Bad brightness values
self.mock_publish.async_publish.assert_called_once_with( async_fire_mqtt_message(hass, 'test_light_rgb',
'test_light_rgb/set', '{"state": "OFF"}', 2, False) '{"state":"ON",'
self.mock_publish.async_publish.reset_mock() '"brightness": "badValue"}')
state = self.hass.states.get('light.test') await hass.async_block_till_done()
assert STATE_OFF == state.state
# Brightness should not have changed
common.turn_on(self.hass, 'light.test', state = hass.states.get('light.test')
brightness=50, color_temp=155, effect='colorloop', assert STATE_ON == state.state
white_value=170) assert 255 == state.attributes.get('brightness')
self.hass.block_till_done()
# Bad white value
assert 'test_light_rgb/set' == \ async_fire_mqtt_message(hass, 'test_light_rgb',
self.mock_publish.async_publish.mock_calls[0][1][0] '{"state":"ON",'
assert 2 == \ '"white_value": "badValue"}')
self.mock_publish.async_publish.mock_calls[0][1][2] await hass.async_block_till_done()
assert self.mock_publish.async_publish.mock_calls[0][1][3] is False
# Get the sent message # White value should not have changed
message_json = json.loads( state = hass.states.get('light.test')
self.mock_publish.async_publish.mock_calls[0][1][1]) assert STATE_ON == state.state
assert 50 == message_json["brightness"] assert 255 == state.attributes.get('white_value')
assert 155 == message_json["color_temp"]
assert 'colorloop' == message_json["effect"]
assert 170 == message_json["white_value"] async def test_default_availability_payload(hass, mqtt_mock):
assert "ON" == message_json["state"] """Test availability by default payload with defined topic."""
assert await async_setup_component(hass, light.DOMAIN, {
state = self.hass.states.get('light.test') light.DOMAIN: {
assert STATE_ON == state.state 'platform': 'mqtt_json',
assert 50 == state.attributes['brightness'] 'name': 'test',
assert 155 == state.attributes['color_temp'] 'state_topic': 'test_light_rgb',
assert 'colorloop' == state.attributes['effect'] 'command_topic': 'test_light_rgb/set',
assert 170 == state.attributes['white_value'] 'availability_topic': 'availability-topic'
}
# Test a color command })
common.turn_on(self.hass, 'light.test',
brightness=50, hs_color=(125, 100)) state = hass.states.get('light.test')
self.hass.block_till_done() assert STATE_UNAVAILABLE == state.state
assert 'test_light_rgb/set' == \ async_fire_mqtt_message(hass, 'availability-topic', 'online')
self.mock_publish.async_publish.mock_calls[0][1][0] await hass.async_block_till_done()
assert 2 == \
self.mock_publish.async_publish.mock_calls[0][1][2] state = hass.states.get('light.test')
assert self.mock_publish.async_publish.mock_calls[0][1][3] is False assert STATE_UNAVAILABLE != state.state
# Get the sent message
message_json = json.loads( async_fire_mqtt_message(hass, 'availability-topic', 'offline')
self.mock_publish.async_publish.mock_calls[1][1][1]) await hass.async_block_till_done()
assert 50 == message_json["brightness"] await hass.async_block_till_done()
assert {
'r': 0, state = hass.states.get('light.test')
'g': 255, assert STATE_UNAVAILABLE == state.state
'b': 21,
} == message_json["color"]
assert "ON" == message_json["state"] async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
state = self.hass.states.get('light.test') assert await async_setup_component(hass, light.DOMAIN, {
assert STATE_ON == state.state light.DOMAIN: {
assert 50 == state.attributes['brightness'] 'platform': 'mqtt_json',
assert (125, 100) == state.attributes['hs_color'] 'name': 'test',
'state_topic': 'test_light_rgb',
def test_sending_hs_color(self): 'command_topic': 'test_light_rgb/set',
"""Test light.turn_on with hs color sends hs color parameters.""" 'availability_topic': 'availability-topic',
assert setup_component(self.hass, light.DOMAIN, { 'payload_available': 'good',
light.DOMAIN: { 'payload_not_available': 'nogood'
'platform': 'mqtt_json', }
'name': 'test', })
'command_topic': 'test_light_rgb/set',
'hs': True, state = hass.states.get('light.test')
} assert STATE_UNAVAILABLE == state.state
})
async_fire_mqtt_message(hass, 'availability-topic', 'good')
common.turn_on(self.hass, 'light.test', hs_color=(180.0, 50.0)) await hass.async_block_till_done()
self.hass.block_till_done()
state = hass.states.get('light.test')
message_json = json.loads( assert STATE_UNAVAILABLE != state.state
self.mock_publish.async_publish.mock_calls[0][1][1])
assert "ON" == message_json["state"] async_fire_mqtt_message(hass, 'availability-topic', 'nogood')
assert { await hass.async_block_till_done()
'h': 180.0, await hass.async_block_till_done()
's': 50.0,
} == message_json["color"] state = hass.states.get('light.test')
assert STATE_UNAVAILABLE == state.state
def test_flash_short_and_long(self):
"""Test for flash length being sent when included."""
assert setup_component(self.hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set',
'flash_time_short': 5,
'flash_time_long': 15,
'qos': 0
}
})
state = self.hass.states.get('light.test')
assert STATE_OFF == state.state
assert 40 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
common.turn_on(self.hass, 'light.test', flash="short")
self.hass.block_till_done()
assert 'test_light_rgb/set' == \
self.mock_publish.async_publish.mock_calls[0][1][0]
assert 0 == \
self.mock_publish.async_publish.mock_calls[0][1][2]
assert self.mock_publish.async_publish.mock_calls[0][1][3] is False
# Get the sent message
message_json = json.loads(
self.mock_publish.async_publish.mock_calls[0][1][1])
assert 5 == message_json["flash"]
assert "ON" == message_json["state"]
self.mock_publish.async_publish.reset_mock()
common.turn_on(self.hass, 'light.test', flash="long")
self.hass.block_till_done()
assert 'test_light_rgb/set' == \
self.mock_publish.async_publish.mock_calls[0][1][0]
assert 0 == \
self.mock_publish.async_publish.mock_calls[0][1][2]
assert self.mock_publish.async_publish.mock_calls[0][1][3] is False
# Get the sent message
message_json = json.loads(
self.mock_publish.async_publish.mock_calls[0][1][1])
assert 15 == message_json["flash"]
assert "ON" == message_json["state"]
def test_transition(self):
"""Test for transition time being sent when included."""
assert setup_component(self.hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set',
'qos': 0
}
})
state = self.hass.states.get('light.test')
assert STATE_OFF == state.state
assert 40 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
common.turn_on(self.hass, 'light.test', transition=10)
self.hass.block_till_done()
assert 'test_light_rgb/set' == \
self.mock_publish.async_publish.mock_calls[0][1][0]
assert 0 == \
self.mock_publish.async_publish.mock_calls[0][1][2]
assert self.mock_publish.async_publish.mock_calls[0][1][3] is False
# Get the sent message
message_json = json.loads(
self.mock_publish.async_publish.mock_calls[0][1][1])
assert 10 == message_json["transition"]
assert "ON" == message_json["state"]
# Transition back off
common.turn_off(self.hass, 'light.test', transition=10)
self.hass.block_till_done()
assert 'test_light_rgb/set' == \
self.mock_publish.async_publish.mock_calls[1][1][0]
assert 0 == \
self.mock_publish.async_publish.mock_calls[1][1][2]
assert self.mock_publish.async_publish.mock_calls[1][1][3] is False
# Get the sent message
message_json = json.loads(
self.mock_publish.async_publish.mock_calls[1][1][1])
assert 10 == message_json["transition"]
assert "OFF" == message_json["state"]
def test_brightness_scale(self):
"""Test for brightness scaling."""
assert setup_component(self.hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_bright_scale',
'command_topic': 'test_light_bright_scale/set',
'brightness': True,
'brightness_scale': 99
}
})
state = self.hass.states.get('light.test')
assert STATE_OFF == state.state
assert state.attributes.get('brightness') is None
assert not state.attributes.get(ATTR_ASSUMED_STATE)
# Turn on the light
fire_mqtt_message(self.hass, 'test_light_bright_scale',
'{"state":"ON"}')
self.hass.block_till_done()
state = self.hass.states.get('light.test')
assert STATE_ON == state.state
assert 255 == state.attributes.get('brightness')
# Turn on the light with brightness
fire_mqtt_message(self.hass, 'test_light_bright_scale',
'{"state":"ON",'
'"brightness": 99}')
self.hass.block_till_done()
state = self.hass.states.get('light.test')
assert STATE_ON == state.state
assert 255 == state.attributes.get('brightness')
def test_invalid_color_brightness_and_white_values(self):
"""Test that invalid color/brightness/white values are ignored."""
assert setup_component(self.hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set',
'brightness': True,
'rgb': True,
'white_value': True,
'qos': '0'
}
})
state = self.hass.states.get('light.test')
assert STATE_OFF == state.state
assert 185 == state.attributes.get(ATTR_SUPPORTED_FEATURES)
assert state.attributes.get('rgb_color') is None
assert state.attributes.get('brightness') is None
assert state.attributes.get('white_value') is None
assert not state.attributes.get(ATTR_ASSUMED_STATE)
# Turn on the light
fire_mqtt_message(self.hass, 'test_light_rgb',
'{"state":"ON",'
'"color":{"r":255,"g":255,"b":255},'
'"brightness": 255,'
'"white_value": 255}')
self.hass.block_till_done()
state = self.hass.states.get('light.test')
assert STATE_ON == state.state
assert (255, 255, 255) == state.attributes.get('rgb_color')
assert 255 == state.attributes.get('brightness')
assert 255 == state.attributes.get('white_value')
# Bad color values
fire_mqtt_message(self.hass, 'test_light_rgb',
'{"state":"ON",'
'"color":{"r":"bad","g":"val","b":"test"}}')
self.hass.block_till_done()
# Color should not have changed
state = self.hass.states.get('light.test')
assert STATE_ON == state.state
assert (255, 255, 255) == state.attributes.get('rgb_color')
# Bad brightness values
fire_mqtt_message(self.hass, 'test_light_rgb',
'{"state":"ON",'
'"brightness": "badValue"}')
self.hass.block_till_done()
# Brightness should not have changed
state = self.hass.states.get('light.test')
assert STATE_ON == state.state
assert 255 == state.attributes.get('brightness')
# Bad white value
fire_mqtt_message(self.hass, 'test_light_rgb',
'{"state":"ON",'
'"white_value": "badValue"}')
self.hass.block_till_done()
# White value should not have changed
state = self.hass.states.get('light.test')
assert STATE_ON == state.state
assert 255 == state.attributes.get('white_value')
def test_default_availability_payload(self):
"""Test availability by default payload with defined topic."""
assert setup_component(self.hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set',
'availability_topic': 'availability-topic'
}
})
state = self.hass.states.get('light.test')
assert STATE_UNAVAILABLE == state.state
fire_mqtt_message(self.hass, 'availability-topic', 'online')
self.hass.block_till_done()
state = self.hass.states.get('light.test')
assert STATE_UNAVAILABLE != state.state
fire_mqtt_message(self.hass, 'availability-topic', 'offline')
self.hass.block_till_done()
state = self.hass.states.get('light.test')
assert STATE_UNAVAILABLE == state.state
def test_custom_availability_payload(self):
"""Test availability by custom payload with defined topic."""
assert setup_component(self.hass, light.DOMAIN, {
light.DOMAIN: {
'platform': 'mqtt_json',
'name': 'test',
'state_topic': 'test_light_rgb',
'command_topic': 'test_light_rgb/set',
'availability_topic': 'availability-topic',
'payload_available': 'good',
'payload_not_available': 'nogood'
}
})
state = self.hass.states.get('light.test')
assert STATE_UNAVAILABLE == state.state
fire_mqtt_message(self.hass, 'availability-topic', 'good')
self.hass.block_till_done()
state = self.hass.states.get('light.test')
assert STATE_UNAVAILABLE != state.state
fire_mqtt_message(self.hass, 'availability-topic', 'nogood')
self.hass.block_till_done()
state = self.hass.states.get('light.test')
assert STATE_UNAVAILABLE == state.state
async def test_discovery_removal(hass, mqtt_mock, caplog): async def test_discovery_removal(hass, mqtt_mock, caplog):