Add support for new mqtt_json light platform. (#2777)
* Add support for new mqtt_json light platform. * Fix W503 errors. * Bring in feedback from @balloob. * Add test coverage for invalid color and brightness data. * Add coverage for transition in turn_off.pull/2779/head
Corban Mailloux
Paulus Schoutsen
parent
aadf6a7750
commit
9c851790dc
|
|
@ -0,0 +1,233 @@
|
|||
"""
|
||||
Support for MQTT JSON lights.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/light.mqtt_json/
|
||||
"""
|
||||
|
||||
import logging
|
||||
import json
|
||||
import voluptuous as vol
|
||||
|
||||
import homeassistant.components.mqtt as mqtt
|
||||
from homeassistant.components.light import (
|
||||
ATTR_BRIGHTNESS, ATTR_RGB_COLOR, ATTR_TRANSITION,
|
||||
ATTR_FLASH, FLASH_LONG, FLASH_SHORT, Light)
|
||||
from homeassistant.const import CONF_NAME, CONF_OPTIMISTIC, CONF_PLATFORM
|
||||
from homeassistant.components.mqtt import (
|
||||
CONF_STATE_TOPIC, CONF_COMMAND_TOPIC, CONF_QOS, CONF_RETAIN)
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DOMAIN = "mqtt_json"
|
||||
|
||||
DEPENDENCIES = ["mqtt"]
|
||||
|
||||
DEFAULT_NAME = "MQTT JSON Light"
|
||||
DEFAULT_OPTIMISTIC = False
|
||||
DEFAULT_BRIGHTNESS = False
|
||||
DEFAULT_RGB = False
|
||||
DEFAULT_FLASH_TIME_SHORT = 2
|
||||
DEFAULT_FLASH_TIME_LONG = 10
|
||||
|
||||
CONF_BRIGHTNESS = "brightness"
|
||||
CONF_RGB = "rgb"
|
||||
CONF_FLASH_TIME_SHORT = "flash_time_short"
|
||||
CONF_FLASH_TIME_LONG = "flash_time_long"
|
||||
|
||||
# Stealing some of these from the base MQTT configs.
|
||||
PLATFORM_SCHEMA = vol.Schema({
|
||||
vol.Required(CONF_PLATFORM): DOMAIN,
|
||||
vol.Optional(CONF_QOS, default=mqtt.DEFAULT_QOS):
|
||||
vol.All(vol.Coerce(int), vol.In([0, 1, 2])),
|
||||
vol.Required(CONF_COMMAND_TOPIC): mqtt.valid_publish_topic,
|
||||
vol.Optional(CONF_RETAIN, default=mqtt.DEFAULT_RETAIN): cv.boolean,
|
||||
vol.Optional(CONF_STATE_TOPIC): mqtt.valid_subscribe_topic,
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(CONF_OPTIMISTIC, default=DEFAULT_OPTIMISTIC): cv.boolean,
|
||||
vol.Optional(CONF_BRIGHTNESS, default=DEFAULT_BRIGHTNESS): cv.boolean,
|
||||
vol.Optional(CONF_RGB, default=DEFAULT_RGB): cv.boolean,
|
||||
vol.Optional(CONF_FLASH_TIME_SHORT, default=DEFAULT_FLASH_TIME_SHORT):
|
||||
cv.positive_int,
|
||||
vol.Optional(CONF_FLASH_TIME_LONG, default=DEFAULT_FLASH_TIME_LONG):
|
||||
cv.positive_int
|
||||
})
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices_callback, discovery_info=None):
|
||||
"""Add MQTT JSON Light."""
|
||||
add_devices_callback([MqttJson(
|
||||
hass,
|
||||
config[CONF_NAME],
|
||||
{
|
||||
key: config.get(key) for key in (
|
||||
CONF_STATE_TOPIC,
|
||||
CONF_COMMAND_TOPIC
|
||||
)
|
||||
},
|
||||
config[CONF_QOS],
|
||||
config[CONF_RETAIN],
|
||||
config[CONF_OPTIMISTIC],
|
||||
config[CONF_BRIGHTNESS],
|
||||
config[CONF_RGB],
|
||||
{
|
||||
key: config.get(key) for key in (
|
||||
CONF_FLASH_TIME_SHORT,
|
||||
CONF_FLASH_TIME_LONG
|
||||
)
|
||||
}
|
||||
)])
|
||||
|
||||
|
||||
class MqttJson(Light):
|
||||
"""MQTT JSON light."""
|
||||
|
||||
# pylint: disable=too-many-arguments,too-many-instance-attributes
|
||||
def __init__(self, hass, name, topic, qos, retain,
|
||||
optimistic, brightness, rgb, flash_times):
|
||||
"""Initialize MQTT JSON light."""
|
||||
self._hass = hass
|
||||
self._name = name
|
||||
self._topic = topic
|
||||
self._qos = qos
|
||||
self._retain = retain
|
||||
self._optimistic = optimistic or topic["state_topic"] is None
|
||||
self._state = False
|
||||
if brightness:
|
||||
self._brightness = 255
|
||||
else:
|
||||
self._brightness = None
|
||||
|
||||
if rgb:
|
||||
self._rgb = [0, 0, 0]
|
||||
else:
|
||||
self._rgb = None
|
||||
|
||||
self._flash_times = flash_times
|
||||
|
||||
def state_received(topic, payload, qos):
|
||||
"""A new MQTT message has been received."""
|
||||
values = json.loads(payload)
|
||||
|
||||
if values["state"] == "ON":
|
||||
self._state = True
|
||||
elif values["state"] == "OFF":
|
||||
self._state = False
|
||||
|
||||
if self._rgb is not None:
|
||||
try:
|
||||
red = int(values["color"]["r"])
|
||||
green = int(values["color"]["g"])
|
||||
blue = int(values["color"]["b"])
|
||||
|
||||
self._rgb = [red, green, blue]
|
||||
except KeyError:
|
||||
pass
|
||||
except ValueError:
|
||||
_LOGGER.warning("Invalid color value received.")
|
||||
|
||||
if self._brightness is not None:
|
||||
try:
|
||||
self._brightness = int(values["brightness"])
|
||||
except KeyError:
|
||||
pass
|
||||
except ValueError:
|
||||
_LOGGER.warning("Invalid brightness value received.")
|
||||
|
||||
self.update_ha_state()
|
||||
|
||||
if self._topic["state_topic"] is not None:
|
||||
mqtt.subscribe(self._hass, self._topic["state_topic"],
|
||||
state_received, self._qos)
|
||||
|
||||
@property
|
||||
def brightness(self):
|
||||
"""Return the brightness of this light between 0..255."""
|
||||
return self._brightness
|
||||
|
||||
@property
|
||||
def rgb_color(self):
|
||||
"""Return the RGB color value."""
|
||||
return self._rgb
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""No polling needed for a MQTT light."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the device if any."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if device is on."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def assumed_state(self):
|
||||
"""Return true if we do optimistic updates."""
|
||||
return self._optimistic
|
||||
|
||||
def turn_on(self, **kwargs):
|
||||
"""Turn the device on."""
|
||||
should_update = False
|
||||
|
||||
message = {"state": "ON"}
|
||||
|
||||
if ATTR_RGB_COLOR in kwargs:
|
||||
message["color"] = {
|
||||
"r": kwargs[ATTR_RGB_COLOR][0],
|
||||
"g": kwargs[ATTR_RGB_COLOR][1],
|
||||
"b": kwargs[ATTR_RGB_COLOR][2]
|
||||
}
|
||||
|
||||
if self._optimistic:
|
||||
self._rgb = kwargs[ATTR_RGB_COLOR]
|
||||
should_update = True
|
||||
|
||||
if ATTR_FLASH in kwargs:
|
||||
flash = kwargs.get(ATTR_FLASH)
|
||||
|
||||
if flash == FLASH_LONG:
|
||||
message["flash"] = self._flash_times[CONF_FLASH_TIME_LONG]
|
||||
elif flash == FLASH_SHORT:
|
||||
message["flash"] = self._flash_times[CONF_FLASH_TIME_SHORT]
|
||||
|
||||
if ATTR_TRANSITION in kwargs:
|
||||
message["transition"] = kwargs[ATTR_TRANSITION]
|
||||
|
||||
if ATTR_BRIGHTNESS in kwargs:
|
||||
message["brightness"] = int(kwargs[ATTR_BRIGHTNESS])
|
||||
|
||||
if self._optimistic:
|
||||
self._brightness = kwargs[ATTR_BRIGHTNESS]
|
||||
should_update = True
|
||||
|
||||
mqtt.publish(self._hass, self._topic["command_topic"],
|
||||
json.dumps(message), self._qos, self._retain)
|
||||
|
||||
if self._optimistic:
|
||||
# Optimistically assume that the light has changed state.
|
||||
self._state = True
|
||||
should_update = True
|
||||
|
||||
if should_update:
|
||||
self.update_ha_state()
|
||||
|
||||
def turn_off(self, **kwargs):
|
||||
"""Turn the device off."""
|
||||
message = {"state": "OFF"}
|
||||
|
||||
if ATTR_TRANSITION in kwargs:
|
||||
message["transition"] = kwargs[ATTR_TRANSITION]
|
||||
|
||||
mqtt.publish(self._hass, self._topic["command_topic"],
|
||||
json.dumps(message), self._qos, self._retain)
|
||||
|
||||
if self._optimistic:
|
||||
# Optimistically assume that the light has changed state.
|
||||
self._state = False
|
||||
self.update_ha_state()
|
||||
|
|
@ -0,0 +1,344 @@
|
|||
"""The tests for the MQTT JSON light platform.
|
||||
|
||||
Configuration for RGB Version with brightness:
|
||||
|
||||
light:
|
||||
platform: mqtt_json
|
||||
name: mqtt_json_light_1
|
||||
state_topic: "home/rgb1"
|
||||
command_topic: "home/rgb1/set"
|
||||
brightness: true
|
||||
rgb: true
|
||||
|
||||
Config without RGB:
|
||||
|
||||
light:
|
||||
platform: mqtt_json
|
||||
name: mqtt_json_light_1
|
||||
state_topic: "home/rgb1"
|
||||
command_topic: "home/rgb1/set"
|
||||
brightness: true
|
||||
|
||||
Config without RGB and brightness:
|
||||
|
||||
light:
|
||||
platform: mqtt_json
|
||||
name: mqtt_json_light_1
|
||||
state_topic: "home/rgb1"
|
||||
command_topic: "home/rgb1/set"
|
||||
"""
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from homeassistant.bootstrap import _setup_component
|
||||
from homeassistant.const import STATE_ON, STATE_OFF, ATTR_ASSUMED_STATE
|
||||
import homeassistant.components.light as light
|
||||
from tests.common import (
|
||||
get_test_home_assistant, mock_mqtt_component, fire_mqtt_message)
|
||||
|
||||
|
||||
class TestLightMQTTJSON(unittest.TestCase):
|
||||
"""Test the MQTT JSON light."""
|
||||
|
||||
def setUp(self): # pylint: disable=invalid-name
|
||||
"""Setup 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
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
def test_fail_setup_if_no_command_topic(self):
|
||||
"""Test if setup fails with no command topic."""
|
||||
self.hass.config.components = ['mqtt']
|
||||
assert not _setup_component(self.hass, light.DOMAIN, {
|
||||
light.DOMAIN: {
|
||||
'platform': 'mqtt_json',
|
||||
'name': 'test',
|
||||
}
|
||||
})
|
||||
self.assertIsNone(self.hass.states.get('light.test'))
|
||||
|
||||
def test_no_color_or_brightness_if_no_config(self):
|
||||
"""Test if there is no color and brightness if they aren't defined."""
|
||||
self.hass.config.components = ['mqtt']
|
||||
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',
|
||||
}
|
||||
})
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
self.assertIsNone(state.attributes.get('rgb_color'))
|
||||
self.assertIsNone(state.attributes.get('brightness'))
|
||||
|
||||
fire_mqtt_message(self.hass, 'test_light_rgb', '{"state":"ON"}')
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
self.assertIsNone(state.attributes.get('rgb_color'))
|
||||
self.assertIsNone(state.attributes.get('brightness'))
|
||||
|
||||
def test_controlling_state_via_topic(self):
|
||||
"""Test the controlling of the state via topic."""
|
||||
self.hass.config.components = ['mqtt']
|
||||
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,
|
||||
'qos': '0'
|
||||
}
|
||||
})
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
self.assertIsNone(state.attributes.get('rgb_color'))
|
||||
self.assertIsNone(state.attributes.get('brightness'))
|
||||
self.assertIsNone(state.attributes.get(ATTR_ASSUMED_STATE))
|
||||
|
||||
# Turn on the light, full white
|
||||
fire_mqtt_message(self.hass, 'test_light_rgb',
|
||||
'{"state":"ON",' +
|
||||
'"color":{"r":255,"g":255,"b":255},' +
|
||||
'"brightness":255}'
|
||||
)
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
self.assertEqual([255, 255, 255], state.attributes.get('rgb_color'))
|
||||
self.assertEqual(255, state.attributes.get('brightness'))
|
||||
|
||||
# Turn the light off
|
||||
fire_mqtt_message(self.hass, 'test_light_rgb', '{"state":"OFF"}')
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
fire_mqtt_message(self.hass, 'test_light_rgb',
|
||||
'{"state":"ON",' +
|
||||
'"brightness":100}'
|
||||
)
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
light_state = self.hass.states.get('light.test')
|
||||
self.hass.pool.block_till_done()
|
||||
self.assertEqual(100,
|
||||
light_state.attributes['brightness'])
|
||||
|
||||
fire_mqtt_message(self.hass, 'test_light_rgb',
|
||||
'{"state":"ON",' +
|
||||
'"color":{"r":125,"g":125,"b":125}}'
|
||||
)
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
light_state = self.hass.states.get('light.test')
|
||||
self.assertEqual([125, 125, 125],
|
||||
light_state.attributes.get('rgb_color'))
|
||||
|
||||
def test_sending_mqtt_commands_and_optimistic(self):
|
||||
"""Test the sending of command in optimistic mode."""
|
||||
self.hass.config.components = ['mqtt']
|
||||
assert _setup_component(self.hass, light.DOMAIN, {
|
||||
light.DOMAIN: {
|
||||
'platform': 'mqtt_json',
|
||||
'name': 'test',
|
||||
'command_topic': 'test_light_rgb/set',
|
||||
'brightness': True,
|
||||
'rgb': True,
|
||||
'qos': 2
|
||||
}
|
||||
})
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
self.assertTrue(state.attributes.get(ATTR_ASSUMED_STATE))
|
||||
|
||||
light.turn_on(self.hass, 'light.test')
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
self.assertEqual(('test_light_rgb/set', '{"state": "ON"}', 2, False),
|
||||
self.mock_publish.mock_calls[-1][1])
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
|
||||
light.turn_off(self.hass, 'light.test')
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
self.assertEqual(('test_light_rgb/set', '{"state": "OFF"}', 2, False),
|
||||
self.mock_publish.mock_calls[-1][1])
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
light.turn_on(self.hass, 'light.test', rgb_color=[75, 75, 75],
|
||||
brightness=50)
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
self.assertEqual('test_light_rgb/set',
|
||||
self.mock_publish.mock_calls[-1][1][0])
|
||||
self.assertEqual(2, self.mock_publish.mock_calls[-1][1][2])
|
||||
self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3])
|
||||
# Get the sent message
|
||||
message_json = json.loads(self.mock_publish.mock_calls[-1][1][1])
|
||||
self.assertEqual(50, message_json["brightness"])
|
||||
self.assertEqual(75, message_json["color"]["r"])
|
||||
self.assertEqual(75, message_json["color"]["g"])
|
||||
self.assertEqual(75, message_json["color"]["b"])
|
||||
self.assertEqual("ON", message_json["state"])
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
self.assertEqual((75, 75, 75), state.attributes['rgb_color'])
|
||||
self.assertEqual(50, state.attributes['brightness'])
|
||||
|
||||
def test_flash_short_and_long(self):
|
||||
"""Test for flash length being sent when included."""
|
||||
self.hass.config.components = ['mqtt']
|
||||
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')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
light.turn_on(self.hass, 'light.test', flash="short")
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
self.assertEqual('test_light_rgb/set',
|
||||
self.mock_publish.mock_calls[-1][1][0])
|
||||
self.assertEqual(0, self.mock_publish.mock_calls[-1][1][2])
|
||||
self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3])
|
||||
# Get the sent message
|
||||
message_json = json.loads(self.mock_publish.mock_calls[-1][1][1])
|
||||
self.assertEqual(5, message_json["flash"])
|
||||
self.assertEqual("ON", message_json["state"])
|
||||
|
||||
light.turn_on(self.hass, 'light.test', flash="long")
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
self.assertEqual('test_light_rgb/set',
|
||||
self.mock_publish.mock_calls[-1][1][0])
|
||||
self.assertEqual(0, self.mock_publish.mock_calls[-1][1][2])
|
||||
self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3])
|
||||
# Get the sent message
|
||||
message_json = json.loads(self.mock_publish.mock_calls[-1][1][1])
|
||||
self.assertEqual(15, message_json["flash"])
|
||||
self.assertEqual("ON", message_json["state"])
|
||||
|
||||
def test_transition(self):
|
||||
"""Test for transition time being sent when included."""
|
||||
self.hass.config.components = ['mqtt']
|
||||
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')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
light.turn_on(self.hass, 'light.test', transition=10)
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
self.assertEqual('test_light_rgb/set',
|
||||
self.mock_publish.mock_calls[-1][1][0])
|
||||
self.assertEqual(0, self.mock_publish.mock_calls[-1][1][2])
|
||||
self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3])
|
||||
# Get the sent message
|
||||
message_json = json.loads(self.mock_publish.mock_calls[-1][1][1])
|
||||
self.assertEqual(10, message_json["transition"])
|
||||
self.assertEqual("ON", message_json["state"])
|
||||
|
||||
# Transition back off
|
||||
light.turn_off(self.hass, 'light.test', transition=10)
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
self.assertEqual('test_light_rgb/set',
|
||||
self.mock_publish.mock_calls[-1][1][0])
|
||||
self.assertEqual(0, self.mock_publish.mock_calls[-1][1][2])
|
||||
self.assertEqual(False, self.mock_publish.mock_calls[-1][1][3])
|
||||
# Get the sent message
|
||||
message_json = json.loads(self.mock_publish.mock_calls[-1][1][1])
|
||||
self.assertEqual(10, message_json["transition"])
|
||||
self.assertEqual("OFF", message_json["state"])
|
||||
|
||||
def test_invalid_color_and_brightness_values(self):
|
||||
"""Test that invalid color/brightness values are ignored."""
|
||||
self.hass.config.components = ['mqtt']
|
||||
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,
|
||||
'qos': '0'
|
||||
}
|
||||
})
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
self.assertIsNone(state.attributes.get('rgb_color'))
|
||||
self.assertIsNone(state.attributes.get('brightness'))
|
||||
self.assertIsNone(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}'
|
||||
)
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
self.assertEqual([255, 255, 255], state.attributes.get('rgb_color'))
|
||||
self.assertEqual(255, state.attributes.get('brightness'))
|
||||
|
||||
# Bad color values
|
||||
fire_mqtt_message(self.hass, 'test_light_rgb',
|
||||
'{"state":"ON",' +
|
||||
'"color":{"r":"bad","g":"val","b":"test"}}'
|
||||
)
|
||||
self.hass.pool.block_till_done()
|
||||
|
||||
# Color should not have changed
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
self.assertEqual([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.pool.block_till_done()
|
||||
|
||||
# Brightness should not have changed
|
||||
state = self.hass.states.get('light.test')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
self.assertEqual(255, state.attributes.get('brightness'))
|
||||
Loading…
Reference in New Issue