1
0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00

Add support for tuya doorbell events (#156540)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
epenet
2025-11-22 15:17:37 +01:00
committed by GitHub
parent c0772f3957
commit ca2e8bfb56
6 changed files with 675 additions and 24 deletions

View File

@@ -2,6 +2,10 @@
from __future__ import annotations
from base64 import b64decode
from dataclasses import dataclass
from typing import Any
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.event import (
@@ -16,68 +20,162 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import TuyaConfigEntry
from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
from .entity import TuyaEntity
from .models import DPCodeEnumWrapper
from .models import (
DPCodeBase64Wrapper,
DPCodeEnumWrapper,
DPCodeStringWrapper,
DPCodeTypeInformationWrapper,
)
class _DPCodeEventWrapper(DPCodeTypeInformationWrapper):
"""Base class for Tuya event wrappers."""
@property
def event_types(self) -> list[str]:
"""Return the event types for the DP code."""
return ["triggered"]
def get_event_type(
self, device: CustomerDevice, updated_status_properties: list[str] | None
) -> str | None:
"""Return the event type."""
if (
updated_status_properties is None
or self.dpcode not in updated_status_properties
):
return None
return "triggered"
def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
"""Return the event attributes."""
return None
class _EventEnumWrapper(DPCodeEnumWrapper, _DPCodeEventWrapper):
"""Wrapper for event enum DP codes."""
@property
def event_types(self) -> list[str]:
"""Return the event types for the enum."""
return self.type_information.range
def get_event_type(
self, device: CustomerDevice, updated_status_properties: list[str] | None
) -> str | None:
"""Return the triggered event type."""
if (
updated_status_properties is None
or self.dpcode not in updated_status_properties
):
return None
return self.read_device_status(device)
class _AlarmMessageWrapper(DPCodeStringWrapper, _DPCodeEventWrapper):
"""Wrapper for a STRING message on DPCode.ALARM_MESSAGE."""
def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
"""Return the event attributes for the alarm message."""
if (raw_value := self._read_device_status_raw(device)) is None:
return None
return {"message": b64decode(raw_value).decode("utf-8")}
class _DoorbellPicWrapper(DPCodeBase64Wrapper, _DPCodeEventWrapper):
"""Wrapper for a RAW message on DPCode.DOORBELL_PIC.
It is expected that the RAW data is base64/utf8 encoded URL of the picture.
"""
def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
"""Return the event attributes for the doorbell picture."""
if (raw_value := self._read_device_status_raw(device)) is None:
return None
return {"message": b64decode(raw_value).decode("utf-8")}
@dataclass(frozen=True)
class TuyaEventEntityDescription(EventEntityDescription):
"""Describe a Tuya Event entity."""
wrapper_class: type[_DPCodeEventWrapper] = _EventEnumWrapper
# All descriptions can be found here. Mostly the Enum data types in the
# default status set of each category (that don't have a set instruction)
# end up being events.
EVENTS: dict[DeviceCategory, tuple[EventEntityDescription, ...]] = {
EVENTS: dict[DeviceCategory, tuple[TuyaEventEntityDescription, ...]] = {
DeviceCategory.SP: (
TuyaEventEntityDescription(
key=DPCode.ALARM_MESSAGE,
device_class=EventDeviceClass.DOORBELL,
translation_key="doorbell_message",
wrapper_class=_AlarmMessageWrapper,
),
TuyaEventEntityDescription(
key=DPCode.DOORBELL_PIC,
device_class=EventDeviceClass.DOORBELL,
translation_key="doorbell_picture",
wrapper_class=_DoorbellPicWrapper,
),
),
DeviceCategory.WXKG: (
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE1,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "1"},
),
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE2,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "2"},
),
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE3,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "3"},
),
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE4,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "4"},
),
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE5,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "5"},
),
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE6,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "6"},
),
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE7,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "7"},
),
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE8,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "8"},
),
EventEntityDescription(
TuyaEventEntityDescription(
key=DPCode.SWITCH_MODE9,
device_class=EventDeviceClass.BUTTON,
translation_key="numbered_button",
translation_placeholders={"button_number": "9"},
),
)
),
}
@@ -102,8 +200,8 @@ async def async_setup_entry(
)
for description in descriptions
if (
dpcode_wrapper := DPCodeEnumWrapper.find_dpcode(
device, description.key, prefer_function=True
dpcode_wrapper := description.wrapper_class.find_dpcode(
device, description.key
)
)
)
@@ -127,14 +225,14 @@ class TuyaEventEntity(TuyaEntity, EventEntity):
device: CustomerDevice,
device_manager: Manager,
description: EventEntityDescription,
dpcode_wrapper: DPCodeEnumWrapper,
dpcode_wrapper: _DPCodeEventWrapper,
) -> None:
"""Init Tuya event entity."""
super().__init__(device, device_manager)
self.entity_description = description
self._attr_unique_id = f"{super().unique_id}{description.key}"
self._dpcode_wrapper = dpcode_wrapper
self._attr_event_types = dpcode_wrapper.type_information.range
self._attr_event_types = dpcode_wrapper.event_types
async def _handle_state_update(
self,
@@ -142,11 +240,14 @@ class TuyaEventEntity(TuyaEntity, EventEntity):
dp_timestamps: dict | None = None,
) -> None:
if (
updated_status_properties is None
or self._dpcode_wrapper.dpcode not in updated_status_properties
or (value := self._dpcode_wrapper.read_device_status(self.device)) is None
):
event_type := self._dpcode_wrapper.get_event_type(
self.device, updated_status_properties
)
) is None:
return
self._trigger_event(value)
self._trigger_event(
event_type,
self._dpcode_wrapper.get_event_attributes(self.device),
)
self.async_write_ha_state()

