Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add MQTT event entity platform (#96876) 

Co-authored-by: Franck Nijhof <frenck@frenck.nl>
pull/97088/head
Jan Bouwhuis 2023-07-23 14:42:14 +02:00 committed by GitHub
parent 26152adb23
commit 1b8e03bb66
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 905 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -25,7 +25,7 @@ from homeassistant.const import (
)
from homeassistant.core import HassJob, HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import TemplateError, Unauthorized
from homeassistant.helpers import config_validation as cv, event, template
from homeassistant.helpers import config_validation as cv, event as ev, template
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import async_get_platforms
@ -340,7 +340,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
unsub()
await hass.async_add_executor_job(write_dump)
event.async_call_later(hass, call.data["duration"], finish_dump)
ev.async_call_later(hass, call.data["duration"], finish_dump)
hass.services.async_register(
DOMAIN,

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

@ -60,6 +60,7 @@ ABBREVIATIONS = {
"ent_pic": "entity_picture",
"err_t": "error_topic",
"err_tpl": "error_template",
"evt_typ": "event_types",
"fanspd_t": "fan_speed_topic",
"fanspd_tpl": "fan_speed_template",
"fanspd_lst": "fan_speed_list",

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

@ -22,6 +22,7 @@ from . import (
climate as climate_platform,
cover as cover_platform,
device_tracker as device_tracker_platform,
event as event_platform,
fan as fan_platform,
humidifier as humidifier_platform,
image as image_platform,
@ -82,6 +83,10 @@ CONFIG_SCHEMA_BASE = vol.Schema(
cv.ensure_list,
[device_tracker_platform.PLATFORM_SCHEMA_MODERN], # type: ignore[has-type]
),
Platform.EVENT.value: vol.All(
cv.ensure_list,
[event_platform.PLATFORM_SCHEMA_MODERN], # type: ignore[has-type]
),
Platform.FAN.value: vol.All(
cv.ensure_list,
[fan_platform.PLATFORM_SCHEMA_MODERN], # type: ignore[has-type]

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

@ -112,6 +112,7 @@ PLATFORMS = [
Platform.CAMERA,
Platform.CLIMATE,
Platform.DEVICE_TRACKER,
Platform.EVENT,
Platform.COVER,
Platform.FAN,
Platform.HUMIDIFIER,
@ -138,6 +139,7 @@ RELOADABLE_PLATFORMS = [
Platform.CLIMATE,
Platform.COVER,
Platform.DEVICE_TRACKER,
Platform.EVENT,
Platform.FAN,
Platform.HUMIDIFIER,
Platform.IMAGE,

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

@ -52,6 +52,7 @@ SUPPORTED_COMPONENTS = [
"cover",
"device_automation",
"device_tracker",
"event",
"fan",
"humidifier",
"image",

221
homeassistant/components/mqtt/event.py Normal file
View File

@ -0,0 +1,221 @@
"""Support for MQTT events."""
from __future__ import annotations
from collections.abc import Callable
import functools
import logging
from typing import Any
import voluptuous as vol
from homeassistant.components import event
from homeassistant.components.event import (
ENTITY_ID_FORMAT,
EventDeviceClass,
EventEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_DEVICE_CLASS,
CONF_NAME,
CONF_VALUE_TEMPLATE,
)
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.json import JSON_DECODE_EXCEPTIONS, json_loads_object
from . import subscription
from .config import MQTT_RO_SCHEMA
from .const import (
CONF_ENCODING,
CONF_QOS,
CONF_STATE_TOPIC,
PAYLOAD_EMPTY_JSON,
PAYLOAD_NONE,
)
from .debug_info import log_messages
from .mixins import (
MQTT_ENTITY_COMMON_SCHEMA,
MqttEntity,
async_setup_entry_helper,
)
from .models import (
MqttValueTemplate,
PayloadSentinel,
ReceiveMessage,
ReceivePayloadType,
)
from .util import get_mqtt_data
_LOGGER = logging.getLogger(__name__)
CONF_EVENT_TYPES = "event_types"
MQTT_EVENT_ATTRIBUTES_BLOCKED = frozenset(
{
event.ATTR_EVENT_TYPE,
event.ATTR_EVENT_TYPES,
}
)
DEFAULT_NAME = "MQTT Event"
DEFAULT_FORCE_UPDATE = False
DEVICE_CLASS_SCHEMA = vol.All(vol.Lower, vol.Coerce(EventDeviceClass))
_PLATFORM_SCHEMA_BASE = MQTT_RO_SCHEMA.extend(
{
vol.Optional(CONF_DEVICE_CLASS): DEVICE_CLASS_SCHEMA,
vol.Optional(CONF_NAME): vol.Any(None, cv.string),
vol.Required(CONF_EVENT_TYPES): vol.All(cv.ensure_list, [cv.string]),
}
).extend(MQTT_ENTITY_COMMON_SCHEMA.schema)
PLATFORM_SCHEMA_MODERN = vol.All(
_PLATFORM_SCHEMA_BASE,
)
DISCOVERY_SCHEMA = vol.All(
_PLATFORM_SCHEMA_BASE.extend({}, extra=vol.REMOVE_EXTRA),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up MQTT event through YAML and through MQTT discovery."""
setup = functools.partial(
_async_setup_entity, hass, async_add_entities, config_entry=config_entry
)
await async_setup_entry_helper(hass, event.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 MQTT event."""
async_add_entities([MqttEvent(hass, config, config_entry, discovery_data)])
class MqttEvent(MqttEntity, EventEntity):
"""Representation of an event that can be updated using MQTT."""
_default_name = DEFAULT_NAME
_entity_id_format = ENTITY_ID_FORMAT
_attributes_extra_blocked = MQTT_EVENT_ATTRIBUTES_BLOCKED
_template: Callable[[ReceivePayloadType, PayloadSentinel], ReceivePayloadType]
def __init__(
self,
hass: HomeAssistant,
config: ConfigType,
config_entry: ConfigEntry,
discovery_data: DiscoveryInfoType | None,
) -> None:
"""Initialize the sensor."""
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_device_class = config.get(CONF_DEVICE_CLASS)
self._attr_event_types = config[CONF_EVENT_TYPES]
self._template = MqttValueTemplate(
self._config.get(CONF_VALUE_TEMPLATE), entity=self
).async_render_with_possible_json_value
def _prepare_subscribe_topics(self) -> None:
"""(Re)Subscribe to topics."""
topics: dict[str, dict[str, Any]] = {}
@callback
@log_messages(self.hass, self.entity_id)
def message_received(msg: ReceiveMessage) -> None:
"""Handle new MQTT messages."""
event_attributes: dict[str, Any] = {}
event_type: str
payload = self._template(msg.payload, PayloadSentinel.DEFAULT)
if (
not payload
or payload is PayloadSentinel.DEFAULT
or payload == PAYLOAD_NONE
or payload == PAYLOAD_EMPTY_JSON
):
_LOGGER.debug(
"Ignoring empty payload '%s' after rendering for topic %s",
payload,
msg.topic,
)
return
try:
event_attributes = json_loads_object(payload)
event_type = str(event_attributes.pop(event.ATTR_EVENT_TYPE))
_LOGGER.debug(
(
"JSON event data detected after processing payload '%s' on"
" topic %s, type %s, attributes %s"
),
payload,
msg.topic,
event_type,
event_attributes,
)
except KeyError:
_LOGGER.warning(
(
"`event_type` missing in JSON event payload, "
" '%s' on topic %s"
),
payload,
msg.topic,
)
return
except JSON_DECODE_EXCEPTIONS:
_LOGGER.warning(
(
"No valid JSON event payload detected, "
"value after processing payload"
" '%s' on topic %s"
),
payload,
msg.topic,
)
return
try:
self._trigger_event(event_type, event_attributes)
except ValueError:
_LOGGER.warning(
"Invalid event type %s for %s received on topic %s, payload %s",
event_type,
self.entity_id,
msg.topic,
payload,
)
return
get_mqtt_data(self.hass).state_write_requests.write_state_request(self)
topics["state_topic"] = {
"topic": self._config[CONF_STATE_TOPIC],
"msg_callback": message_received,
"qos": self._config[CONF_QOS],
"encoding": self._config[CONF_ENCODING] or None,
}
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)

673
tests/components/mqtt/test_event.py Normal file
View File

@ -0,0 +1,673 @@
"""The tests for the MQTT event platform."""
import copy
import json
from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components import event, mqtt
from homeassistant.components.mqtt.event import MQTT_EVENT_ATTRIBUTES_BLOCKED
from homeassistant.const import (
STATE_UNKNOWN,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .test_common import (
help_test_availability_when_connection_lost,
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_list_payload,
help_test_default_availability_list_payload_all,
help_test_default_availability_list_payload_any,
help_test_default_availability_list_single,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update_attr,
help_test_discovery_update_availability,
help_test_entity_category,
help_test_entity_debug_info,
help_test_entity_debug_info_message,
help_test_entity_debug_info_remove,
help_test_entity_debug_info_update_entity_id,
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_disabled_by_default,
help_test_entity_id_update_discovery_update,
help_test_entity_id_update_subscriptions,
help_test_entity_name,
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_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,
)
from tests.typing import MqttMockHAClientGenerator, MqttMockPahoClient
DEFAULT_CONFIG = {
mqtt.DOMAIN: {
event.DOMAIN: {
"name": "test",
"state_topic": "test-topic",
"event_types": ["press"],
}
}
}
@pytest.fixture(autouse=True)
def event_platform_only():
"""Only setup the event platform to speed up tests."""
with patch("homeassistant.components.mqtt.PLATFORMS", [Platform.EVENT]):
yield
@pytest.mark.freeze_time("2023-08-01 00:00:00+00:00")
@pytest.mark.parametrize("hass_config", [DEFAULT_CONFIG])
async def test_setting_event_value_via_mqtt_message(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the an MQTT event with attributes."""
await mqtt_mock_entry()
async_fire_mqtt_message(
hass, "test-topic", '{"event_type": "press", "duration": "short" }'
)
state = hass.states.get("event.test")
assert state.state == "2023-08-01T00:00:00.000+00:00"
assert state.attributes.get("duration") == "short"
@pytest.mark.freeze_time("2023-08-01 00:00:00+00:00")
@pytest.mark.parametrize("hass_config", [DEFAULT_CONFIG])
@pytest.mark.parametrize(
("message", "log"),
[
(
'{"event_type": "press", "duration": "short" ',
"No valid JSON event payload detected",
),
('{"event_type": "invalid", "duration": "short" }', "Invalid event type"),
('{"event_type": 2, "duration": "short" }', "Invalid event type"),
('{"event_type": null, "duration": "short" }', "Invalid event type"),
(
'{"event": "press", "duration": "short" }',
"`event_type` missing in JSON event payload",
),
],
)
async def test_setting_event_value_with_invalid_payload(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
message: str,
log: str,
) -> None:
"""Test the an MQTT event with attributes."""
await mqtt_mock_entry()
async_fire_mqtt_message(hass, "test-topic", message)
state = hass.states.get("event.test")
assert state is not None
assert state.state == STATE_UNKNOWN
assert log in caplog.text
@pytest.mark.freeze_time("2023-08-01 00:00:00+00:00")
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
event.DOMAIN: {
"name": "test",
"state_topic": "test-topic",
"event_types": ["press"],
"value_template": '{"event_type": "press", "val": "{{ value_json.val | is_defined }}", "par": "{{ value_json.par }}"}',
}
}
}
],
)
async def test_setting_event_value_via_mqtt_json_message_and_default_current_state(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test processing an event via MQTT with fall back to current state."""
await mqtt_mock_entry()
async_fire_mqtt_message(
hass, "test-topic", '{ "val": "valcontent", "par": "parcontent" }'
)
state = hass.states.get("event.test")
assert state.state == "2023-08-01T00:00:00.000+00:00"
assert state.attributes.get("val") == "valcontent"
assert state.attributes.get("par") == "parcontent"
freezer.move_to("2023-08-01 00:00:10+00:00")
async_fire_mqtt_message(hass, "test-topic", '{ "par": "invalidcontent" }')
state = hass.states.get("event.test")
assert state.state == "2023-08-01T00:00:00.000+00:00"
assert state.attributes.get("val") == "valcontent"
assert state.attributes.get("par") == "parcontent"
@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, event.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, event.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, event.DOMAIN, DEFAULT_CONFIG
)
async def test_default_availability_list_payload(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test availability by default payload with defined topic."""
await help_test_default_availability_list_payload(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
async def test_default_availability_list_payload_all(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test availability by default payload with defined topic."""
await help_test_default_availability_list_payload_all(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
async def test_default_availability_list_payload_any(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test availability by default payload with defined topic."""
await help_test_default_availability_list_payload_any(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
async def test_default_availability_list_single(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test availability list and availability_topic are mutually exclusive."""
await help_test_default_availability_list_single(
hass, caplog, event.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, event.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_availability(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test availability discovery update."""
await help_test_discovery_update_availability(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
event.DOMAIN: {
"name": "test",
"state_topic": "test-topic",
"event_types": ["press"],
"device_class": "foobarnotreal",
}
}
}
],
)
async def test_invalid_device_class(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test device_class option with invalid value."""
with pytest.raises(AssertionError):
await mqtt_mock_entry()
assert (
"Invalid config for [mqtt]: expected EventDeviceClass or one of" in caplog.text
)
async def test_setting_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_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
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,
event.DOMAIN,
DEFAULT_CONFIG,
MQTT_EVENT_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, event.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,
event.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,
event.DOMAIN,
DEFAULT_CONFIG,
)
async def test_discovery_update_attr(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test update of discovered MQTTAttributes."""
await help_test_discovery_update_attr(
hass,
mqtt_mock_entry,
caplog,
event.DOMAIN,
DEFAULT_CONFIG,
)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
event.DOMAIN: [
{
"name": "Test 1",
"state_topic": "test-topic",
"event_types": ["press"],
"unique_id": "TOTALLY_UNIQUE",
},
{
"name": "Test 2",
"state_topic": "test-topic",
"event_types": ["press"],
"unique_id": "TOTALLY_UNIQUE",
},
]
}
}
],
)
async def test_unique_id(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test unique id option only creates one event per unique_id."""
await help_test_unique_id(hass, mqtt_mock_entry, event.DOMAIN)
async def test_discovery_removal_event(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test removal of discovered event."""
data = '{ "name": "test", "state_topic": "test_topic", "event_types": ["press"] }'
await help_test_discovery_removal(hass, mqtt_mock_entry, caplog, event.DOMAIN, data)
async def test_discovery_update_event_template(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test update of discovered mqtt event template."""
await mqtt_mock_entry()
config = {"name": "test", "state_topic": "test_topic", "event_types": ["press"]}
config1 = copy.deepcopy(config)
config2 = copy.deepcopy(config)
config1["name"] = "Beer"
config2["name"] = "Milk"
config1["state_topic"] = "event/state1"
config2["state_topic"] = "event/state1"
config1[
"value_template"
] = '{"event_type": "press", "val": "{{ value_json.val | int }}"}'
config2[
"value_template"
] = '{"event_type": "press", "val": "{{ value_json.val | int * 2 }}"}'
async_fire_mqtt_message(hass, "homeassistant/event/bla/config", json.dumps(config1))
await hass.async_block_till_done()
async_fire_mqtt_message(hass, "event/state1", '{"val":100}')
await hass.async_block_till_done()
state = hass.states.get("event.beer")
assert state is not None
assert state.attributes.get("val") == "100"
async_fire_mqtt_message(hass, "homeassistant/event/bla/config", json.dumps(config2))
await hass.async_block_till_done()
async_fire_mqtt_message(hass, "event/state1", '{"val":100}')
await hass.async_block_till_done()
state = hass.states.get("event.beer")
assert state is not None
assert state.attributes.get("val") == "200"
@pytest.mark.no_fail_on_log_exception
async def test_discovery_broken(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test handling of bad discovery message."""
data1 = '{ "name": "Beer", "state_topic": "test_topic#", "event_types": ["press"] }'
data2 = '{ "name": "Milk", "state_topic": "test_topic", "event_types": ["press"] }'
await help_test_discovery_broken(
hass, mqtt_mock_entry, caplog, event.DOMAIN, data1, data2
)
async def test_entity_device_info_with_connection(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT event device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT event device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock_entry, event.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, event.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, event.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, event.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, event.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_hub(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT event device registry integration."""
await mqtt_mock_entry()
registry = dr.async_get(hass)
hub = registry.async_get_or_create(
config_entry_id="123",
connections=set(),
identifiers={("mqtt", "hub-id")},
manufacturer="manufacturer",
model="hub",
)
data = json.dumps(
{
"name": "Test 1",
"state_topic": "test-topic",
"event_types": ["press"],
"device": {"identifiers": ["helloworld"], "via_device": "hub-id"},
"unique_id": "veryunique",
}
)
async_fire_mqtt_message(hass, "homeassistant/event/bla/config", data)
await hass.async_block_till_done()
device = registry.async_get_device(identifiers={("mqtt", "helloworld")})
assert device is not None
assert device.via_device_id == hub.id
async def test_entity_debug_info(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT event debug info."""
await help_test_entity_debug_info(
hass, mqtt_mock_entry, event.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, event.DOMAIN, DEFAULT_CONFIG, None
)
async def test_entity_debug_info_remove(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT event debug info."""
await help_test_entity_debug_info_remove(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_debug_info_update_entity_id(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test MQTT event debug info."""
await help_test_entity_debug_info_update_entity_id(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_disabled_by_default(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test entity disabled by default."""
await help_test_entity_disabled_by_default(
hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG
)
@pytest.mark.no_fail_on_log_exception
async def test_entity_category(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test entity category."""
await help_test_entity_category(hass, mqtt_mock_entry, event.DOMAIN, DEFAULT_CONFIG)
@pytest.mark.parametrize(
"hass_config",
[
{
mqtt.DOMAIN: {
event.DOMAIN: {
"name": "test",
"state_topic": "test-topic",
"event_types": ["press"],
"value_template": '{ "event_type": "press", "val": \
{% if state_attr(entity_id, "friendly_name") == "test" %} \
"{{ value | int + 1 }}" \
{% else %} \
"{{ value }}" \
{% endif %}}',
}
}
}
],
)
async def test_value_template_with_entity_id(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator
) -> None:
"""Test the access to attributes in value_template via the entity_id."""
await mqtt_mock_entry()
async_fire_mqtt_message(hass, "test-topic", "100")
state = hass.states.get("event.test")
assert state.attributes.get("val") == "101"
async def test_reloadable(
hass: HomeAssistant,
mqtt_client_mock: MqttMockPahoClient,
) -> None:
"""Test reloading the MQTT platform."""
domain = event.DOMAIN
config = DEFAULT_CONFIG
await help_test_reloadable(hass, mqtt_client_mock, domain, config)
@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 = event.DOMAIN
assert hass.states.get(f"{platform}.test")
async def test_unload_entry(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
) -> None:
"""Test unloading the config entry."""
domain = event.DOMAIN
config = DEFAULT_CONFIG
await help_test_unload_config_entry_with_platform(
hass, mqtt_mock_entry, domain, config
)
@pytest.mark.parametrize(
("expected_friendly_name", "device_class"),
[("test", None), ("Doorbell", "doorbell"), ("Motion", "motion")],
)
async def test_entity_name(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
expected_friendly_name: str | None,
device_class: str | None,
) -> None:
"""Test the entity name setup."""
domain = event.DOMAIN
config = DEFAULT_CONFIG
await help_test_entity_name(
hass, mqtt_mock_entry, domain, config, expected_friendly_name, device_class
)