/
/
/
1"""Tests for correlating natively connected devices to their Home Assistant representation."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from types import SimpleNamespace
8from typing import Any
9from unittest.mock import AsyncMock, MagicMock, patch
10
11from music_assistant_models.player import PlayerMedia
12
13from music_assistant.providers.hass import HomeAssistantProvider
14
15MEDIA_ANNOUNCE = 1048576
16MAC = "aa:bb:cc:dd:ee:ff"
17
18
19class _Cache:
20 """Provide the slice of the cache controller that @use_cache relies on."""
21
22 def __init__(self) -> None:
23 self.entries: dict[str, Any] = {}
24
25 async def get_with_freshness(self, key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
26 """Return the (data, is_fresh, found) triplet for the given key."""
27 await asyncio.sleep(0)
28 if key not in self.entries:
29 return None, False, False
30 return self.entries[key], True, True
31
32 async def set(self, key: str, data: Any, **kwargs: Any) -> None:
33 """Store data under the given key."""
34 self.entries[key] = data
35
36
37def _provider(
38 devices: list[dict[str, Any]],
39 entities: list[dict[str, Any]],
40 states: list[dict[str, Any]],
41) -> HomeAssistantProvider:
42 async def _send_command(command: str, **_kwargs: Any) -> dict[str, Any]:
43 assert command == "config/entity_registry/list_for_display"
44 # Home Assistant leaves disabled entities out of the registry listing
45 return {
46 "entity_categories": {},
47 "entities": [
48 {"ei": entity["entity_id"], "pl": entity["platform"], "di": entity["device_id"]}
49 for entity in entities
50 if entity["disabled_by"] is None
51 ],
52 }
53
54 provider = HomeAssistantProvider.__new__(HomeAssistantProvider)
55 provider.hass = SimpleNamespace(
56 get_device_registry=AsyncMock(return_value=devices),
57 send_command=AsyncMock(side_effect=_send_command),
58 )
59 provider._entity_registry = None
60 provider._entity_registry_lock = asyncio.Lock()
61 # the device registry lookup runs through @use_cache, which needs a cache to talk to
62 provider.config = SimpleNamespace(instance_id="hass--test") # type: ignore[assignment]
63 provider.manifest = SimpleNamespace(domain="hass") # type: ignore[assignment]
64 provider.mass = SimpleNamespace( # type: ignore[assignment]
65 cache=_Cache(),
66 create_task=lambda coro, **_kwargs: asyncio.ensure_future(coro),
67 )
68 provider.get_states = AsyncMock(return_value=states) # type: ignore[method-assign]
69 provider.logger = logging.getLogger("test.hass")
70 return provider
71
72
73def _device(device_id: str = "dev1", name_by_user: str | None = None) -> dict[str, Any]:
74 return {
75 "id": device_id,
76 "name": "Kitchen Speaker",
77 "name_by_user": name_by_user,
78 "connections": [["mac", MAC.upper()]],
79 }
80
81
82def _entity(
83 entity_id: str,
84 device_id: str = "dev1",
85 platform: str = "esphome",
86 disabled_by: str | None = None,
87) -> dict[str, Any]:
88 return {
89 "entity_id": entity_id,
90 "platform": platform,
91 "device_id": device_id,
92 "disabled_by": disabled_by,
93 }
94
95
96def _state(entity_id: str, supported_features: int) -> dict[str, Any]:
97 return {"entity_id": entity_id, "attributes": {"supported_features": supported_features}}
98
99
100async def test_correlates_name_and_announce_entity() -> None:
101 """A device is matched case-insensitively on MAC with its announce-capable entity."""
102 provider = _provider(
103 [_device(name_by_user="Kitchen")],
104 [_entity("media_player.kitchen")],
105 [_state("media_player.kitchen", MEDIA_ANNOUNCE)],
106 )
107 result = await provider.get_media_player_device_infos([MAC.upper()], platform="esphome")
108 assert result == {MAC: {"name": "Kitchen", "announce_entity_id": "media_player.kitchen"}}
109
110
111async def test_entity_without_announce_support() -> None:
112 """A matched device without an announce-capable entity still yields its name."""
113 provider = _provider(
114 [_device()],
115 [_entity("media_player.kitchen")],
116 [_state("media_player.kitchen", 0)],
117 )
118 result = await provider.get_media_player_device_infos([MAC], platform="esphome")
119 assert result == {MAC: {"name": "Kitchen Speaker", "announce_entity_id": None}}
120
121
122async def test_ignores_disabled_and_foreign_entities() -> None:
123 """Disabled entities and entities of other integrations are not considered."""
124 provider = _provider(
125 [_device()],
126 [
127 _entity("media_player.disabled", disabled_by="user"),
128 _entity("media_player.other", platform="cast"),
129 _entity("sensor.kitchen_temperature"),
130 ],
131 [],
132 )
133 result = await provider.get_media_player_device_infos([MAC], platform="esphome")
134 assert result == {MAC: {"name": "Kitchen Speaker", "announce_entity_id": None}}
135
136
137async def test_unknown_devices_are_absent() -> None:
138 """Devices unknown to Home Assistant are absent from the result."""
139 provider = _provider([_device()], [], [])
140 result = await provider.get_media_player_device_infos(["11:22:33:44:55:66"], platform="esphome")
141 assert result == {}
142
143
144async def test_empty_input_skips_registry_fetch() -> None:
145 """An empty lookup does not hit the Home Assistant registries."""
146 provider = _provider([], [], [])
147 assert await provider.get_media_player_device_infos([], platform="esphome") == {}
148 provider.hass.get_device_registry.assert_not_awaited()
149
150
151async def test_play_announcement_on_entity() -> None:
152 """An announcement is played via HA's announce feature and awaited for its duration."""
153 provider = HomeAssistantProvider.__new__(HomeAssistantProvider)
154 provider.hass = SimpleNamespace(call_service=AsyncMock())
155 provider.mass = MagicMock()
156 provider.mass.streams.get_announcement_duration = AsyncMock(return_value=7)
157 announcement = PlayerMedia(uri="http://mass.local/announcement.mp3")
158 with patch("music_assistant.providers.hass.asyncio.sleep", AsyncMock()) as sleep:
159 await provider.play_announcement_on_entity("media_player.kitchen", announcement)
160 provider.hass.call_service.assert_awaited_once_with(
161 domain="media_player",
162 service="play_media",
163 service_data={
164 "media_content_id": "http://mass.local/announcement.mp3",
165 "media_content_type": "music",
166 "announce": True,
167 },
168 target={"entity_id": "media_player.kitchen"},
169 )
170 # the length is resolved after the announcement was handed to the entity
171 provider.mass.streams.get_announcement_duration.assert_awaited_once_with(announcement)
172 sleep.assert_awaited_once_with(7)
173