View File

@@ -2,7 +2,6 @@
from __future__ import annotations
from abc import ABC, abstractmethod
import base64
from dataclasses import dataclass
from typing import Any, Literal, Self, cast, overload
@@ -162,10 +161,11 @@ _TYPE_INFORMATION_MAPPINGS: dict[DPType, type[TypeInformation]] = {
DPType.INTEGER: IntegerTypeData,
DPType.JSON: TypeInformation,
DPType.RAW: TypeInformation,
DPType.STRING: TypeInformation,
}
class DPCodeWrapper(ABC):
class DPCodeWrapper:
"""Base DPCode wrapper.
Used as a common interface for referring to a DPCode, and
@@ -186,12 +186,12 @@ class DPCodeWrapper(ABC):
"""
return device.status.get(self.dpcode)
@abstractmethod
def read_device_status(self, device: CustomerDevice) -> Any | None:
"""Read the device value for the dpcode.
The raw device status is converted to a Home Assistant value.
"""
raise NotImplementedError
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value.
@@ -363,6 +363,16 @@ class DPCodeIntegerWrapper(DPCodeTypeInformationWrapper[IntegerTypeData]):
)
class DPCodeStringWrapper(DPCodeTypeInformationWrapper[TypeInformation]):
"""Wrapper to extract information from a STRING value."""
DPTYPE = DPType.STRING
def read_device_status(self, device: CustomerDevice) -> str | None:
"""Read the device value for the dpcode."""
return self._read_device_status_raw(device)
class DPCodeBitmapBitWrapper(DPCodeWrapper):
"""Simple wrapper for a specific bit in bitmap values."""

View File

