starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-05 10:05:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add notify platform to Habitica (#150553)

Browse Source
This commit is contained in:
Manu
2025-09-30 22:35:55 +02:00
committed by GitHub
parent db3b070ed0
commit 4ac89f6849
13 changed files with 809 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/habitica/__init__.py
+17 -1
View File
@@ -4,9 +4,14 @@ from uuid import UUID
from habiticalib import Habitica
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.const import CONF_API_KEY, CONF_URL, CONF_VERIFY_SSL, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_registry as er,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
@@ -27,6 +32,7 @@ PLATFORMS = [
Platform.BUTTON,
Platform.CALENDAR,
Platform.IMAGE,
Platform.NOTIFY,
Platform.SENSOR,
Platform.SWITCH,
Platform.TODO,
@@ -46,6 +52,7 @@ async def async_setup_entry(
"""Set up habitica from a config entry."""
party_added_by_this_entry: UUID | None = None
device_reg = dr.async_get(hass)
entity_registry = er.async_get(hass)
session = async_get_clientsession(
hass, verify_ssl=config_entry.data.get(CONF_VERIFY_SSL, True)
@@ -96,6 +103,15 @@ async def async_setup_entry(
device.id, remove_config_entry_id=config_entry.entry_id
)
notify_entities = [
entry.entity_id
for entry in entity_registry.entities.values()
if entry.domain == NOTIFY_DOMAIN
and entry.config_entry_id == config_entry.entry_id
]
for entity_id in notify_entities:
entity_registry.async_remove(entity_id)
hass.config_entries.async_schedule_reload(config_entry.entry_id)
coordinator.async_add_listener(_party_update_listener)
homeassistant/components/habitica/binary_sensor.py
+1 -1
View File
@@ -121,4 +121,4 @@ class HabiticaPartyBinarySensorEntity(HabiticaPartyBase, BinarySensorEntity):
@property
def is_on(self) -> bool | None:
"""If the binary sensor is on."""
return self.coordinator.data.quest.active
return self.coordinator.data.party.quest.active
homeassistant/components/habitica/coordinator.py
+20 -3
View File
@@ -9,6 +9,7 @@ from datetime import timedelta
from io import BytesIO
import logging
from typing import Any
from uuid import UUID
from aiohttp import ClientError
from habiticalib import (
@@ -48,6 +49,14 @@ class HabiticaData:
tasks: list[TaskData]
@dataclass
class HabiticaPartyData:
"""Habitica party data."""
party: GroupData
members: dict[UUID, UserData]
type HabiticaConfigEntry = ConfigEntry[HabiticaDataUpdateCoordinator]
@@ -192,11 +201,19 @@ class HabiticaDataUpdateCoordinator(HabiticaBaseCoordinator[HabiticaData]):
return png.getvalue()
class HabiticaPartyCoordinator(HabiticaBaseCoordinator[GroupData]):
class HabiticaPartyCoordinator(HabiticaBaseCoordinator[HabiticaPartyData]):
"""Habitica Party Coordinator."""
_update_interval = timedelta(minutes=15)
async def _update_data(self) -> GroupData:
async def _update_data(self) -> HabiticaPartyData:
"""Fetch the latest party data."""
return (await self.habitica.get_group()).data
return HabiticaPartyData(
party=(await self.habitica.get_group()).data,
members={
member.id: member
for member in (await self.habitica.get_group_members()).data
if member.id
},
)
homeassistant/components/habitica/entity.py
+2 -2
View File
@@ -68,14 +68,14 @@ class HabiticaPartyBase(CoordinatorEntity[HabiticaPartyCoordinator]):
super().__init__(coordinator)
if TYPE_CHECKING:
assert config_entry.unique_id
unique_id = f"{config_entry.unique_id}_{coordinator.data.id!s}"
unique_id = f"{config_entry.unique_id}_{coordinator.data.party.id!s}"
self.entity_description = entity_description
self._attr_unique_id = f"{unique_id}_{entity_description.key}"
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
manufacturer=MANUFACTURER,
model=NAME,
name=coordinator.data.summary,
name=coordinator.data.party.summary,
identifiers={(DOMAIN, unique_id)},
via_device=(DOMAIN, config_entry.unique_id),
)
homeassistant/components/habitica/icons.json
+5
View File
@@ -194,6 +194,11 @@
"quest_running": {
"default": "mdi:script-text-play"
}
},
"notify": {
"party_chat": {
"default": "mdi:forum"
}
}
},
"services": {
homeassistant/components/habitica/image.py
+1 -1
View File
@@ -128,7 +128,7 @@ class HabiticaPartyImage(HabiticaPartyBase, ImageEntity):
"""Return URL of image."""
return (
f"{ASSETS_URL}quest_{key}.png"
if (key := self.coordinator.data.quest.key)
if (key := self.coordinator.data.party.quest.key)
else None
)
homeassistant/components/habitica/notify.py
+202
View File
@@ -0,0 +1,202 @@
"""Notify platform for the Habitica integration."""
from __future__ import annotations
from abc import abstractmethod
from enum import StrEnum
from typing import TYPE_CHECKING
from uuid import UUID
from aiohttp import ClientError
from habiticalib import (
GroupData,
HabiticaException,
NotAuthorizedError,
NotFoundError,
TooManyRequestsError,
UserData,
)
from homeassistant.components.notify import (
DOMAIN as NOTIFY_DOMAIN,
NotifyEntity,
NotifyEntityDescription,
)
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HABITICA_KEY
from .const import DOMAIN
from .coordinator import HabiticaConfigEntry, HabiticaDataUpdateCoordinator
from .entity import HabiticaBase
PARALLEL_UPDATES = 10
class HabiticaNotify(StrEnum):
"""Habitica Notifier."""
PARTY_CHAT = "party_chat"
PRIVATE_MESSAGE = "private_message"
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HabiticaConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the notify entity platform."""
members_added: set[UUID] = set()
entity_registry = er.async_get(hass)
coordinator = config_entry.runtime_data
if party := coordinator.data.user.party.id:
party_coordinator = hass.data[HABITICA_KEY][party]
async_add_entities(
[HabiticaPartyChatNotifyEntity(coordinator, party_coordinator.data.party)]
)
@callback
def add_entities() -> None:
nonlocal members_added
new_members = set(party_coordinator.data.members.keys()) - members_added
if TYPE_CHECKING:
assert coordinator.data.user.id
new_members.discard(coordinator.data.user.id)
if new_members:
async_add_entities(
HabiticaPrivateMessageNotifyEntity(
coordinator, party_coordinator.data.members[member]
)
for member in new_members
)
members_added |= new_members
delete_members = members_added - set(party_coordinator.data.members.keys())
for member in delete_members:
if entity_id := entity_registry.async_get_entity_id(
NOTIFY_DOMAIN,
DOMAIN,
f"{coordinator.config_entry.unique_id}_{member!s}_{HabiticaNotify.PRIVATE_MESSAGE}",
):
entity_registry.async_remove(entity_id)
members_added.discard(member)
party_coordinator.async_add_listener(add_entities)
add_entities()
class HabiticaBaseNotifyEntity(HabiticaBase, NotifyEntity):
"""Habitica base notify entity."""
def __init__(
self,
coordinator: HabiticaDataUpdateCoordinator,
) -> None:
"""Initialize a Habitica entity."""
super().__init__(coordinator, self.entity_description)
@abstractmethod
async def _send_message(self, message: str) -> None:
"""Send a Habitica message."""
async def async_send_message(self, message: str, title: str | None = None) -> None:
"""Send a message."""
try:
await self._send_message(message)
except NotAuthorizedError as e:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="send_message_forbidden",
translation_placeholders={
**self.translation_placeholders,
"reason": e.error.message,
},
) from e
except NotFoundError as e:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="send_message_not_found",
translation_placeholders={
**self.translation_placeholders,
"reason": e.error.message,
},
) from e
except TooManyRequestsError as e:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="setup_rate_limit_exception",
translation_placeholders={"retry_after": str(e.retry_after)},
) from e
except HabiticaException as e:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="service_call_exception",
translation_placeholders={"reason": e.error.message},
) from e
except ClientError as e:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="service_call_exception",
translation_placeholders={"reason": str(e)},
) from e
class HabiticaPartyChatNotifyEntity(HabiticaBaseNotifyEntity):
"""Representation of a Habitica party chat notify entity."""
def __init__(
self,
coordinator: HabiticaDataUpdateCoordinator,
party: GroupData,
) -> None:
"""Initialize a Habitica entity."""
self._attr_translation_placeholders = {CONF_NAME: party.name}
self.entity_description = NotifyEntityDescription(
key=HabiticaNotify.PARTY_CHAT,
translation_key=HabiticaNotify.PARTY_CHAT,
)
self.party = party
super().__init__(coordinator)
async def _send_message(self, message: str) -> None:
"""Send a Habitica party chat message."""
await self.coordinator.habitica.send_group_message(
message=message,
group_id=self.party.id,
)
class HabiticaPrivateMessageNotifyEntity(HabiticaBaseNotifyEntity):
"""Representation of a Habitica private message notify entity."""
def __init__(
self,
coordinator: HabiticaDataUpdateCoordinator,
member: UserData,
) -> None:
"""Initialize a Habitica entity."""
self._attr_translation_placeholders = {CONF_NAME: member.profile.name or ""}
self.entity_description = NotifyEntityDescription(
key=f"{member.id!s}_{HabiticaNotify.PRIVATE_MESSAGE}",
translation_key=HabiticaNotify.PRIVATE_MESSAGE,
)
self.member = member
super().__init__(coordinator)
async def _send_message(self, message: str) -> None:
"""Send a Habitica private message."""
if TYPE_CHECKING:
assert self.member.id
await self.coordinator.habitica.send_private_message(
message=message,
to_user_id=self.member.id,
)
homeassistant/components/habitica/sensor.py
+7 -3
View File
@@ -445,7 +445,9 @@ class HabiticaPartySensor(HabiticaPartyBase, SensorEntity):
def native_value(self) -> StateType:
"""Return the state of the device."""
return self.entity_description.value_fn(self.coordinator.data, self.content)
return self.entity_description.value_fn(
self.coordinator.data.party, self.content
)
@property
def entity_picture(self) -> str | None:
@@ -453,7 +455,9 @@ class HabiticaPartySensor(HabiticaPartyBase, SensorEntity):
pic = self.entity_description.entity_picture
entity_picture = (
pic if isinstance(pic, str) or pic is None else pic(self.coordinator.data)
pic
if isinstance(pic, str) or pic is None
else pic(self.coordinator.data.party)
)
return (
@@ -468,5 +472,5 @@ class HabiticaPartySensor(HabiticaPartyBase, SensorEntity):
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return entity specific state attributes."""
if func := self.entity_description.attributes_fn:
return func(self.coordinator.data, self.content)
return func(self.coordinator.data.party, self.content)
return None
homeassistant/components/habitica/strings.json
+14
View File
@@ -264,6 +264,14 @@
"name": "[%key:component::habitica::common::quest_name%]"
}
},
"notify": {
"party_chat": {
"name": "Party chat"
},
"private_message": {
"name": "Private message: {name}"
}
},
"sensor": {
"display_name": {
"name": "Display name",
@@ -572,6 +580,12 @@
},
"frequency_not_monthly": {
"message": "Unable to update task, monthly repeat settings apply only to monthly recurring dailies."
},
"send_message_forbidden": {
"message": "You are not allowed to send messages to {name}. ({reason})"
},
"send_message_not_found": {
"message": "Unable to send message, {name} not found. ({reason})"
}
},
"issues": {
tests/components/habitica/fixtures/party_members_2.json
+238
View File
@@ -0,0 +1,238 @@
{
"success": true,
"data": [
{
"_id": "a380546a-94be-4b8e-8a0b-23e0d5c03303",
"auth": {
"local": {
"username": "test-username"
},
"timestamps": {
"created": "2024-10-19T18:43:39.782Z",
"loggedin": "2024-10-31T16:13:35.048Z",
"updated": "2024-10-31T16:15:56.552Z"
}
},
"achievements": {
"ultimateGearSets": {
"healer": false,
"wizard": false,
"rogue": false,
"warrior": false
},
"streak": 0,
"challenges": [],
"perfect": 1,
"quests": {},
"purchasedEquipment": true,
"completedTask": true,
"partyUp": true
},
"backer": {},
"contributor": {},
"flags": {
"verifiedUsername": true,
"classSelected": true
},
"items": {
"gear": {
"owned": {
"headAccessory_special_blackHeadband": true,
"headAccessory_special_blueHeadband": true,
"headAccessory_special_greenHeadband": true,
"headAccessory_special_pinkHeadband": true,
"headAccessory_special_redHeadband": true,
"headAccessory_special_whiteHeadband": true,
"headAccessory_special_yellowHeadband": true,
"eyewear_special_blackTopFrame": true,
"eyewear_special_blueTopFrame": true,
"eyewear_special_greenTopFrame": true,
"eyewear_special_pinkTopFrame": true,
"eyewear_special_redTopFrame": true,
"eyewear_special_whiteTopFrame": true,
"eyewear_special_yellowTopFrame": true,
"eyewear_special_blackHalfMoon": true,
"eyewear_special_blueHalfMoon": true,
"eyewear_special_greenHalfMoon": true,
"eyewear_special_pinkHalfMoon": true,
"eyewear_special_redHalfMoon": true,
"eyewear_special_whiteHalfMoon": true,
"eyewear_special_yellowHalfMoon": true,
"armor_special_bardRobes": true,
"weapon_special_fall2024Warrior": true,
"shield_special_fall2024Warrior": true,
"head_special_fall2024Warrior": true,
"armor_special_fall2024Warrior": true,
"back_mystery_201402": true,
"body_mystery_202003": true,
"head_special_bardHat": true,
"weapon_wizard_0": true
},
"equipped": {
"weapon": "weapon_special_fall2024Warrior",
"armor": "armor_special_fall2024Warrior",
"head": "head_special_fall2024Warrior",
"shield": "shield_special_fall2024Warrior",
"back": "back_mystery_201402",
"headAccessory": "headAccessory_special_pinkHeadband",
"eyewear": "eyewear_special_pinkHalfMoon",
"body": "body_mystery_202003"
},
"costume": {
"armor": "armor_base_0",
"head": "head_base_0",
"shield": "shield_base_0"
}
},
"special": {
"snowball": 99,
"spookySparkles": 99,
"shinySeed": 99,
"seafoam": 99,
"valentine": 0,
"valentineReceived": [],
"nye": 0,
"nyeReceived": [],
"greeting": 0,
"greetingReceived": [],
"thankyou": 0,
"thankyouReceived": [],
"birthday": 0,
"birthdayReceived": [],
"congrats": 0,
"congratsReceived": [],
"getwell": 0,
"getwellReceived": [],
"goodluck": 0,
"goodluckReceived": []
},
"pets": {
"Rat-Shade": 1,
"Gryphatrice-Jubilant": 1
},
"currentPet": "Gryphatrice-Jubilant",
"eggs": {
"Cactus": 1,
"Fox": 2,
"Wolf": 1
},
"hatchingPotions": {
"CottonCandyBlue": 1,
"RoyalPurple": 1
},
"food": {
"Meat": 2,
"Chocolate": 1,
"CottonCandyPink": 1,
"Candy_Zombie": 1
},
"mounts": {
"Velociraptor-Base": true,
"Gryphon-Gryphatrice": true
},
"currentMount": "Gryphon-Gryphatrice",
"quests": {
"dustbunnies": 1,
"vice1": 1,
"atom1": 1,
"moonstone1": 1,
"goldenknight1": 1,
"basilist": 1
},
"lastDrop": {
"date": "2024-10-31T16:13:34.952Z",
"count": 0
}
},
"party": {
"quest": {
"progress": {
"up": 0,
"down": 0,
"collectedItems": 0,
"collect": {}
},
"RSVPNeeded": false,
"key": "dustbunnies"
},
"order": "level",
"orderAscending": "ascending",
"_id": "94cd398c-2240-4320-956e-6d345cf2c0de"
},
"preferences": {
"size": "slim",
"hair": {
"color": "red",
"base": 3,
"bangs": 1,
"beard": 0,
"mustache": 0,
"flower": 1
},
"skin": "915533",
"shirt": "blue",
"chair": "handleless_pink",
"costume": false,
"sleep": false,
"disableClasses": false,
"tasks": {
"groupByChallenge": false,
"confirmScoreNotes": false,
"mirrorGroupTasks": [],
"activeFilter": {
"habit": "all",
"daily": "all",
"todo": "remaining",
"reward": "all"
}
},
"background": "violet"
},
"profile": {
"name": "test-user"
},
"stats": {
"hp": 50,
"mp": 150.8,
"exp": 127,
"gp": 19.08650199252128,
"lvl": 99,
"class": "wizard",
"points": 0,
"str": 0,
"con": 0,
"int": 0,
"per": 0,
"buffs": {
"str": 50,
"int": 50,
"per": 50,
"con": 50,
"stealth": 0,
"streaks": false,
"seafoam": false,
"shinySeed": false,
"snowball": false,
"spookySparkles": false
},
"training": {
"int": 0,
"per": 0,
"str": 0,
"con": 0
},
"toNextLevel": 3580,
"maxHealth": 50,
"maxMP": 228
},
"inbox": {
"optOut": false
},
"loginIncentives": 6,
"id": "a380546a-94be-4b8e-8a0b-23e0d5c03303"
}
],
"notifications": [],
"userV": 96,
"appVersion": "5.29.0"
}
tests/components/habitica/snapshots/test_notify.ambr
+99
View File
@@ -0,0 +1,99 @@
# serializer version: 1
# name: test_notify_platform[notify.test_user_party_chat-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.test_user_party_chat',
'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': 'Party chat',
'platform': 'habitica',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': <HabiticaNotify.PARTY_CHAT: 'party_chat'>,
'unique_id': 'a380546a-94be-4b8e-8a0b-23e0d5c03303_party_chat',
'unit_of_measurement': None,
})
# ---
# name: test_notify_platform[notify.test_user_party_chat-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'test-user Party chat',
'supported_features': <NotifyEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'notify.test_user_party_chat',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_notify_platform[notify.test_user_private_message_test_partymember_displayname-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.test_user_private_message_test_partymember_displayname',
'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': 'Private message: test-partymember-displayname',
'platform': 'habitica',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': <HabiticaNotify.PRIVATE_MESSAGE: 'private_message'>,
'unique_id': 'a380546a-94be-4b8e-8a0b-23e0d5c03303_ffce870c-3ff3-4fa4-bad1-87612e52b8e7_private_message',
'unit_of_measurement': None,
})
# ---
# name: test_notify_platform[notify.test_user_private_message_test_partymember_displayname-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'test-user Private message: test-partymember-displayname',
'supported_features': <NotifyEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'notify.test_user_private_message_test_partymember_displayname',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
tests/components/habitica/test_init.py
+12 -1
View File
@@ -139,7 +139,7 @@ async def test_remove_party_and_reload(
freezer: FrozenDateTimeFactory,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test we leave the party and device is removed."""
"""Test we leave the party and device/notifiers are removed."""
group_id = "1e87097c-4c03-4f8c-a475-67cc7da7f409"
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
@@ -154,6 +154,11 @@ async def test_remove_party_and_reload(
is not None
)
assert hass.states.get("notify.test_user_party_chat")
assert hass.states.get(
"notify.test_user_private_message_test_partymember_displayname"
)
habitica.get_user.return_value = HabiticaUserResponse.from_json(
await async_load_fixture(hass, "user_no_party.json", DOMAIN)
)
@@ -168,3 +173,9 @@ async def test_remove_party_and_reload(
)
is None
)
assert hass.states.get("notify.test_user_party_chat") is None
assert (
hass.states.get("notify.test_user_private_message_test_partymember_displayname")
is None
)
tests/components/habitica/test_notify.py
+191
View File
@@ -0,0 +1,191 @@
"""Tests for the Habitica notify platform."""
from collections.abc import AsyncGenerator
from datetime import timedelta
from typing import Any
from unittest.mock import AsyncMock, patch
from uuid import UUID
from aiohttp import ClientError
from freezegun.api import FrozenDateTimeFactory, freeze_time
from habiticalib import HabiticaGroupMembersResponse
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.habitica.const import DOMAIN
from homeassistant.components.notify import (
ATTR_MESSAGE,
DOMAIN as NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from .conftest import (
ERROR_BAD_REQUEST,
ERROR_NOT_AUTHORIZED,
ERROR_NOT_FOUND,
ERROR_TOO_MANY_REQUESTS,
)
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
async_load_fixture,
snapshot_platform,
)
@pytest.fixture(autouse=True)
async def notify_only() -> AsyncGenerator[None]:
"""Enable only the notify platform."""
with patch(
"homeassistant.components.habitica.PLATFORMS",
[Platform.NOTIFY],
):
yield
@pytest.mark.usefixtures("habitica")
async def test_notify_platform(
hass: HomeAssistant,
config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test setup of the notify platform."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
@pytest.mark.parametrize(
("entity_id", "call_method", "call_args"),
[
(
"notify.test_user_party_chat",
"send_group_message",
{"group_id": UUID("1e87097c-4c03-4f8c-a475-67cc7da7f409")},
),
(
"notify.test_user_private_message_test_partymember_displayname",
"send_private_message",
{"to_user_id": UUID("ffce870c-3ff3-4fa4-bad1-87612e52b8e7")},
),
],
)
@freeze_time("2025-08-13T00:00:00+00:00")
async def test_send_message(
hass: HomeAssistant,
config_entry: MockConfigEntry,
habitica: AsyncMock,
entity_id: str,
call_method: str,
call_args: dict[str, Any],
) -> None:
"""Test send message."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_UNKNOWN
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
{
ATTR_ENTITY_ID: entity_id,
ATTR_MESSAGE: "Greetings, fellow adventurer",
},
blocking=True,
)
state = hass.states.get(entity_id)
assert state
assert state.state == "2025-08-13T00:00:00+00:00"
getattr(habitica, call_method).assert_called_once_with(
message="Greetings, fellow adventurer", **call_args
)
@pytest.mark.parametrize(
"exception",
[
ERROR_BAD_REQUEST,
ERROR_NOT_AUTHORIZED,
ERROR_NOT_FOUND,
ERROR_TOO_MANY_REQUESTS,
ClientError,
],
)
async def test_send_message_exceptions(
hass: HomeAssistant,
config_entry: MockConfigEntry,
habitica: AsyncMock,
exception: Exception,
) -> None:
"""Test send message exceptions."""
habitica.send_group_message.side_effect = exception
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
{
ATTR_ENTITY_ID: "notify.test_user_party_chat",
ATTR_MESSAGE: "Greetings, fellow adventurer",
},
blocking=True,
)
async def test_remove_stale_entities(
hass: HomeAssistant,
config_entry: MockConfigEntry,
habitica: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test removing stale private message entities."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert hass.states.get(
"notify.test_user_private_message_test_partymember_displayname"
)
habitica.get_group_members.return_value = HabiticaGroupMembersResponse.from_json(
await async_load_fixture(hass, "party_members_2.json", DOMAIN)
)
freezer.tick(timedelta(minutes=15))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
hass.states.get("notify.test_user_private_message_test_partymember_displayname")
is None
)