Add notify entity to LaMetric (#181971)

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Joost Lekkerkerker
2026-09-16 19:39:50 +02:00
committed by GitHub
co-authored by Claude
parent 5971baddcc
commit e952751b2d
6 changed files with 223 additions and 5 deletions
@@ -9,6 +9,7 @@ from homeassistant.const import Platform
DOMAIN: Final = "lametric"
PLATFORMS = [
Platform.BUTTON,
Platform.NOTIFY,
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
+4 -2
View File
@@ -34,12 +34,14 @@ def lametric_exception_handler[_LaMetricEntityT: LaMetricEntity, **_P](
self.coordinator.last_update_success = False
self.coordinator.async_update_listeners()
raise HomeAssistantError(
"Error communicating with the LaMetric device"
translation_domain=DOMAIN,
translation_key="communication_error",
) from error
except LaMetricError as error:
raise HomeAssistantError(
"Invalid response from the LaMetric device"
translation_domain=DOMAIN,
translation_key="invalid_response",
) from error
return handler
+43 -2
View File
@@ -15,15 +15,56 @@ from demetriek import (
Sound,
)
from homeassistant.components.notify import ATTR_DATA, BaseNotificationService
from homeassistant.components.notify import (
ATTR_DATA,
BaseNotificationService,
NotifyEntity,
)
from homeassistant.const import CONF_ICON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.enum import try_parse_enum
from .const import CONF_CYCLES, CONF_ICON_TYPE, CONF_PRIORITY, CONF_SOUND
from .coordinator import LaMetricConfigEntry
from .coordinator import LaMetricConfigEntry, LaMetricDataUpdateCoordinator
from .entity import LaMetricEntity
from .helpers import lametric_exception_handler
PARALLEL_UPDATES = 1
async def async_setup_entry(
hass: HomeAssistant,
entry: LaMetricConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up LaMetric notify entity based on a config entry."""
async_add_entities([LaMetricNotifyEntity(entry.runtime_data)])
class LaMetricNotifyEntity(LaMetricEntity, NotifyEntity):
"""Representation of a LaMetric notify entity."""
_attr_translation_key = "message"
def __init__(self, coordinator: LaMetricDataUpdateCoordinator) -> None:
"""Initialize the notify entity."""
super().__init__(coordinator=coordinator)
self._attr_unique_id = f"{coordinator.data.serial_number}-message"
@lametric_exception_handler
@override
async def async_send_message(self, message: str, title: str | None = None) -> None:
"""Send a message to the LaMetric device."""
await self.coordinator.lametric.notify(
notification=Notification(
icon_type=NotificationIconType.NONE,
priority=NotificationPriority.INFO,
model=Model(frames=[Simple(text=message)]),
)
)
async def async_get_service(
@@ -65,6 +65,11 @@
"name": "Dismiss current notification"
}
},
"notify": {
"message": {
"name": "Message"
}
},
"number": {
"brightness": {
"name": "Brightness"
@@ -93,6 +98,14 @@
}
}
},
"exceptions": {
"communication_error": {
"message": "Error communicating with the LaMetric device"
},
"invalid_response": {
"message": "Invalid response from the LaMetric device"
}
},
"selector": {
"icon_type": {
"options": {
@@ -0,0 +1,52 @@
# serializer version: 1
# name: test_all_entities[notify][notify.frenck_s_lametric_message-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'notify',
'entity_category': None,
'entity_id': 'notify.frenck_s_lametric_message',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Message',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Message',
'platform': 'lametric',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'message',
'unique_id': 'SA110405124500W00BS9-message',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[notify][notify.frenck_s_lametric_message-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: "Frenck's LaMetric Message",
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <NotifyEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'notify.frenck_s_lametric_message',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
+110 -1
View File
@@ -3,7 +3,9 @@
from unittest.mock import MagicMock
from demetriek import (
LaMetricConnectionError,
LaMetricError,
Model,
Notification,
NotificationIconType,
NotificationPriority,
@@ -12,18 +14,34 @@ from demetriek import (
Simple,
)
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.lametric.const import DOMAIN
from homeassistant.components.notify import (
ATTR_DATA,
ATTR_MESSAGE,
DOMAIN as NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
NOTIFY_SERVICE = "frenck_s_lametric"
ENTITY_ID = "notify.frenck_s_lametric_message"
pytestmark = pytest.mark.usefixtures("init_integration")
pytestmark = [
pytest.mark.parametrize("init_integration", [Platform.NOTIFY], indirect=True),
pytest.mark.usefixtures("init_integration"),
]
async def test_notification_defaults(
@@ -122,3 +140,94 @@ async def test_notification_error(
},
blocking=True,
)
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.freeze_time("2022-09-19 12:07:30")
async def test_send_message(
hass: HomeAssistant,
mock_lametric: MagicMock,
) -> None:
"""Test sending a message through the LaMetric notify entity."""
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
{
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_MESSAGE: "The way to get started is to quit talking and begin doing",
},
blocking=True,
)
mock_lametric.notify.assert_called_once_with(
notification=Notification(
icon_type=NotificationIconType.NONE,
priority=NotificationPriority.INFO,
model=Model(
frames=[
Simple(
text="The way to get started is to quit talking and begin doing"
)
]
),
)
)
state = hass.states.get(ENTITY_ID)
assert state
assert state.state == "2022-09-19T12:07:30+00:00"
@pytest.mark.parametrize(
("side_effect", "translation_key", "expected_state"),
[
pytest.param(
LaMetricError,
"invalid_response",
STATE_UNKNOWN,
id="error",
),
pytest.param(
LaMetricConnectionError,
"communication_error",
STATE_UNAVAILABLE,
id="connection_error",
),
],
)
async def test_send_message_error(
hass: HomeAssistant,
mock_lametric: MagicMock,
side_effect: type[LaMetricError],
translation_key: str,
expected_state: str,
) -> None:
"""Test error handling of the LaMetric notify entity."""
mock_lametric.notify.side_effect = side_effect
with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
{
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_MESSAGE: "It's failure that gives you the proper perspective",
},
blocking=True,
)
assert err.value.translation_domain == DOMAIN
assert err.value.translation_key == translation_key
state = hass.states.get(ENTITY_ID)
assert state
assert state.state == expected_state