@@ -117,6 +117,12 @@
}
},
"event": {
"doorbell_message": {
"name": "Doorbell message"
},
"doorbell_picture": {
"name": "Doorbell picture"
},
"numbered_button": {
"name": "Button {button_number}",
"state_attributes": {

View File

@@ -214,6 +214,9 @@ async def _create_device(hass: HomeAssistant, mock_device_code: str) -> Customer
(dp_type := device.function.get(key)) and dp_type.type == "Json"
):
device.status[key] = json_dumps(value)
if value == "**REDACTED**":
# It was redacted, which may cause issue with b64decode
device.status[key] = ""
return device

View File

@@ -1,4 +1,26 @@
# serializer version: 1
# name: test_alarm_message_event[event.intercom_doorbell_message-alarm_message-eyJzb21lIjogImpzb24iLCAicmFuZG9tIjogImRhdGEifQ==-sp_csr2fqitalj5o0tq]
ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'Intercom Doorbell message',
'message': '{"some": "json", "random": "data"}',
})
# ---
# name: test_alarm_message_event[event.intercom_doorbell_picture-doorbell_pic-aHR0cHM6Ly9zb21lLXBpY3R1cmUtdXJsLmNvbS9pbWFnZS5qcGc=-sp_csr2fqitalj5o0tq]
ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'Intercom Doorbell picture',
'message': 'https://some-picture-url.com/image.jpg',
})
# ---
# name: test_platform_setup_and_discovery[event.bathroom_smart_switch_button_1-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@@ -178,3 +200,467 @@
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---
# name: test_platform_setup_and_discovery[event.burocam_doorbell_message-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'triggered',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.burocam_doorbell_message',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell message',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell_message',
'unique_id': 'tuya.svjjuwykgijjedurpsalarm_message',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[event.burocam_doorbell_message-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'Bürocam Doorbell message',
'message': '',
}),
'context': <ANY>,
'entity_id': 'event.burocam_doorbell_message',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---
# name: test_platform_setup_and_discovery[event.c9_doorbell_message-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'triggered',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.c9_doorbell_message',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell message',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell_message',
'unique_id': 'tuya.fjdyw5ld2f5f5ddspsalarm_message',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[event.c9_doorbell_message-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'C9 Doorbell message',
'message': '',
}),
'context': <ANY>,
'entity_id': 'event.c9_doorbell_message',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---
# name: test_platform_setup_and_discovery[event.c9_doorbell_picture-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'triggered',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.c9_doorbell_picture',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell picture',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell_picture',
'unique_id': 'tuya.fjdyw5ld2f5f5ddspsdoorbell_pic',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[event.c9_doorbell_picture-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'C9 Doorbell picture',
'message': '',
}),
'context': <ANY>,
'entity_id': 'event.c9_doorbell_picture',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---
# name: test_platform_setup_and_discovery[event.garage_camera_doorbell_message-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'triggered',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.garage_camera_doorbell_message',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell message',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell_message',
'unique_id': 'tuya.53fnjncm3jywuaznpsalarm_message',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[event.garage_camera_doorbell_message-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'Garage Camera Doorbell message',
'message': '',
}),
'context': <ANY>,
'entity_id': 'event.garage_camera_doorbell_message',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---
# name: test_platform_setup_and_discovery[event.intercom_doorbell_message-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'triggered',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.intercom_doorbell_message',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell message',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell_message',
'unique_id': 'tuya.qt0o5jlatiqf2rscpsalarm_message',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[event.intercom_doorbell_message-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'Intercom Doorbell message',
'message': '',
}),
'context': <ANY>,
'entity_id': 'event.intercom_doorbell_message',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---
# name: test_platform_setup_and_discovery[event.intercom_doorbell_picture-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'triggered',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.intercom_doorbell_picture',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell picture',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell_picture',
'unique_id': 'tuya.qt0o5jlatiqf2rscpsdoorbell_pic',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[event.intercom_doorbell_picture-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'Intercom Doorbell picture',
'message': '',
}),
'context': <ANY>,
'entity_id': 'event.intercom_doorbell_picture',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---
# name: test_platform_setup_and_discovery[event.mirilla_puerta_doorbell_message-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'triggered',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.mirilla_puerta_doorbell_message',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell message',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell_message',
'unique_id': 'tuya.i6xywcsymer1kmb6psalarm_message',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[event.mirilla_puerta_doorbell_message-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'Mirilla puerta Doorbell message',
'message': '',
}),
'context': <ANY>,
'entity_id': 'event.mirilla_puerta_doorbell_message',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---
# name: test_platform_setup_and_discovery[event.mirilla_puerta_doorbell_picture-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'triggered',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.mirilla_puerta_doorbell_picture',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.DOORBELL: 'doorbell'>,
'original_icon': None,
'original_name': 'Doorbell picture',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'doorbell_picture',
'unique_id': 'tuya.i6xywcsymer1kmb6psdoorbell_pic',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[event.mirilla_puerta_doorbell_picture-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'doorbell',
'event_type': 'triggered',
'event_types': list([
'triggered',
]),
'friendly_name': 'Mirilla puerta Doorbell picture',
'message': '',
}),
'context': <ANY>,
'entity_id': 'event.mirilla_puerta_doorbell_picture',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-11-01T12:14:15.000+00:00',
})
# ---

View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import base64
from unittest.mock import patch
import pytest
@@ -38,3 +39,47 @@ async def test_platform_setup_and_discovery(
)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
"mock_device_code",
["sp_csr2fqitalj5o0tq"],
)
@pytest.mark.parametrize(
("entity_id", "dpcode", "value"),
[
(
"event.intercom_doorbell_picture",
"doorbell_pic",
base64.b64encode(b"https://some-picture-url.com/image.jpg"),
),
(
"event.intercom_doorbell_message",
"alarm_message",
base64.b64encode(b'{"some": "json", "random": "data"}'),
),
],
)
@patch("homeassistant.components.tuya.PLATFORMS", [Platform.EVENT])
async def test_alarm_message_event(
hass: HomeAssistant,
mock_manager: Manager,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
mock_listener: MockDeviceListener,
snapshot: SnapshotAssertion,
entity_id: str,
dpcode: str,
value: str,
) -> None:
"""Test alarm message event."""
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
mock_device.status[dpcode] = value
await mock_listener.async_send_device_update(hass, mock_device, mock_device.status)
# Verify event was triggered with correct type and decoded URL
state = hass.states.get(entity_id)
assert state.attributes == snapshot
assert state.attributes["message"]