jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-10-16 23:29:30 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
be5dfcd06d68e8ae423fedc287d05552e44cd2ea
core/tests/components/mqtt/test_light_json.py

2738 lines
86 KiB
Python
Raw Blame History

"""The tests for the MQTT JSON light platform.
Configuration with RGB, brightness, color temp, effect, and XY:
mqtt:
light:
schema: json
name: mqtt_json_light_1
state_topic: "home/rgb1"
command_topic: "home/rgb1/set"
brightness: true
color_temp: true
effect: true
rgb: true
xy: true
Configuration with RGB, brightness, color temp (mireds) and effect:
mqtt:
light:
schema: json
name: mqtt_json_light_1
state_topic: "home/rgb1"
command_topic: "home/rgb1/set"
brightness: true
color_temp: true
color_temp_kelvin: false
effect: true
rgb: true
Configuration with RGB, brightness and color temp (Kelvin):
mqtt:
light:
schema: json
name: mqtt_json_light_1
state_topic: "home/rgb1"
command_topic: "home/rgb1/set"
brightness: true
rgb: true
color_temp: true
color_temp_kelvin: true
Configuration with RGB, brightness:
mqtt:
light:
schema: json
name: mqtt_json_light_1
state_topic: "home/rgb1"
command_topic: "home/rgb1/set"
brightness: true
rgb: true
Config without RGB:
mqtt:
light:
schema: 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"
Config with brightness and scale:
light:
platform: mqtt_json
name: test
state_topic: "mqtt_json_light_1"
command_topic: "mqtt_json_light_1/set"
brightness: true
brightness_scale: 99
"""
import copy
import json
from typing import Any
from unittest.mock import call, patch
import pytest
from homeassistant.components import light, mqtt
from homeassistant.components.mqtt.light.schema_basic import (
MQTT_LIGHT_ATTRIBUTES_BLOCKED,
)
from homeassistant.components.mqtt.models import PublishPayloadType
from homeassistant.const import (
ATTR_ASSUMED_STATE,
ATTR_SUPPORTED_FEATURES,
STATE_OFF,
STATE_ON,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers import entity_registry as er
from homeassistant.util.json import json_loads
from .common import (
help_custom_config,
help_test_availability_when_connection_lost,
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_discovery_update_unchanged,
help_test_encoding_subscribable_topics,
help_test_entity_debug_info_message,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update_discovery_update,
help_test_entity_id_update_subscriptions,
help_test_publishing_with_custom_encoding,
help_test_reloadable,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_setting_blocked_attribute_via_mqtt_json_message,
help_test_skipped_async_ha_write_state,
help_test_unique_id,
help_test_update_with_json_attrs_bad_json,
help_test_update_with_json_attrs_not_dict,
)
from tests.common import async_fire_mqtt_message, mock_restore_cache
from tests.components.light import common
from tests.typing import MqttMockHAClientGenerator, MqttMockPahoClient
DEFAULT_CONFIG = {
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test-topic",
}
}
}
COLOR_MODES_CONFIG = {
mqtt.DOMAIN: {
light.DOMAIN: {
"brightness": True,
"effect": True,
"command_topic": "test_light_rgb/set",
"name": "test",
"schema": "json",
"supported_color_modes": [
"color_temp",
"hs",
"rgb",
"rgbw",
"rgbww",
"white",
"xy",
],
"qos": 0,
}
}
}
GROUP_MEMBER_1_TOPIC = "homeassistant/light/member_1/config"
GROUP_MEMBER_2_TOPIC = "homeassistant/light/member_2/config"
GROUP_TOPIC = "homeassistant/light/group/config"
GROUP_DISCOVERY_MEMBER_1_CONFIG = json.dumps(
{
"schema": "json",
"command_topic": "test-command-topic-member1",
"unique_id": "very_unique_member1",
"name": "member1",
"default_entity_id": "light.member1",
}
)
GROUP_DISCOVERY_MEMBER_2_CONFIG = json.dumps(
{
"schema": "json",
"command_topic": "test-command-topic-member2",
"unique_id": "very_unique_member2",
"name": "member2",
"default_entity_id": "light.member2",
}
)
GROUP_DISCOVERY_LIGHT_GROUP_CONFIG = json.dumps(
{
"schema": "json",
"command_topic": "test-command-topic-group",
"state_topic": "test-state-topic-group",
"unique_id": "very_unique_group",
"name": "group",
"default_entity_id": "light.group",
"group": ["very_unique_member1", "very_unique_member2"],
}
)
class JsonValidator:
"""Helper to compare JSON."""
def __init__(self, jsondata: bytes | str) -> None:
"""Initialize JSON validator."""
self.jsondata = jsondata
def __eq__(self, other: bytes | str) -> bool: # type:ignore[override]
"""Compare JSON data."""
return json_loads(self.jsondata) == json_loads(other)
@pytest.mark.parametrize("hass_config", [DEFAULT_CONFIG])
async def test_simple_on_off_light(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test if setup fails with no command topic."""
assert await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state and state.state == STATE_UNKNOWN
assert state.attributes["supported_color_modes"] == ["onoff"]
@pytest.mark.parametrize(
"hass_config", [{mqtt.DOMAIN: {light.DOMAIN: {"schema": "json", "name": "test"}}}]
)
async def test_fail_setup_if_no_command_topic(
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test if setup fails with no command topic."""
assert await mqtt_mock_entry()
assert "required key not provided" in caplog.text
@pytest.mark.parametrize(
("hass_config", "error"),
[
(
help_custom_config(
light.DOMAIN,
COLOR_MODES_CONFIG,
({"supported_color_modes": ["onoff", "rgb"]},),
),
"Invalid supported_color_modes ['onoff', 'rgb']",
),
(
help_custom_config(
light.DOMAIN,
COLOR_MODES_CONFIG,
({"supported_color_modes": ["brightness", "rgb"]},),
),
"Invalid supported_color_modes ['brightness', 'rgb']",
),
(
help_custom_config(
light.DOMAIN,
COLOR_MODES_CONFIG,
({"supported_color_modes": ["unknown"]},),
),
"value must be one of [<ColorMode.",
),
],
)
async def test_fail_setup_if_color_modes_invalid(
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
error: str,
) -> None:
"""Test if setup fails if supported color modes is invalid."""
assert await mqtt_mock_entry()
assert error in caplog.text
@pytest.mark.parametrize("hass_config", [COLOR_MODES_CONFIG])
async def test_turn_on_with_unknown_color_mode_optimistic(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test setup and turn with unknown color_mode in optimistic mode."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
# Turn on the light without brightness or color_temp attributes
await common.async_turn_on(hass, "light.test")
state = hass.states.get("light.test")
assert state.attributes.get("color_mode") == light.ColorMode.UNKNOWN
assert state.attributes.get("brightness") is None
assert state.attributes.get("color_temp_kelvin") is None
assert state.state == STATE_ON
# Turn on the light with brightness or color_temp attributes
await common.async_turn_on(
hass, "light.test", brightness=50, color_temp_kelvin=5208
)
state = hass.states.get("light.test")
assert state.attributes.get("color_mode") == light.ColorMode.COLOR_TEMP
assert state.attributes.get("brightness") == 50
assert state.attributes.get("color_temp_kelvin") == 5208
assert state.state == STATE_ON
@pytest.mark.parametrize(
"hass_config",
[
(
help_custom_config(
light.DOMAIN,
COLOR_MODES_CONFIG,
({"state_topic": "test_light"},),
)
)
],
)
async def test_controlling_state_with_unknown_color_mode(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test setup and turn with unknown color_mode in optimistic mode."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
# Send `on` state but omit other attributes
async_fire_mqtt_message(
hass,
"test_light",
'{"state": "ON"}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get(light.ATTR_COLOR_TEMP_KELVIN) is None
assert state.attributes.get(light.ATTR_BRIGHTNESS) is None
assert state.attributes.get(light.ATTR_COLOR_MODE) == light.ColorMode.UNKNOWN
# Send complete light state
async_fire_mqtt_message(
hass,
"test_light",
'{"state": "ON", "brightness": 50, "color_mode": "color_temp", "color_temp": 192}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get(light.ATTR_COLOR_TEMP_KELVIN) == 5208
assert state.attributes.get(light.ATTR_BRIGHTNESS) == 50
assert state.attributes.get(light.ATTR_COLOR_MODE) == light.ColorMode.COLOR_TEMP
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
}
}
}
],
)
async def test_no_color_brightness_color_temp_if_no_topics(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test for no RGB, brightness, color temp, effector XY."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
expected_features = (
light.LightEntityFeature.FLASH | light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == expected_features
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("brightness") is None
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") is None
assert state.attributes.get("xy_color") is None
assert state.attributes.get("hs_color") is None
async_fire_mqtt_message(hass, "test_light_rgb", '{"state":"ON"}')
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("brightness") is None
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") is None
assert state.attributes.get("xy_color") is None
assert state.attributes.get("hs_color") is None
async_fire_mqtt_message(hass, "test_light_rgb", '{"state":"OFF"}')
state = hass.states.get("light.test")
assert state.state == STATE_OFF
async_fire_mqtt_message(hass, "test_light_rgb", '{"state": null}')
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"supported_color_modes": ["brightness"],
}
}
},
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"brightness": True,
}
}
},
],
)
async def test_brightness_only(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test brightness only light.
There are two possible configurations for brightness only light:
1) Set up "brightness" as supported color mode.
2) Set "brightness" flag to true.
"""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == [
light.ColorMode.BRIGHTNESS
]
expected_features = (
light.LightEntityFeature.FLASH | light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == expected_features
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("brightness") is None
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") is None
assert state.attributes.get("xy_color") is None
assert state.attributes.get("hs_color") is None
async_fire_mqtt_message(hass, "test_light_rgb", '{"state":"ON", "brightness": 50}')
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("brightness") == 50
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") is None
assert state.attributes.get("xy_color") is None
assert state.attributes.get("hs_color") is None
async_fire_mqtt_message(hass, "test_light_rgb", '{"state":"OFF"}')
state = hass.states.get("light.test")
assert state.state == STATE_OFF
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"supported_color_modes": ["color_temp"],
}
}
},
],
)
async def test_color_temp_only(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test a light that only support color_temp as supported color mode."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == [
light.ColorMode.COLOR_TEMP
]
expected_features = (
light.LightEntityFeature.FLASH | light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) == expected_features
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("brightness") is None
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") is None
assert state.attributes.get("xy_color") is None
assert state.attributes.get("hs_color") is None
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_mode": "color_temp", "color_temp": 250, "brightness": 50}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") == (255, 206, 166)
assert state.attributes.get("brightness") == 50
assert state.attributes.get("color_temp_kelvin") == 4000
assert state.attributes.get("effect") is None
assert state.attributes.get("xy_color") == (0.42, 0.365)
assert state.attributes.get("hs_color") == (26.812, 34.87)
async_fire_mqtt_message(hass, "test_light_rgb", '{"state":"OFF"}')
state = hass.states.get("light.test")
assert state.state == STATE_OFF
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"color_temp_kelvin": True,
"effect": True,
"supported_color_modes": ["color_temp", "hs"],
"qos": "0",
}
}
}
],
)
async def test_controlling_state_color_temp_kelvin(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the controlling of the state via topic in Kelvin mode."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
color_modes = [light.ColorMode.COLOR_TEMP, light.ColorMode.HS]
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == color_modes
expected_features = (
light.LightEntityFeature.EFFECT
| light.LightEntityFeature.FLASH
| light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) is expected_features
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("brightness") is None
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") 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
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON",'
'"color":{"h": 44.098, "s": 2.43},'
'"color_mode": "hs",'
'"brightness":255,'
'"color_temp":155,'
'"effect":"colorloop"}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") == (255, 253, 249)
assert state.attributes.get("brightness") == 255
assert state.attributes.get("color_temp_kelvin") is None # rgb color has priority
assert state.attributes.get("effect") == "colorloop"
assert state.attributes.get("xy_color") == (0.328, 0.333)
assert state.attributes.get("hs_color") == (44.098, 2.43)
# Turn on the light
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON",'
'"brightness":255,'
'"color":null,'
'"color_mode":"color_temp",'
'"color_temp":6451,' # Kelvin
'"effect":"colorloop"}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") == (
255,
253,
249,
) # temp converted to color
assert state.attributes.get("brightness") == 255
assert state.attributes.get("color_temp_kelvin") == 6451
assert state.attributes.get("effect") == "colorloop"
assert state.attributes.get("xy_color") == (0.328, 0.333) # temp converted to color
assert state.attributes.get("hs_color") == (44.098, 2.43) # temp converted to color
@pytest.mark.parametrize(
("hass_config", "expected_features"),
[
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
}
}
},
light.LightEntityFeature.FLASH | light.LightEntityFeature.TRANSITION,
),
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"flash": True,
"transition": True,
}
}
},
light.LightEntityFeature.FLASH | light.LightEntityFeature.TRANSITION,
),
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"flash": True,
"transition": False,
}
}
},
light.LightEntityFeature.FLASH,
),
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"flash": False,
"transition": True,
}
}
},
light.LightEntityFeature.TRANSITION,
),
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"flash": False,
"transition": False,
}
}
},
light.LightEntityFeature(0),
),
],
ids=[
"default",
"explicit_on",
"flash_only",
"transition_only",
"no_flash_not_transition",
],
)
async def test_flash_and_transition_feature_flags(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
expected_features: light.LightEntityFeature,
) -> None:
"""Test for no RGB, brightness, color temp, effector XY."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) is expected_features
@pytest.mark.parametrize(
"hass_config",
[
help_custom_config(
light.DOMAIN, COLOR_MODES_CONFIG, ({"state_topic": "test_light_rgb"},)
)
],
)
async def test_controlling_state_via_topic(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test the controlling of the state via topic for a light supporting color mode."""
supported_color_modes = ["color_temp", "hs", "rgb", "rgbw", "rgbww", "white", "xy"]
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
expected_features = (
light.LightEntityFeature.EFFECT
| light.LightEntityFeature.FLASH
| light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) is expected_features
assert state.attributes.get("brightness") is None
assert state.attributes.get("color_mode") is None
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") is None
assert state.attributes.get("hs_color") is None
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("rgbw_color") is None
assert state.attributes.get("rgbww_color") is None
assert state.attributes.get("supported_color_modes") == supported_color_modes
assert state.attributes.get("xy_color") is None
assert not state.attributes.get(ATTR_ASSUMED_STATE)
# Turn on the light, rgbww mode, additional values in the update
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON",'
'"color_mode":"rgbww",'
'"color":{"r":255,"g":128,"b":64, "c": 32, "w": 16, "x": 1, "y": 1},'
'"brightness":255,'
'"color_temp":155,'
'"effect":"colorloop"}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") == 255
assert state.attributes.get("color_mode") == "rgbww"
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") == "colorloop"
assert state.attributes.get("hs_color") == (20.552, 70.98)
assert state.attributes.get("rgb_color") == (255, 136, 74)
assert state.attributes.get("rgbw_color") is None
assert state.attributes.get("rgbww_color") == (255, 128, 64, 32, 16)
assert state.attributes.get("xy_color") == (0.571, 0.361)
# Light turned off
async_fire_mqtt_message(hass, "test_light_rgb", '{"state":"OFF"}')
state = hass.states.get("light.test")
assert state.state == STATE_OFF
# Light turned on, brightness 100
async_fire_mqtt_message(hass, "test_light_rgb", '{"state":"ON", "brightness":100}')
state = hass.states.get("light.test")
assert state.attributes["brightness"] == 100
# Zero brightness value is ignored
async_fire_mqtt_message(hass, "test_light_rgb", '{"state":"ON", "brightness":0}')
state = hass.states.get("light.test")
assert state.attributes["brightness"] == 100
# RGB color
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_mode":"rgb", "color":{"r":64,"g":128,"b":255}}',
)
state = hass.states.get("light.test")
assert state.attributes.get("color_mode") == "rgb"
assert state.attributes.get("rgb_color") == (64, 128, 255)
# RGBW color
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_mode":"rgbw", "color":{"r":64,"g":128,"b":255,"w":32}}',
)
state = hass.states.get("light.test")
assert state.attributes.get("color_mode") == "rgbw"
assert state.attributes.get("rgbw_color") == (64, 128, 255, 32)
# XY color
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_mode":"xy", "color":{"x":0.135,"y":0.235}}',
)
state = hass.states.get("light.test")
assert state.attributes.get("color_mode") == "xy"
assert state.attributes.get("xy_color") == (0.135, 0.235)
# HS color
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_mode":"hs", "color":{"h":180,"s":50}}',
)
state = hass.states.get("light.test")
assert state.attributes.get("color_mode") == "hs"
assert state.attributes.get("hs_color") == (180.0, 50.0)
# Color temp
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_mode":"color_temp", "color_temp":155}',
)
state = hass.states.get("light.test")
assert state.attributes.get("color_mode") == "color_temp"
assert state.attributes.get("color_temp_kelvin") == 6451
# White
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_mode":"white", "brightness":123}',
)
state = hass.states.get("light.test")
assert state.attributes.get("color_mode") == "white"
assert state.attributes.get("brightness") == 123
# Effect
async_fire_mqtt_message(
hass, "test_light_rgb", '{"state":"ON", "effect":"other_effect"}'
)
state = hass.states.get("light.test")
assert state.attributes.get("effect") == "other_effect"
# Invalid color mode
async_fire_mqtt_message(
hass, "test_light_rgb", '{"state":"ON", "color_mode":"col_temp"}'
)
assert "Invalid color mode 'col_temp' received" in caplog.text
caplog.clear()
# Incomplete color
async_fire_mqtt_message(
hass, "test_light_rgb", '{"state":"ON", "color_mode":"rgb"}'
)
assert (
"Invalid or incomplete color value '{'state': 'ON', 'color_mode': 'rgb'}' received"
in caplog.text
)
caplog.clear()
# Invalid color
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_mode":"rgb", "color":{"r":64,"g":128,"b":"cow"}}',
)
assert (
"Invalid or incomplete color value '{'state': 'ON', 'color_mode': 'rgb', 'color': {'r': 64, 'g': 128, 'b': 'cow'}}' received"
in caplog.text
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"command_topic": "test_light_rgb/set",
"effect": True,
"name": "test",
"qos": 2,
"schema": "json",
"supported_color_modes": [
"color_temp",
"hs",
"rgb",
"rgbw",
"rgbww",
"white",
"xy",
],
}
}
}
],
)
async def test_sending_mqtt_commands_and_optimistic(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the sending of command in optimistic mode for a light supporting color mode."""
supported_color_modes = ["color_temp", "hs", "rgb", "rgbw", "rgbww", "white", "xy"]
fake_state = State(
"light.test",
"on",
{
"brightness": 95,
"color_temp_kelvin": 10000,
"color_mode": "rgb",
"effect": "random",
"hs_color": [100, 100],
},
)
mock_restore_cache(hass, (fake_state,))
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_ON
expected_features = (
light.LightEntityFeature.EFFECT
| light.LightEntityFeature.FLASH
| light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) is expected_features
assert state.attributes.get("brightness") == 95
assert state.attributes.get("color_mode") == "rgb"
assert state.attributes.get("color_temp_kelvin") is None
assert state.attributes.get("effect") == "random"
assert state.attributes.get("hs_color") is None
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("rgbw_color") is None
assert state.attributes.get("rgbww_color") is None
assert state.attributes.get("supported_color_modes") == supported_color_modes
assert state.attributes.get("white") is None
assert state.attributes.get("xy_color") is None
assert state.attributes.get(ATTR_ASSUMED_STATE)
# Turn the light on
await common.async_turn_on(hass, "light.test")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", '{"state":"ON"}', 2, False
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_ON
# Turn the light on with color temperature
await common.async_turn_on(hass, "light.test", color_temp_kelvin=11111)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator('{"state":"ON","color_temp":90}'),
2,
False,
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_ON
# Turn the light off
await common.async_turn_off(hass, "light.test")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", '{"state":"OFF"}', 2, False
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_OFF
# Set hs color
await common.async_turn_on(hass, "light.test", brightness=75, hs_color=(359, 78))
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
assert state.attributes["color_mode"] == "hs"
assert state.attributes["hs_color"] == (359, 78)
assert state.attributes["rgb_color"] == (255, 56, 59)
assert state.attributes["xy_color"] == (0.654, 0.301)
assert state.attributes["rgbw_color"] is None
assert state.attributes["rgbww_color"] is None
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"h": 359.0, "s": 78.0}, "brightness": 75}'
),
2,
False,
)
mqtt_mock.async_publish.reset_mock()
# Set rgb color
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
assert state.attributes["color_mode"] == "rgb"
assert state.attributes["hs_color"] == (30.118, 100.0)
assert state.attributes["rgb_color"] == (255, 128, 0)
assert state.attributes["xy_color"] == (0.611, 0.375)
assert state.attributes["rgbw_color"] is None
assert state.attributes["rgbww_color"] is None
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 255, "g": 128, "b": 0} }'),
2,
False,
)
mqtt_mock.async_publish.reset_mock()
# Set rgbw color
await common.async_turn_on(hass, "light.test", rgbw_color=(255, 128, 0, 123))
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
assert state.attributes["color_mode"] == "rgbw"
assert state.attributes["rgbw_color"] == (255, 128, 0, 123)
assert state.attributes["hs_color"] == (30.0, 67.451)
assert state.attributes["rgb_color"] == (255, 169, 83)
assert state.attributes["rgbww_color"] is None
assert state.attributes["xy_color"] == (0.526, 0.393)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"r": 255, "g": 128, "b": 0, "w": 123} }'
),
2,
False,
)
mqtt_mock.async_publish.reset_mock()
# Set rgbww color
await common.async_turn_on(hass, "light.test", rgbww_color=(255, 128, 0, 45, 32))
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
assert state.attributes["color_mode"] == "rgbww"
assert state.attributes["rgbww_color"] == (255, 128, 0, 45, 32)
assert state.attributes["hs_color"] == (29.872, 92.157)
assert state.attributes["rgb_color"] == (255, 137, 20)
assert state.attributes["rgbw_color"] is None
assert state.attributes["xy_color"] == (0.596, 0.382)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"r": 255, "g": 128, "b": 0, "c": 45, "w": 32} }'
),
2,
False,
)
mqtt_mock.async_publish.reset_mock()
# Set xy color
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=(0.123, 0.223)
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 50
assert state.attributes["color_mode"] == "xy"
assert state.attributes["hs_color"] == (196.471, 100.0)
assert state.attributes["rgb_color"] == (0, 185, 255)
assert state.attributes["xy_color"] == (0.123, 0.223)
assert state.attributes["rgbw_color"] is None
assert state.attributes["rgbww_color"] is None
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"x": 0.123, "y": 0.223}, "brightness": 50}'
),
2,
False,
)
mqtt_mock.async_publish.reset_mock()
# Set to white
await common.async_turn_on(hass, "light.test", white=75)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
assert state.attributes["color_mode"] == "white"
assert state.attributes["hs_color"] is None
assert state.attributes["rgb_color"] is None
assert state.attributes["xy_color"] is None
assert state.attributes["rgbw_color"] is None
assert state.attributes["rgbww_color"] is None
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "white": 75}'),
2,
False,
)
mqtt_mock.async_publish.reset_mock()
# Set to white, brightness also present in turn_on
await common.async_turn_on(hass, "light.test", brightness=60, white=80)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 60
assert state.attributes["color_mode"] == "white"
assert state.attributes["hs_color"] is None
assert state.attributes["rgb_color"] is None
assert state.attributes["xy_color"] is None
assert state.attributes["rgbw_color"] is None
assert state.attributes["rgbww_color"] is None
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "white": 60}'),
2,
False,
)
mqtt_mock.async_publish.reset_mock()
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"supported_color_modes": ["hs"],
}
}
}
],
)
async def test_sending_hs_color(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test light.turn_on with hs color sends hs color parameters."""
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
mqtt_mock.reset_mock()
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"h": 210.824, "s": 100.0},'
' "brightness": 50}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"h": 359.0, "s": 78.0},'
' "brightness": 50}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"h": 30.118, "s": 100.0}}'),
0,
False,
),
],
any_order=True,
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"supported_color_modes": ["rgb"],
}
}
}
],
)
async def test_sending_rgb_color_no_brightness(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test light.turn_on with hs color sends rgb color parameters."""
await hass.async_block_till_done()
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
await common.async_turn_on(
hass, "light.test", rgb_color=(255, 128, 0), brightness=255
)
mqtt_mock.async_publish.assert_has_calls(
[
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 0, "g": 24, "b": 50}}'),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 50, "g": 11, "b": 12}}'),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 255, "g": 128, "b": 0}}'),
0,
False,
),
],
any_order=True,
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"command_topic": "test_light_rgb/set",
"name": "test",
"schema": "json",
"supported_color_modes": ["rgb", "rgbw", "rgbww"],
}
}
}
],
)
async def test_sending_rgb_color_no_brightness2(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test light.turn_on with hs color sends rgb color parameters."""
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
await common.async_turn_on(
hass, "light.test", rgb_color=(255, 128, 0), brightness=255
)
await common.async_turn_on(
hass, "light.test", rgbw_color=(128, 64, 32, 16), brightness=128
)
await common.async_turn_on(
hass, "light.test", rgbww_color=(128, 64, 32, 16, 8), brightness=64
)
mqtt_mock.async_publish.assert_has_calls(
[
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 0, "g": 24, "b": 50}}'),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 50, "g": 11, "b": 12}}'),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 255, "g": 128, "b": 0}}'),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"r": 64, "g": 32, "b": 16, "w": 8}}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"r": 32, "g": 16, "b": 8, "c": 4, "w": 2}}'
),
0,
False,
),
],
any_order=True,
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"supported_color_modes": ["rgb"],
"brightness": True,
}
}
}
],
)
async def test_sending_rgb_color_with_brightness(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test light.turn_on with hs color sends rgb color parameters."""
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=255, hs_color=(359, 78))
await common.async_turn_on(hass, "light.test", brightness=1)
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"r": 0, "g": 124, "b": 255},'
' "brightness": 50}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"r": 255, "g": 56, "b": 59},'
' "brightness": 255}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "brightness": 1}'),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 255, "g": 128, "b": 0}}'),
0,
False,
),
],
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"supported_color_modes": ["rgb"],
"brightness": True,
"brightness_scale": 100,
}
}
}
],
)
async def test_sending_rgb_color_with_scaled_brightness(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test light.turn_on with hs color sends rgb color parameters."""
await hass.async_block_till_done()
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=255, hs_color=(359, 78))
await common.async_turn_on(hass, "light.test", brightness=1)
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"r": 0, "g": 124, "b": 255},'
' "brightness": 20}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"r": 255, "g": 56, "b": 59},'
' "brightness": 100}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "brightness": 1}'),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"r": 255, "g": 128, "b": 0}}'),
0,
False,
),
],
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"brightness_scale": 100,
"supported_color_modes": ["hs", "white"],
"white_scale": 50,
}
}
}
],
)
async def test_sending_scaled_white(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test light.turn_on with scaled white."""
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(hass, "light.test", brightness=128)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", JsonValidator('{"state":"ON", "brightness":50}'), 0, False
)
mqtt_mock.async_publish.reset_mock()
await common.async_turn_on(hass, "light.test", brightness=255, white=25)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", JsonValidator('{"state":"ON", "white":50}'), 0, False
)
mqtt_mock.async_publish.reset_mock()
await common.async_turn_on(hass, "light.test", white=25)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", JsonValidator('{"state":"ON", "white":5}'), 0, False
)
mqtt_mock.async_publish.reset_mock()
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"supported_color_modes": ["xy"],
}
}
}
],
)
async def test_sending_xy_color(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test light.turn_on with hs color sends xy color parameters."""
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"x": 0.123, "y": 0.123},'
' "brightness": 50}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator(
'{"state": "ON", "color": {"x": 0.654, "y": 0.301},'
' "brightness": 50}'
),
0,
False,
),
call(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "color": {"x": 0.611, "y": 0.375}}'),
0,
False,
),
],
any_order=True,
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"effect": True,
"qos": 0,
}
}
}
],
)
async def test_effect(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test for effect being sent when included."""
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
expected_features = (
light.LightEntityFeature.EFFECT
| light.LightEntityFeature.FLASH
| light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) is expected_features
await common.async_turn_on(hass, "light.test")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", JsonValidator('{"state": "ON"}'), 0, False
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("effect") is None
await common.async_turn_on(hass, "light.test", effect="rainbow")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "effect": "rainbow"}'),
0,
False,
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("effect") == "rainbow"
await common.async_turn_on(hass, "light.test", effect="colorloop")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "effect": "colorloop"}'),
0,
False,
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("effect") == "colorloop"
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"flash_time_short": 5,
"flash_time_long": 15,
"qos": 0,
}
}
}
],
)
async def test_flash_short_and_long(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test for flash length being sent when included."""
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
expected_features = (
light.LightEntityFeature.FLASH | light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) is expected_features
await common.async_turn_on(hass, "light.test", flash="short")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", JsonValidator('{"state": "ON", "flash": 5}'), 0, False
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_ON
await common.async_turn_on(hass, "light.test", flash="long")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", JsonValidator('{"state": "ON", "flash": 15}'), 0, False
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_ON
await common.async_turn_off(hass, "light.test", flash="short")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", JsonValidator('{"state": "OFF", "flash": 5}'), 0, False
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_OFF
await common.async_turn_off(hass, "light.test", flash="long")
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", JsonValidator('{"state": "OFF", "flash": 15}'), 0, False
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_OFF
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_light_rgb/set",
"qos": 0,
}
}
}
],
)
async def test_transition(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test for transition time being sent when included."""
mqtt_mock = await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
expected_features = (
light.LightEntityFeature.FLASH | light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) is expected_features
await common.async_turn_on(hass, "light.test", transition=15)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator('{"state": "ON", "transition": 15}'),
0,
False,
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_ON
await common.async_turn_off(hass, "light.test", transition=30)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator('{"state": "OFF", "transition": 30}'),
0,
False,
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("light.test")
assert state.state == STATE_OFF
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_bright_scale",
"command_topic": "test_light_bright_scale/set",
"supported_color_modes": ["brightness"],
"brightness_scale": 99,
}
}
}
],
)
async def test_brightness_scale(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test for brightness scaling."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
assert state.attributes.get("brightness") is None
assert not state.attributes.get(ATTR_ASSUMED_STATE)
# Turn on the light
async_fire_mqtt_message(hass, "test_light_bright_scale", '{"state":"ON"}')
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") is None
# Turn on the light with brightness
async_fire_mqtt_message(
hass, "test_light_bright_scale", '{"state":"ON", "brightness": 99}'
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") == 255
# Turn on the light with half brightness
async_fire_mqtt_message(
hass, "test_light_bright_scale", '{"state":"ON", "brightness": 50}'
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") == 129
# Test limmiting max brightness
async_fire_mqtt_message(
hass, "test_light_bright_scale", '{"state":"ON", "brightness": 103}'
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") == 255
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_bright_scale",
"command_topic": "test_light_bright_scale/set",
"brightness": True,
"brightness_scale": 99,
"supported_color_modes": ["hs", "white"],
"white_scale": 50,
}
}
}
],
)
async def test_white_scale(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test for white scaling."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
assert state.attributes.get("brightness") is None
assert not state.attributes.get(ATTR_ASSUMED_STATE)
# Turn on the light
async_fire_mqtt_message(hass, "test_light_bright_scale", '{"state":"ON"}')
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") is None
# Turn on the light with brightness
async_fire_mqtt_message(
hass,
"test_light_bright_scale",
'{"state":"ON", "brightness": 99, "color_mode":"hs", "color":{"h":180,"s":50}}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") == 255
# Turn on the light with white - white_scale is NOT used
async_fire_mqtt_message(
hass,
"test_light_bright_scale",
'{"state":"ON", "color_mode":"white", "brightness": 50}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") == 129
async def test_light_group_discovery_members_before_group(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the discovery of a light group and linked entity IDs.
The members are discovered first, so they are known in the entity registry.
"""
await mqtt_mock_entry()
# Discover light group members
async_fire_mqtt_message(hass, GROUP_MEMBER_1_TOPIC, GROUP_DISCOVERY_MEMBER_1_CONFIG)
async_fire_mqtt_message(hass, GROUP_MEMBER_2_TOPIC, GROUP_DISCOVERY_MEMBER_2_CONFIG)
await hass.async_block_till_done()
# Discover group
async_fire_mqtt_message(hass, GROUP_TOPIC, GROUP_DISCOVERY_LIGHT_GROUP_CONFIG)
await hass.async_block_till_done()
assert hass.states.get("light.member1") is not None
assert hass.states.get("light.member2") is not None
group_state = hass.states.get("light.group")
assert group_state is not None
assert group_state.attributes.get("entity_id") == ["light.member1", "light.member2"]
assert group_state.attributes.get("icon") == "mdi:lightbulb-group"
async def test_light_group_discovery_group_before_members(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
entity_registry: er.EntityRegistry,
) -> None:
"""Test the discovery of a light group and linked entity IDs.
The group is discovered first, so the group members are
not (all) known yet in the entity registry.
The entity property should be updates as soon as member entities
are discovered, updated or removed.
"""
await mqtt_mock_entry()
# Discover group
async_fire_mqtt_message(hass, GROUP_TOPIC, GROUP_DISCOVERY_LIGHT_GROUP_CONFIG)
await hass.async_block_till_done()
# Discover light group members
async_fire_mqtt_message(hass, GROUP_MEMBER_1_TOPIC, GROUP_DISCOVERY_MEMBER_1_CONFIG)
async_fire_mqtt_message(hass, GROUP_MEMBER_2_TOPIC, GROUP_DISCOVERY_MEMBER_2_CONFIG)
await hass.async_block_till_done()
assert hass.states.get("light.member1") is not None
assert hass.states.get("light.member2") is not None
group_state = hass.states.get("light.group")
assert group_state is not None
assert group_state.attributes.get("entity_id") == ["light.member1", "light.member2"]
assert group_state.attributes.get("icon") == "mdi:lightbulb-group"
# Remove member 1
async_fire_mqtt_message(hass, GROUP_MEMBER_1_TOPIC, "")
await hass.async_block_till_done()
assert hass.states.get("light.member1") is None
assert hass.states.get("light.member2") is not None
group_state = hass.states.get("light.group")
assert group_state is not None
assert group_state.attributes.get("entity_id") == ["light.member2"]
# Rename member 2
entity_registry.async_update_entity(
"light.member2", new_entity_id="light.member2_updated"
)
await hass.async_block_till_done()
group_state = hass.states.get("light.group")
assert group_state is not None
assert group_state.attributes.get("entity_id") == ["light.member2_updated"]
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"brightness": True,
"supported_color_modes": ["hs", "color_temp"],
"qos": "0",
}
}
}
],
)
async def test_invalid_values(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that invalid color/brightness/etc. values are ignored."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
color_modes = [light.ColorMode.COLOR_TEMP, light.ColorMode.HS]
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == color_modes
expected_features = (
light.LightEntityFeature.FLASH | light.LightEntityFeature.TRANSITION
)
assert state.attributes.get(ATTR_SUPPORTED_FEATURES) is expected_features
assert state.attributes.get("rgb_color") is None
assert state.attributes.get("brightness") is None
assert state.attributes.get("color_temp_kelvin") is None
assert not state.attributes.get(ATTR_ASSUMED_STATE)
# Turn on the light
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON",'
'"color":{"r":255,"g":255,"b":255},'
'"brightness": 255,'
'"color_mode": "color_temp",'
'"color_temp": 100,'
'"effect": "rainbow"}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
# Color converttrd from color_temp to rgb
assert state.attributes.get("rgb_color") == (202, 218, 255)
assert state.attributes.get("brightness") == 255
assert state.attributes.get("color_temp_kelvin") == 10000
# Empty color value
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color":{}, "color_mode": "rgb"}',
)
# Color should not have changed
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") == (202, 218, 255)
# Bad HS color values
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color":{"h":"bad","s":"val"}, "color_mode": "hs"}',
)
# Color should not have changed
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") == (202, 218, 255)
# Bad RGB color values
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color":{"r":"bad","g":"val","b":"test"}, "color_mode": "rgb"}',
)
# Color should not have changed
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") == (202, 218, 255)
# Bad XY color values
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color":{"x":"bad","y":"val"}, "color_mode": "xy"}',
)
# Color should not have changed
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("rgb_color") == (202, 218, 255)
# Bad brightness values
async_fire_mqtt_message(
hass, "test_light_rgb", '{"state":"ON", "brightness": "badValue"}'
)
# Brightness should not have changed
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("brightness") == 255
# Unset color and set a valid color temperature
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color": null, "color_temp": 100, "color_mode": "color_temp"}',
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("color_temp_kelvin") == 10000
# Bad color temperature
async_fire_mqtt_message(
hass,
"test_light_rgb",
'{"state":"ON", "color_temp": "badValue", "color_mode": "color_temp"}',
)
assert (
"Invalid or incomplete color value '{'state': 'ON', 'color_temp': "
"'badValue', 'color_mode': 'color_temp'}' "
"received for entity light.test" in caplog.text
)
# Color temperature should not have changed
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes.get("color_temp_kelvin") == 10000
@pytest.mark.parametrize("hass_config", [DEFAULT_CONFIG])
async def test_availability_when_connection_lost(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test availability after MQTT disconnection."""
await help_test_availability_when_connection_lost(
hass, mqtt_mock_entry, light.DOMAIN
)
@pytest.mark.parametrize("hass_config", [DEFAULT_CONFIG])
async def test_availability_without_topic(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock_entry, light.DOMAIN, DEFAULT_CONFIG
)
async def test_default_availability_payload(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock_entry, light.DOMAIN, DEFAULT_CONFIG
)
async def test_custom_availability_payload(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test availability by custom payload with defined topic."""
await help_test_custom_availability_payload(
hass, mqtt_mock_entry, light.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_via_mqtt_json_message_single_light(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry, light.DOMAIN, DEFAULT_CONFIG
)
@pytest.mark.parametrize(
"hass_config",
[
help_custom_config(
light.DOMAIN,
DEFAULT_CONFIG,
(
{
"unique_id": "very_unique_member_1",
"name": "Part 1",
"default_entity_id": "light.member_1",
},
{
"unique_id": "very_unique_member_2",
"name": "Part 2",
"default_entity_id": "light.member_2",
},
{
"unique_id": "very_unique_group",
"name": "My group",
"default_entity_id": "light.my_group",
"json_attributes_topic": "attr-topic",
"group": [
"very_unique_member_1",
"very_unique_member_2",
"member_3_not_exists",
],
},
),
)
],
)
async def test_setting_attribute_via_mqtt_json_message_light_group(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the setting of attribute via MQTT with JSON payload."""
await mqtt_mock_entry()
async_fire_mqtt_message(hass, "attr-topic", '{ "val": "100" }')
state = hass.states.get("light.my_group")
assert state and state.attributes.get("val") == "100"
assert state.attributes.get("entity_id") == ["light.member_1", "light.member_2"]
assert state.attributes.get("icon") == "mdi:lightbulb-group"
async def test_setting_blocked_attribute_via_mqtt_json_message(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_blocked_attribute_via_mqtt_json_message(
hass,
mqtt_mock_entry,
light.DOMAIN,
DEFAULT_CONFIG,
MQTT_LIGHT_ATTRIBUTES_BLOCKED,
)
async def test_setting_attribute_with_template(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock_entry, light.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock_entry, caplog, light.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_json(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_json(
hass, mqtt_mock_entry, caplog, light.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test update of discovered MQTTAttributes."""
await help_test_discovery_update_attr(
hass, mqtt_mock_entry, light.DOMAIN, DEFAULT_CONFIG
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
light.DOMAIN: [
{
"name": "Test 1",
"schema": "json",
"state_topic": "test-topic",
"command_topic": "test_topic",
"unique_id": "TOTALLY_UNIQUE",
},
{
"name": "Test 2",
"schema": "json",
"state_topic": "test-topic",
"command_topic": "test_topic",
"unique_id": "TOTALLY_UNIQUE",
},
]
}
}
],
)
async def test_unique_id(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test unique id option only creates one light per unique_id."""
await help_test_unique_id(hass, mqtt_mock_entry, light.DOMAIN)
async def test_discovery_removal(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test removal of discovered mqtt_json lights."""
data = '{ "name": "test", "schema": "json", "command_topic": "test_topic" }'
await help_test_discovery_removal(hass, mqtt_mock_entry, light.DOMAIN, data)
async def test_discovery_update_light(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test update of discovered light."""
config1 = {
"name": "Beer",
"schema": "json",
"state_topic": "test_topic",
"command_topic": "test_topic",
}
config2 = {
"name": "Milk",
"schema": "json",
"state_topic": "test_topic",
"command_topic": "test_topic",
}
await help_test_discovery_update(
hass, mqtt_mock_entry, light.DOMAIN, config1, config2
)
async def test_discovery_update_unchanged_light(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test update of discovered light."""
data1 = (
'{ "name": "Beer",'
' "schema": "json",'
' "state_topic": "test_topic",'
' "command_topic": "test_topic" }'
)
with patch(
"homeassistant.components.mqtt.light.schema_json.MqttLightJson.discovery_update"
) as discovery_update:
await help_test_discovery_update_unchanged(
hass, mqtt_mock_entry, light.DOMAIN, data1, discovery_update
)
@pytest.mark.no_fail_on_log_exception
async def test_discovery_broken(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test handling of bad discovery message."""
data1 = '{ "name": "Beer" }'
data2 = (
'{ "name": "Milk",'
' "schema": "json",'
' "state_topic": "test_topic",'
' "command_topic": "test_topic" }'
)
await help_test_discovery_broken(hass, mqtt_mock_entry, light.DOMAIN, data1, data2)
async def test_entity_device_info_with_connection(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT light device registry integration."""
await help_test_entity_device_info_with_connection(
hass,
mqtt_mock_entry,
light.DOMAIN,
DEFAULT_CONFIG,
)
async def test_entity_device_info_with_identifier(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT light device registry integration."""
await help_test_entity_device_info_with_identifier(
hass,
mqtt_mock_entry,
light.DOMAIN,
DEFAULT_CONFIG,
)
async def test_entity_device_info_update(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test device registry update."""
await help_test_entity_device_info_update(
hass,
mqtt_mock_entry,
light.DOMAIN,
DEFAULT_CONFIG,
)
async def test_entity_device_info_remove(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test device registry remove."""
await help_test_entity_device_info_remove(
hass,
mqtt_mock_entry,
light.DOMAIN,
DEFAULT_CONFIG,
)
async def test_entity_id_update_subscriptions(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT subscriptions are managed when entity_id is updated."""
await help_test_entity_id_update_subscriptions(
hass, mqtt_mock_entry, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update_discovery_update(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT discovery update when entity_id is updated."""
await help_test_entity_id_update_discovery_update(
hass, mqtt_mock_entry, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_debug_info_message(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT debug info."""
await help_test_entity_debug_info_message(
hass,
mqtt_mock_entry,
light.DOMAIN,
DEFAULT_CONFIG,
light.SERVICE_TURN_ON,
command_payload='{"state":"ON"}',
state_payload='{"state":"ON"}',
)
@pytest.mark.parametrize(
("hass_config", "min_kelvin", "max_kelvin"),
[
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_max_mireds/set",
"supported_color_modes": ["color_temp"],
"max_mireds": 370, # 2702 Kelvin
}
}
},
2702,
light.DEFAULT_MAX_KELVIN,
),
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_max_mireds/set",
"supported_color_modes": ["color_temp"],
"min_mireds": 150, # 6666 Kelvin
}
}
},
light.DEFAULT_MIN_KELVIN,
6666,
),
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_max_mireds/set",
"supported_color_modes": ["color_temp"],
"min_kelvin": 2702,
}
}
},
2702,
light.DEFAULT_MAX_KELVIN,
),
(
{
mqtt.DOMAIN: {
light.DOMAIN: {
"schema": "json",
"name": "test",
"command_topic": "test_max_mireds/set",
"supported_color_modes": ["color_temp"],
"max_kelvin": 6666,
}
}
},
light.DEFAULT_MIN_KELVIN,
6666,
),
],
)
async def test_min_max_kelvin(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
min_kelvin: int,
max_kelvin: int,
) -> None:
"""Test setting min_color_temp_kelvin and max_color_temp_kelvin."""
await mqtt_mock_entry()
state = hass.states.get("light.test")
assert state.attributes.get("min_color_temp_kelvin") == min_kelvin
assert state.attributes.get("max_color_temp_kelvin") == max_kelvin
@pytest.mark.parametrize(
("service", "topic", "parameters", "payload", "template", "tpl_par", "tpl_output"),
[
(
light.SERVICE_TURN_ON,
"command_topic",
None,
'{"state":"ON"}',
None,
None,
None,
),
(
light.SERVICE_TURN_OFF,
"command_topic",
None,
'{"state":"OFF"}',
None,
None,
None,
),
],
)
async def test_publishing_with_custom_encoding(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
service: str,
topic: str,
parameters: dict[str, Any],
payload: str,
template: str | None,
tpl_par: str,
tpl_output: PublishPayloadType,
) -> None:
"""Test publishing MQTT payload with different encoding."""
domain = light.DOMAIN
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG)
if topic == "effect_command_topic":
config[mqtt.DOMAIN][domain]["effect_list"] = ["random", "color_loop"]
await help_test_publishing_with_custom_encoding(
hass,
mqtt_mock_entry,
caplog,
domain,
config,
service,
topic,
parameters,
payload,
template,
tpl_par=tpl_par,
tpl_output=tpl_output,
)
async def test_reloadable(
hass: HomeAssistant, mqtt_client_mock: MqttMockPahoClient
) -> None:
"""Test reloading the MQTT platform."""
domain = light.DOMAIN
config = DEFAULT_CONFIG
await help_test_reloadable(hass, mqtt_client_mock, domain, config)
@pytest.mark.parametrize(
("topic", "value", "attribute", "attribute_value", "init_payload"),
[
(
"state_topic",
'{ "state": "ON", "brightness": 200, "color_mode":"hs", "color":{"h":180,"s":50} }',
"brightness",
200,
None,
),
],
)
async def test_encoding_subscribable_topics(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
topic: str,
value: str,
attribute: str | None,
attribute_value: Any,
init_payload: tuple[str, str] | None,
) -> None:
"""Test handling of incoming encoded payload."""
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][light.DOMAIN])
config["color_mode"] = True
config["supported_color_modes"] = [
"color_temp",
"hs",
"xy",
"rgb",
"rgbw",
"rgbww",
]
await help_test_encoding_subscribable_topics(
hass,
mqtt_mock_entry,
light.DOMAIN,
config,
topic,
value,
attribute,
attribute_value,
init_payload,
skip_raw_test=True,
)
@pytest.mark.parametrize(
"hass_config",
[DEFAULT_CONFIG, {"mqtt": [DEFAULT_CONFIG["mqtt"]]}],
ids=["platform_key", "listed"],
)
async def test_setup_manual_entity_from_yaml(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test setup manual configured MQTT entity."""
await mqtt_mock_entry()
platform = light.DOMAIN
assert hass.states.get(f"{platform}.test")
@pytest.mark.parametrize(
"hass_config",
[
help_custom_config(
light.DOMAIN,
DEFAULT_CONFIG,
(
{
"effect": True,
"supported_color_modes": [
"color_temp",
"hs",
"xy",
"rgb",
"rgbw",
"rgbww",
"white",
],
"effect_list": ["effect1", "effect2"],
"availability_topic": "availability-topic",
"json_attributes_topic": "json-attributes-topic",
"state_topic": "test-topic",
},
),
)
],
)
@pytest.mark.parametrize(
("topic", "payload1", "payload2"),
[
("test-topic", '{"state":"ON"}', '{"state":"OFF"}'),
("availability-topic", "online", "offline"),
("json-attributes-topic", '{"attr1": "val1"}', '{"attr1": "val2"}'),
(
"test-topic",
'{"state":"ON","effect":"effect1"}',
'{"state":"ON","effect":"effect2"}',
),
(
"test-topic",
'{"state":"ON","brightness":255}',
'{"state":"ON","brightness":96}',
),
(
"test-topic",
'{"state":"ON","brightness":96}',
'{"state":"ON","color_mode":"white","brightness":96}',
),
(
"test-topic",
'{"state":"ON","color_mode":"color_temp", "color_temp": 200}',
'{"state":"ON","color_mode":"color_temp", "color_temp": 2400}',
),
(
"test-topic",
'{"state":"ON","color_mode":"hs", "color": {"h":24.0,"s":100.0}}',
'{"state":"ON","color_mode":"hs", "color": {"h":24.0,"s":90.0}}',
),
(
"test-topic",
'{"state":"ON","color_mode":"xy","color": {"x":0.14,"y":0.131}}',
'{"state":"ON","color_mode":"xy","color": {"x":0.16,"y": 0.100}}',
),
(
"test-topic",
'{"state":"ON","brightness":255,"color_mode":"rgb","color":{"r":128,"g":128,"b":255}}',
'{"state":"ON","brightness":255,"color_mode":"rgb","color": {"r":255,"g":128,"b":255}}',
),
(
"test-topic",
'{"state":"ON","color_mode":"rgbw","color":{"r":128,"g":128,"b":255,"w":128}}',
'{"state":"ON","color_mode":"rgbw","color": {"r":128,"g":128,"b":255,"w":255}}',
),
(
"test-topic",
'{"state":"ON","color_mode":"rgbww","color":{"r":128,"g":128,"b":255,"c":32,"w":128}}',
'{"state":"ON","color_mode":"rgbww","color": {"r":128,"g":128,"b":255,"c":16,"w":128}}',
),
],
)
async def test_skipped_async_ha_write_state(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
topic: str,
payload1: str,
payload2: str,
) -> None:
"""Test a write state command is only called when there is change."""
await mqtt_mock_entry()
await help_test_skipped_async_ha_write_state(hass, topic, payload1, payload2)
Reference in New Issue View Git Blame Copy Permalink