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

Add mqtt text platform (#82884)

Browse Source
This commit is contained in:
Jan Bouwhuis 2022-11-29 20:31:05 +01:00 committed by GitHub
parent b4e30c4033
commit 2785b2b52f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 906 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
homeassistant/components/mqtt/abbreviations.py
View File

@ -169,6 +169,7 @@ ABBREVIATIONS = {
"pr_mode_stat_t": "preset_mode_state_topic",
"pr_mode_val_tpl": "preset_mode_value_template",
"pr_modes": "preset_modes",
"ptrn": "pattern",
"r_tpl": "red_template",
"rel_s": "release_summary",
"rel_u": "release_url",

4
homeassistant/components/mqtt/config_integration.py
View File

@ -32,6 +32,7 @@ from . import (
sensor as sensor_platform,
siren as siren_platform,
switch as switch_platform,
text as text_platform,
update as update_platform,
vacuum as vacuum_platform,
)
@ -130,6 +131,9 @@ PLATFORM_CONFIG_SCHEMA_BASE = vol.Schema(
Platform.SWITCH.value: vol.All(
cv.ensure_list, [switch_platform.PLATFORM_SCHEMA_MODERN] # type: ignore[has-type]
),
Platform.TEXT.value: vol.All(
cv.ensure_list, [text_platform.PLATFORM_SCHEMA_MODERN] # type: ignore[has-type]
),
Platform.UPDATE.value: vol.All(
cv.ensure_list, [update_platform.PLATFORM_SCHEMA_MODERN] # type: ignore[has-type]
),

2
homeassistant/components/mqtt/const.py
View File

@ -101,6 +101,7 @@ PLATFORMS = [
Platform.SENSOR,
Platform.SIREN,
Platform.SWITCH,
Platform.TEXT,
Platform.UPDATE,
Platform.VACUUM,
]
@ -122,6 +123,7 @@ RELOADABLE_PLATFORMS = [
Platform.SENSOR,
Platform.SIREN,
Platform.SWITCH,
Platform.TEXT,
Platform.UPDATE,
Platform.VACUUM,
]

1
homeassistant/components/mqtt/discovery.py
View File

@ -63,6 +63,7 @@ SUPPORTED_COMPONENTS = [
"sensor",
"switch",
"tag",
"text",
"update",
"vacuum",
]

224
homeassistant/components/mqtt/text.py Normal file
View File

@ -0,0 +1,224 @@
"""Support for MQTT text platform."""
from __future__ import annotations
from collections.abc import Callable
import functools
import logging
import re
from typing import Any
import voluptuous as vol
from homeassistant.components import text
from homeassistant.components.text import TextEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_MODE,
CONF_NAME,
CONF_OPTIMISTIC,
CONF_VALUE_TEMPLATE,
MAX_LENGTH_STATE_STATE,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import subscription
from .config import MQTT_RW_SCHEMA
from .const import (
CONF_COMMAND_TEMPLATE,
CONF_COMMAND_TOPIC,
CONF_ENCODING,
CONF_QOS,
CONF_RETAIN,
CONF_STATE_TOPIC,
)
from .debug_info import log_messages
from .mixins import MQTT_ENTITY_COMMON_SCHEMA, MqttEntity, async_setup_entry_helper
from .models import (
MqttCommandTemplate,
MqttValueTemplate,
PublishPayloadType,
ReceiveMessage,
ReceivePayloadType,
)
from .util import get_mqtt_data
_LOGGER = logging.getLogger(__name__)
CONF_MAX = "max"
CONF_MIN = "min"
CONF_PATTERN = "pattern"
DEFAULT_NAME = "MQTT Text"
DEFAULT_OPTIMISTIC = False
DEFAULT_PAYLOAD_RESET = "None"
MQTT_TEXT_ATTRIBUTES_BLOCKED = frozenset(
{
text.ATTR_MAX,
text.ATTR_MIN,
text.ATTR_MODE,
text.ATTR_PATTERN,
}
)
def valid_text_size_configuration(config: ConfigType) -> ConfigType:
"""Validate that the text length configuration is valid, throws if it isn't."""
if config[CONF_MIN] >= config[CONF_MAX]:
raise ValueError("text length min must be >= max")
if config[CONF_MAX] > MAX_LENGTH_STATE_STATE:
raise ValueError(f"max text length must be <= {MAX_LENGTH_STATE_STATE}")
return config
_PLATFORM_SCHEMA_BASE = MQTT_RW_SCHEMA.extend(
{
vol.Optional(CONF_COMMAND_TEMPLATE): cv.template,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_MAX, default=MAX_LENGTH_STATE_STATE): cv.positive_int,
vol.Optional(CONF_MIN, default=0): cv.positive_int,
vol.Optional(CONF_MODE, default=text.TextMode.TEXT): vol.In(
[text.TextMode.TEXT, text.TextMode.PASSWORD]
),
vol.Optional(CONF_OPTIMISTIC, default=DEFAULT_OPTIMISTIC): cv.boolean,
vol.Optional(CONF_PATTERN): cv.is_regex,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
},
).extend(MQTT_ENTITY_COMMON_SCHEMA.schema)
DISCOVERY_SCHEMA = vol.All(
_PLATFORM_SCHEMA_BASE.extend({}, extra=vol.REMOVE_EXTRA),
valid_text_size_configuration,
)
PLATFORM_SCHEMA_MODERN = vol.All(_PLATFORM_SCHEMA_BASE, valid_text_size_configuration)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up MQTT text through configuration.yaml and dynamically through MQTT discovery."""
setup = functools.partial(
_async_setup_entity, hass, async_add_entities, config_entry=config_entry
)
await async_setup_entry_helper(hass, text.DOMAIN, setup, DISCOVERY_SCHEMA)
async def _async_setup_entity(
hass: HomeAssistant,
async_add_entities: AddEntitiesCallback,
config: ConfigType,
config_entry: ConfigEntry,
discovery_data: DiscoveryInfoType | None = None,
) -> None:
"""Set up the MQTT text."""
async_add_entities([MqttTextEntity(hass, config, config_entry, discovery_data)])
class MqttTextEntity(MqttEntity, TextEntity):
"""Representation of the MQTT text entity."""
_attributes_extra_blocked = MQTT_TEXT_ATTRIBUTES_BLOCKED
_entity_id_format = text.ENTITY_ID_FORMAT
_compiled_pattern: re.Pattern[Any] | None
_optimistic: bool
_command_template: Callable[[PublishPayloadType], PublishPayloadType]
_value_template: Callable[[ReceivePayloadType], ReceivePayloadType]
def __init__(
self,
hass: HomeAssistant,
config: ConfigType,
config_entry: ConfigEntry,
discovery_data: DiscoveryInfoType | None = None,
) -> None:
"""Initialize MQTT text entity."""
self._attr_native_value = None
MqttEntity.__init__(self, hass, config, config_entry, discovery_data)
@staticmethod
def config_schema() -> vol.Schema:
"""Return the config schema."""
return DISCOVERY_SCHEMA
def _setup_from_config(self, config: ConfigType) -> None:
"""(Re)Setup the entity."""
self._attr_native_max = config[CONF_MAX]
self._attr_native_min = config[CONF_MIN]
self._attr_mode = config[CONF_MODE]
self._compiled_pattern = config.get(CONF_PATTERN)
self._attr_pattern = (
self._compiled_pattern.pattern if self._compiled_pattern else None
)
self._command_template = MqttCommandTemplate(
config.get(CONF_COMMAND_TEMPLATE),
entity=self,
).async_render
self._value_template = MqttValueTemplate(
config.get(CONF_VALUE_TEMPLATE),
entity=self,
).async_render_with_possible_json_value
optimistic: bool = config[CONF_OPTIMISTIC]
self._optimistic = optimistic or config.get(CONF_STATE_TOPIC) is None
def _prepare_subscribe_topics(self) -> None:
"""(Re)Subscribe to topics."""
topics: dict[str, Any] = {}
def add_subscription(
topics: dict[str, Any], topic: str, msg_callback: Callable
) -> None:
if self._config.get(topic) is not None:
topics[topic] = {
"topic": self._config[topic],
"msg_callback": msg_callback,
"qos": self._config[CONF_QOS],
"encoding": self._config[CONF_ENCODING] or None,
}
@callback
@log_messages(self.hass, self.entity_id)
def handle_state_message_received(msg: ReceiveMessage) -> None:
"""Handle receiving state message via MQTT."""
payload = str(self._value_template(msg.payload))
self._attr_native_value = payload
get_mqtt_data(self.hass).state_write_requests.write_state_request(self)
add_subscription(topics, CONF_STATE_TOPIC, handle_state_message_received)
self._sub_state = subscription.async_prepare_subscribe_topics(
self.hass, self._sub_state, topics
)
async def _subscribe_topics(self) -> None:
"""(Re)Subscribe to topics."""
await subscription.async_subscribe_topics(self.hass, self._sub_state)
@property
def assumed_state(self) -> bool:
"""Return true if we do optimistic updates."""
return self._optimistic
async def async_set_value(self, value: str) -> None:
"""Change the text."""
payload = self._command_template(value)
await self.async_publish(
self._config[CONF_COMMAND_TOPIC],
payload,
self._config[CONF_QOS],
self._config[CONF_RETAIN],
self._config[CONF_ENCODING],
)
if self._optimistic:
self._attr_native_value = value
self.async_write_ha_state()

674
tests/components/mqtt/test_text.py Normal file
View File

@ -0,0 +1,674 @@
"""The tests for the MQTT text platform."""
from __future__ import annotations
from unittest.mock import patch
import pytest
from homeassistant.components import mqtt, text
from homeassistant.const import (
ATTR_ASSUMED_STATE,
ATTR_ENTITY_ID,
STATE_UNKNOWN,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .test_common import (
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_setup_manual_entity_from_yaml,
help_test_unique_id,
help_test_unload_config_entry_with_platform,
help_test_update_with_json_attrs_bad_json,
help_test_update_with_json_attrs_not_dict,
)
from tests.common import async_fire_mqtt_message
DEFAULT_CONFIG = {
mqtt.DOMAIN: {text.DOMAIN: {"name": "test", "command_topic": "test-topic"}}
}
@pytest.fixture(autouse=True)
def text_platform_only():
"""Only setup the text platform to speed up tests."""
with patch("homeassistant.components.mqtt.PLATFORMS", [Platform.TEXT]):
yield
async def async_set_value(
hass: HomeAssistant, entity_id: str, value: str | None
) -> None:
"""Set input_text to value."""
await hass.services.async_call(
text.DOMAIN,
text.SERVICE_SET_VALUE,
{ATTR_ENTITY_ID: entity_id, text.ATTR_VALUE: value},
blocking=True,
)
async def test_controlling_state_via_topic(
hass: HomeAssistant, mqtt_mock_entry_with_yaml_config
) -> None:
"""Test the controlling state via topic."""
assert await async_setup_component(
hass,
mqtt.DOMAIN,
{
mqtt.DOMAIN: {
text.DOMAIN: {
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"mode": "password",
}
}
},
)
await hass.async_block_till_done()
await mqtt_mock_entry_with_yaml_config()
state = hass.states.get("text.test")
assert state.state == STATE_UNKNOWN
assert state.attributes[text.ATTR_MODE] == "password"
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "state-topic", "some state")
state = hass.states.get("text.test")
assert state.state == "some state"
async_fire_mqtt_message(hass, "state-topic", "some other state")
state = hass.states.get("text.test")
assert state.state == "some other state"
async_fire_mqtt_message(hass, "state-topic", "")
state = hass.states.get("text.test")
assert state.state == ""
async def test_controlling_validation_state_via_topic(
hass, mqtt_mock_entry_with_yaml_config
) -> None:
"""Test the validation of a received state."""
assert await async_setup_component(
hass,
mqtt.DOMAIN,
{
mqtt.DOMAIN: {
text.DOMAIN: {
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"mode": "text",
"min": 2,
"max": 10,
"pattern": "(y|n)",
}
}
},
)
await hass.async_block_till_done()
await mqtt_mock_entry_with_yaml_config()
state = hass.states.get("text.test")
assert state.state == STATE_UNKNOWN
assert state.attributes[text.ATTR_MODE] == "text"
async_fire_mqtt_message(hass, "state-topic", "yes")
state = hass.states.get("text.test")
assert state.state == "yes"
# test pattern error
with pytest.raises(ValueError):
async_fire_mqtt_message(hass, "state-topic", "other")
await hass.async_block_till_done()
state = hass.states.get("text.test")
assert state.state == "yes"
# test text size to large
with pytest.raises(ValueError):
async_fire_mqtt_message(hass, "state-topic", "yesyesyesyes")
await hass.async_block_till_done()
state = hass.states.get("text.test")
assert state.state == "yes"
# test text size to small
with pytest.raises(ValueError):
async_fire_mqtt_message(hass, "state-topic", "y")
await hass.async_block_till_done()
state = hass.states.get("text.test")
assert state.state == "yes"
async_fire_mqtt_message(hass, "state-topic", "no")
await hass.async_block_till_done()
state = hass.states.get("text.test")
assert state.state == "no"
async def test_attribute_validation_max_greater_then_min(hass) -> None:
"""Test the validation of min and max configuration attributes."""
assert not await async_setup_component(
hass,
mqtt.DOMAIN,
{
mqtt.DOMAIN: {
text.DOMAIN: {
"name": "test",
"command_topic": "command-topic",
"min": 20,
"max": 10,
}
}
},
)
async def test_attribute_validation_max_not_greater_then_max_state_length(hass) -> None:
"""Test the max value of of max configuration attribute."""
assert not await async_setup_component(
hass,
mqtt.DOMAIN,
{
mqtt.DOMAIN: {
text.DOMAIN: {
"name": "test",
"command_topic": "command-topic",
"min": 20,
"max": 257,
}
}
},
)
async def test_sending_mqtt_commands_and_optimistic(
hass, mqtt_mock_entry_with_yaml_config
):
"""Test the sending MQTT commands in optimistic mode."""
assert await async_setup_component(
hass,
mqtt.DOMAIN,
{
mqtt.DOMAIN: {
text.DOMAIN: {
"name": "test",
"command_topic": "command-topic",
"qos": "2",
}
}
},
)
await hass.async_block_till_done()
mqtt_mock = await mqtt_mock_entry_with_yaml_config()
state = hass.states.get("text.test")
assert state.state == STATE_UNKNOWN
assert state.attributes.get(ATTR_ASSUMED_STATE)
await async_set_value(hass, "text.test", "some other state")
await hass.async_block_till_done()
mqtt_mock.async_publish.assert_called_once_with(
"command-topic", "some other state", 2, False
)
mqtt_mock.async_publish.reset_mock()
state = hass.states.get("text.test")
assert state.state == "some other state"
await async_set_value(hass, "text.test", "some new state")
mqtt_mock.async_publish.assert_called_once_with(
"command-topic", "some new state", 2, False
)
state = hass.states.get("text.test")
assert state.state == "some new state"
async def test_set_text_validation(hass, mqtt_mock_entry_with_yaml_config):
"""Test the initial state in optimistic mode."""
assert await async_setup_component(
hass,
mqtt.DOMAIN,
{
mqtt.DOMAIN: {
text.DOMAIN: {
"name": "test",
"command_topic": "command-topic",
"mode": "text",
"min": 2,
"max": 10,
"pattern": "(y|n)",
}
}
},
)
await hass.async_block_till_done()
await mqtt_mock_entry_with_yaml_config()
state = hass.states.get("text.test")
assert state.state == STATE_UNKNOWN
assert state.attributes.get(ATTR_ASSUMED_STATE)
# text too long
with pytest.raises(ValueError):
await async_set_value(hass, "text.test", "yes yes yes yes")
# text too short
with pytest.raises(ValueError):
await async_set_value(hass, "text.test", "y")
# text not matching pattern
with pytest.raises(ValueError):
await async_set_value(hass, "text.test", "other")
await async_set_value(hass, "text.test", "no")
state = hass.states.get("text.test")
assert state.state == "no"
async def test_availability_when_connection_lost(
hass, mqtt_mock_entry_with_yaml_config
):
"""Test availability after MQTT disconnection."""
await help_test_availability_when_connection_lost(
hass, mqtt_mock_entry_with_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_availability_without_topic(hass, mqtt_mock_entry_with_yaml_config):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock_entry_with_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_default_availability_payload(hass, mqtt_mock_entry_with_yaml_config):
"""Test availability by default payload with defined topic."""
config = {
mqtt.DOMAIN: {
text.DOMAIN: {
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
}
}
}
await help_test_default_availability_payload(
hass,
mqtt_mock_entry_with_yaml_config,
text.DOMAIN,
config,
True,
"state-topic",
"some state",
)
async def test_custom_availability_payload(hass, mqtt_mock_entry_with_yaml_config):
"""Test availability by custom payload with defined topic."""
config = {
mqtt.DOMAIN: {
text.DOMAIN: {
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
}
}
}
await help_test_custom_availability_payload(
hass,
mqtt_mock_entry_with_yaml_config,
text.DOMAIN,
config,
True,
"state-topic",
"1",
)
async def test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry_with_yaml_config
):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry_with_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_blocked_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry_no_yaml_config
):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_blocked_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry_no_yaml_config, text.DOMAIN, DEFAULT_CONFIG, {}
)
async def test_setting_attribute_with_template(hass, mqtt_mock_entry_with_yaml_config):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock_entry_with_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(
hass, mqtt_mock_entry_with_yaml_config, caplog
):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass,
mqtt_mock_entry_with_yaml_config,
caplog,
text.DOMAIN,
DEFAULT_CONFIG,
)
async def test_update_with_json_attrs_bad_json(
hass, mqtt_mock_entry_with_yaml_config, caplog
):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_json(
hass,
mqtt_mock_entry_with_yaml_config,
caplog,
text.DOMAIN,
DEFAULT_CONFIG,
)
async def test_discovery_update_attr(hass, mqtt_mock_entry_no_yaml_config, caplog):
"""Test update of discovered MQTTAttributes."""
await help_test_discovery_update_attr(
hass,
mqtt_mock_entry_no_yaml_config,
caplog,
text.DOMAIN,
DEFAULT_CONFIG,
)
async def test_unique_id(hass, mqtt_mock_entry_with_yaml_config):
"""Test unique id option only creates one text per unique_id."""
config = {
mqtt.DOMAIN: {
text.DOMAIN: [
{
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "command-topic",
"unique_id": "TOTALLY_UNIQUE",
},
{
"name": "Test 2",
"state_topic": "test-topic",
"command_topic": "command-topic",
"unique_id": "TOTALLY_UNIQUE",
},
]
}
}
await help_test_unique_id(
hass, mqtt_mock_entry_with_yaml_config, text.DOMAIN, config
)
async def test_discovery_removal_text(hass, mqtt_mock_entry_no_yaml_config, caplog):
"""Test removal of discovered text entity."""
data = (
'{ "name": "test",'
' "state_topic": "test_topic",'
' "command_topic": "test_topic" }'
)
await help_test_discovery_removal(
hass, mqtt_mock_entry_no_yaml_config, caplog, text.DOMAIN, data
)
async def test_discovery_text_update(hass, mqtt_mock_entry_no_yaml_config, caplog):
"""Test update of discovered text entity."""
config1 = {
"name": "Beer",
"command_topic": "command-topic",
"state_topic": "state-topic",
}
config2 = {
"name": "Milk",
"command_topic": "command-topic",
"state_topic": "state-topic",
}
await help_test_discovery_update(
hass, mqtt_mock_entry_no_yaml_config, caplog, text.DOMAIN, config1, config2
)
async def test_discovery_update_unchanged_update(
hass, mqtt_mock_entry_no_yaml_config, caplog
):
"""Test update of discovered update."""
data1 = '{ "name": "Beer", "state_topic": "text-topic", "command_topic": "command-topic"}'
with patch(
"homeassistant.components.mqtt.text.MqttTextEntity.discovery_update"
) as discovery_update:
await help_test_discovery_update_unchanged(
hass,
mqtt_mock_entry_no_yaml_config,
caplog,
text.DOMAIN,
data1,
discovery_update,
)
async def test_discovery_update_text(hass, mqtt_mock_entry_no_yaml_config, caplog):
"""Test update of discovered text entity."""
config1 = {"name": "Beer", "command_topic": "cmd-topic1"}
config2 = {"name": "Milk", "command_topic": "cmd-topic2"}
await help_test_discovery_update(
hass, mqtt_mock_entry_no_yaml_config, caplog, text.DOMAIN, config1, config2
)
async def test_discovery_update_unchanged_climate(
hass, mqtt_mock_entry_no_yaml_config, caplog
):
"""Test update of discovered text entity."""
data1 = '{ "name": "Beer", "command_topic": "cmd-topic" }'
with patch(
"homeassistant.components.mqtt.text.MqttTextEntity.discovery_update"
) as discovery_update:
await help_test_discovery_update_unchanged(
hass,
mqtt_mock_entry_no_yaml_config,
caplog,
text.DOMAIN,
data1,
discovery_update,
)
@pytest.mark.no_fail_on_log_exception
async def test_discovery_broken(hass, mqtt_mock_entry_no_yaml_config, caplog):
"""Test handling of bad discovery message."""
data1 = '{ "name": "Beer" }'
data2 = (
'{ "name": "Milk",'
' "state_topic": "test_topic",'
' "command_topic": "test_topic" }'
)
await help_test_discovery_broken(
hass, mqtt_mock_entry_no_yaml_config, caplog, text.DOMAIN, data1, data2
)
async def test_entity_device_info_with_connection(hass, mqtt_mock_entry_no_yaml_config):
"""Test MQTT text device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock_entry_no_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock_entry_no_yaml_config):
"""Test MQTT text device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock_entry_no_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock_entry_no_yaml_config):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock_entry_no_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock_entry_no_yaml_config):
"""Test device registry remove."""
await help_test_entity_device_info_remove(
hass, mqtt_mock_entry_no_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update_subscriptions(hass, mqtt_mock_entry_with_yaml_config):
"""Test MQTT subscriptions are managed when entity_id is updated."""
await help_test_entity_id_update_subscriptions(
hass, mqtt_mock_entry_with_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update_discovery_update(hass, mqtt_mock_entry_no_yaml_config):
"""Test MQTT discovery update when entity_id is updated."""
await help_test_entity_id_update_discovery_update(
hass, mqtt_mock_entry_no_yaml_config, text.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_debug_info_message(hass, mqtt_mock_entry_no_yaml_config):
"""Test MQTT debug info."""
await help_test_entity_debug_info_message(
hass, mqtt_mock_entry_no_yaml_config, text.DOMAIN, DEFAULT_CONFIG, None
)
@pytest.mark.parametrize(
"service,topic,parameters,payload,template",
[
(
text.SERVICE_SET_VALUE,
"command_topic",
{text.ATTR_VALUE: "some text"},
"some text",
"command_template",
),
],
)
async def test_publishing_with_custom_encoding(
hass,
mqtt_mock_entry_with_yaml_config,
caplog,
service,
topic,
parameters,
payload,
template,
):
"""Test publishing MQTT payload with different encoding."""
domain = text.DOMAIN
config = DEFAULT_CONFIG
await help_test_publishing_with_custom_encoding(
hass,
mqtt_mock_entry_with_yaml_config,
caplog,
domain,
config,
service,
topic,
parameters,
payload,
template,
)
async def test_reloadable(hass, mqtt_mock_entry_with_yaml_config, caplog, tmp_path):
"""Test reloading the MQTT platform."""
domain = text.DOMAIN
config = DEFAULT_CONFIG
await help_test_reloadable(
hass, mqtt_mock_entry_with_yaml_config, caplog, tmp_path, domain, config
)
@pytest.mark.parametrize(
"topic,value,attribute,attribute_value",
[
("state_topic", "some text", None, "some text"),
],
)
async def test_encoding_subscribable_topics(
hass,
mqtt_mock_entry_with_yaml_config,
caplog,
topic,
value,
attribute,
attribute_value,
):
"""Test handling of incoming encoded payload."""
await help_test_encoding_subscribable_topics(
hass,
mqtt_mock_entry_with_yaml_config,
caplog,
text.DOMAIN,
DEFAULT_CONFIG[mqtt.DOMAIN][text.DOMAIN],
topic,
value,
attribute,
attribute_value,
)
async def test_setup_manual_entity_from_yaml(hass):
"""Test setup manual configured MQTT entity."""
platform = text.DOMAIN
await help_test_setup_manual_entity_from_yaml(hass, DEFAULT_CONFIG)
assert hass.states.get(f"{platform}.test")
async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
"""Test unloading the config entry."""
domain = text.DOMAIN
config = DEFAULT_CONFIG
await help_test_unload_config_entry_with_platform(
hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
)