Add notify platform to Amazon Devices (#145589)

* Add notify platform to Amazon Devices

* apply review comment

* leftover

* tests leftovers

* remove sound notification

* missing await
This commit is contained in:
Simone Chemelli 2025-05-26 18:17:32 +03:00 committed by GitHub
parent c2a5e1aaf9
commit b7ce0f63a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 250 additions and 0 deletions

View File

@ -7,6 +7,7 @@ from .coordinator import AmazonConfigEntry, AmazonDevicesCoordinator
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.NOTIFY,
Platform.SWITCH,
]

View File

@ -0,0 +1,74 @@
"""Support for notification entity."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any, Final
from aioamazondevices.api import AmazonDevice, AmazonEchoApi
from homeassistant.components.notify import NotifyEntity, NotifyEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import AmazonConfigEntry
from .entity import AmazonEntity
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
class AmazonNotifyEntityDescription(NotifyEntityDescription):
"""Amazon Devices notify entity description."""
method: Callable[[AmazonEchoApi, AmazonDevice, str], Awaitable[None]]
subkey: str
NOTIFY: Final = (
AmazonNotifyEntityDescription(
key="speak",
translation_key="speak",
subkey="AUDIO_PLAYER",
method=lambda api, device, message: api.call_alexa_speak(device, message),
),
AmazonNotifyEntityDescription(
key="announce",
translation_key="announce",
subkey="AUDIO_PLAYER",
method=lambda api, device, message: api.call_alexa_announcement(
device, message
),
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: AmazonConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Amazon Devices notification entity based on a config entry."""
coordinator = entry.runtime_data
async_add_entities(
AmazonNotifyEntity(coordinator, serial_num, sensor_desc)
for sensor_desc in NOTIFY
for serial_num in coordinator.data
if sensor_desc.subkey in coordinator.data[serial_num].capabilities
)
class AmazonNotifyEntity(AmazonEntity, NotifyEntity):
"""Binary sensor notify platform."""
entity_description: AmazonNotifyEntityDescription
async def async_send_message(
self, message: str, title: str | None = None, **kwargs: Any
) -> None:
"""Send a message."""
await self.entity_description.method(self.coordinator.api, self.device, message)

View File

@ -43,6 +43,14 @@
"name": "Bluetooth"
}
},
"notify": {
"speak": {
"name": "Speak"
},
"announce": {
"name": "Announce"
}
},
"switch": {
"do_not_disturb": {
"name": "Do not disturb"

View File

@ -0,0 +1,97 @@
# serializer version: 1
# name: test_all_entities[notify.echo_test_announce-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'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.echo_test_announce',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Announce',
'platform': 'amazon_devices',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'announce',
'unique_id': 'echo_test_serial_number-announce',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[notify.echo_test_announce-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Echo Test Announce',
'supported_features': <NotifyEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'notify.echo_test_announce',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_all_entities[notify.echo_test_speak-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'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.echo_test_speak',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Speak',
'platform': 'amazon_devices',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'speak',
'unique_id': 'echo_test_serial_number-speak',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[notify.echo_test_speak-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Echo Test Speak',
'supported_features': <NotifyEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'notify.echo_test_speak',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@ -0,0 +1,70 @@
"""Tests for the Amazon Devices notify platform."""
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.notify import (
ATTR_MESSAGE,
DOMAIN as NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
)
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_amazon_devices_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all entities."""
with patch("homeassistant.components.amazon_devices.PLATFORMS", [Platform.NOTIFY]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
"mode",
["speak", "announce"],
)
async def test_notify_send_message(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_amazon_devices_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mode: str,
) -> None:
"""Test notify send message."""
await setup_integration(hass, mock_config_entry)
entity_id = f"notify.echo_test_{mode}"
now = dt_util.parse_datetime("2021-01-09 12:00:00+00:00")
assert now
freezer.move_to(now)
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
{
ATTR_ENTITY_ID: entity_id,
ATTR_MESSAGE: "Test Message",
},
blocking=True,
)
assert (state := hass.states.get(entity_id))
assert state.state == now.isoformat()