/
/
/
1"""Tests for the Home Assistant player control entity search."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from collections.abc import AsyncIterator, Callable
8from contextlib import asynccontextmanager
9from typing import TYPE_CHECKING, Any, cast
10from unittest.mock import AsyncMock, MagicMock, patch
11
12import pytest
13from music_assistant_models.auth import Scope
14from music_assistant_models.errors import InvalidDataError, ProviderUnavailableError
15
16from music_assistant.constants import CONF_LOG_LEVEL
17from music_assistant.helpers.api import APICommandHandler
18from music_assistant.providers.hass import (
19 CONF_AUTH_TOKEN,
20 CONF_URL,
21 CONF_VERIFY_SSL,
22 SEARCH_CONTROL_ENTITIES_COMMAND,
23 HomeAssistantProvider,
24 setup,
25)
26from music_assistant.providers.hass.constants import (
27 CONF_MUTE_CONTROLS,
28 CONF_POWER_CONTROLS,
29 CONF_VOLUME_CONTROLS,
30 CONTROL_DOMAINS,
31 MediaPlayerEntityFeature,
32)
33from music_assistant.providers.hass.control_entities import (
34 CONTROL_TYPE_CAPABILITIES,
35 CONTROL_TYPE_DOMAINS,
36 HassControlEntity,
37 HassControlEntityGroup,
38 HassControlEntitySearchResult,
39)
40from music_assistant.providers.hass.helpers import get_control_capabilities
41from tests.common import use_real_create_task
42
43if TYPE_CHECKING:
44 from hass_client.models import State
45
46REGISTRY_LIST_COMMAND = "config/entity_registry/list_for_display"
47
48FULL_MEDIA_PLAYER_FEATURES = int(
49 MediaPlayerEntityFeature.TURN_ON
50 | MediaPlayerEntityFeature.TURN_OFF
51 | MediaPlayerEntityFeature.VOLUME_SET
52 | MediaPlayerEntityFeature.VOLUME_MUTE
53)
54
55AREAS: list[dict[str, Any]] = [
56 {"area_id": "area_living", "name": "Living Room", "aliases": [], "picture": None},
57 {"area_id": "area_kitchen", "name": "Kitchen", "aliases": [], "picture": None},
58 {"area_id": "area_office", "name": "Office", "aliases": [], "picture": None},
59]
60
61DEVICES: list[dict[str, Any]] = [
62 {"id": "dev_living", "name": "Living Room Amp", "name_by_user": None, "area_id": "area_living"},
63 {
64 "id": "dev_kitchen",
65 "name": "Speaker",
66 "name_by_user": "Kitchen Speaker",
67 "area_id": "area_kitchen",
68 },
69 {"id": "dev_attic", "name": "Attic Box", "name_by_user": None, "area_id": None},
70]
71
72# entity_id -> (device_id, area_id override, friendly_name, extra state attributes)
73ENTITIES: dict[str, tuple[str | None, str | None, str, dict[str, Any]]] = {
74 "media_player.living_amp": (
75 "dev_living",
76 None,
77 "Amplifier",
78 {"supported_features": FULL_MEDIA_PLAYER_FEATURES},
79 ),
80 "media_player.mass_player": (
81 "dev_living",
82 None,
83 "MA Player",
84 {"supported_features": FULL_MEDIA_PLAYER_FEATURES, "mass_player_type": "player"},
85 ),
86 "media_player.featureless": (None, None, "Featureless Player", {"supported_features": 0}),
87 # an uninterpretable supported_features value must not take the whole search down
88 "media_player.broken_features": (None, None, "Broken Receiver", {"supported_features": []}),
89 "switch.kitchen_power": ("dev_kitchen", None, "Kitchen Power", {}),
90 "number.kitchen_volume": ("dev_kitchen", "area_office", "Kitchen Volume", {}),
91 "input_boolean.standalone": (None, "area_living", "Standalone Toggle", {}),
92 "input_number.attic_volume": ("dev_attic", None, "Attic Volume", {}),
93 # not a control domain at all, so it must never surface
94 "light.hallway": (None, "area_living", "Hallway Light", {}),
95}
96
97
98class _Cache:
99 """Provide the slice of the cache controller that @use_cache relies on."""
100
101 def __init__(self) -> None:
102 self.entries: dict[str, Any] = {}
103
104 async def get_with_freshness(self, key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
105 """Return the (data, is_fresh, found) triplet for the given key."""
106 # the real controller reads the cache database here, so yield like it does:
107 # @use_cache stores in the background, and only a yield lets that store land
108 await asyncio.sleep(0)
109 if key not in self.entries:
110 return None, False, False
111 return self.entries[key], True, True
112
113 async def set(self, key: str, data: Any, **kwargs: Any) -> None:
114 """Store data under the given key."""
115 self.entries[key] = data
116
117
118class _HomeAssistantClient:
119 """Serve the registries and states of a small Home Assistant install."""
120
121 def __init__(self) -> None:
122 self.connected = False
123 self.listener_started = asyncio.Event()
124 # entity_ids per subscribe_entities call, in call order
125 self.state_requests: list[list[str]] = []
126 self.registry_list_calls = 0
127 self.device_registry_calls = 0
128 self.area_registry_calls = 0
129 self.event_subscriptions: list[tuple[str, Callable[[dict[str, Any]], None]]] = []
130 self.send_command = AsyncMock(side_effect=self._send_command)
131
132 async def connect(self) -> None:
133 """Connect the client."""
134 self.connected = True
135
136 async def start_listening(self) -> None:
137 """Listen until the provider stops the listener task."""
138 self.listener_started.set()
139 await asyncio.Event().wait()
140
141 async def disconnect(self) -> None:
142 """Disconnect the client."""
143 self.connected = False
144
145 async def get_device_registry(self) -> list[dict[str, Any]]:
146 """Return the device registry listing."""
147 self.device_registry_calls += 1
148 return DEVICES
149
150 async def get_area_registry(self) -> list[dict[str, Any]]:
151 """Return the area registry listing."""
152 self.area_registry_calls += 1
153 return AREAS
154
155 async def subscribe_entities(
156 self, cb_func: Callable[[dict[str, Any]], None], entity_ids: list[str]
157 ) -> Callable[[], None]:
158 """Deliver the requested states and return the unsubscribe callable."""
159 self.state_requests.append(list(entity_ids))
160 initial = {
161 entity_id: {"s": "idle", "a": _attributes(entity_id)}
162 for entity_id in entity_ids
163 if entity_id in ENTITIES
164 }
165 asyncio.get_running_loop().call_soon(cb_func, {"a": initial})
166 return lambda: None
167
168 async def subscribe_events(
169 self, cb_func: Callable[[dict[str, Any]], None], event_type: str
170 ) -> Callable[[], None]:
171 """Register the event callback after command responses can be received."""
172 await self.listener_started.wait()
173 self.event_subscriptions.append((event_type, cb_func))
174 return lambda: None
175
176 def fire_event(self, event_type: str, data: dict[str, Any]) -> None:
177 """Deliver an event to every subscriber of the given event type."""
178 for subscribed_type, cb_func in self.event_subscriptions:
179 if subscribed_type == event_type:
180 cb_func({"event_type": event_type, "data": data})
181
182 async def _send_command(self, command: str, **kwargs: Any) -> Any:
183 """Return the response Home Assistant sends for the given websocket command."""
184 if command != REGISTRY_LIST_COMMAND:
185 return {}
186 await self.listener_started.wait()
187 self.registry_list_calls += 1
188 return {
189 "entity_categories": {},
190 "entities": [
191 {"ei": entity_id, "pl": "test", "di": device_id, "ai": area_id}
192 for entity_id, (device_id, area_id, _, _) in ENTITIES.items()
193 ],
194 }
195
196
197def _config() -> MagicMock:
198 """Return a provider config exposing the persisted values via get_value."""
199 persisted_values: dict[str, Any] = {
200 CONF_URL: "http://homeassistant.local:8123",
201 CONF_AUTH_TOKEN: "token",
202 CONF_VERIFY_SSL: True,
203 CONF_LOG_LEVEL: "GLOBAL",
204 CONF_POWER_CONTROLS: [],
205 CONF_MUTE_CONTROLS: [],
206 CONF_VOLUME_CONTROLS: [],
207 }
208 config = MagicMock()
209 config.instance_id = "hass--test"
210 config.name = "Home Assistant"
211 config.get_value.side_effect = persisted_values.get
212 config.values = {}
213 return config
214
215
216def _mass() -> MagicMock:
217 """Return the Music Assistant dependencies used during provider startup."""
218 mass = MagicMock()
219 mass.cache = _Cache()
220 mass.http_session = MagicMock()
221 mass.http_session_no_ssl = MagicMock()
222 use_real_create_task(mass)
223 mass.players.register_or_update_player_control = AsyncMock()
224 mass.config.get = MagicMock(return_value={})
225 mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
226 return mass
227
228
229def _attributes(entity_id: str) -> dict[str, Any]:
230 """Return the state attributes of the given entity."""
231 _, _, friendly_name, extra = ENTITIES[entity_id]
232 return {"friendly_name": friendly_name, **extra}
233
234
235def _entity_ids(groups: list[HassControlEntityGroup]) -> list[str]:
236 """Return the entity IDs of all groups, in result order."""
237 return [entity["entity_id"] for group in groups for entity in group["entities"]]
238
239
240@asynccontextmanager
241async def _start_provider() -> AsyncIterator[tuple[HomeAssistantProvider, _HomeAssistantClient]]:
242 """Start the provider with a connected mocked Home Assistant client."""
243 hass = _HomeAssistantClient()
244 manifest = MagicMock()
245 manifest.domain = "hass"
246 manifest.name = "Home Assistant"
247 with patch("music_assistant.providers.hass.HomeAssistantClient", return_value=hass):
248 provider = await setup(_mass(), manifest, _config())
249 assert isinstance(provider, HomeAssistantProvider)
250 async with asyncio.timeout(5):
251 await provider.handle_async_init()
252 try:
253 yield provider, hass
254 finally:
255 await provider.unload()
256
257
258async def test_search_command_is_exposed_on_the_api() -> None:
259 """Expose the search as a read-only API command for as long as the provider is loaded."""
260 async with _start_provider() as (provider, _):
261 mass = cast("MagicMock", provider.mass)
262 unregister = mass.register_api_command.return_value
263 mass.register_api_command.assert_called_once_with(
264 SEARCH_CONTROL_ENTITIES_COMMAND,
265 provider.search_control_entities,
266 required_scope=Scope.CONFIG_PROVIDERS_READ,
267 )
268 # the API layer must be able to resolve the command's signature and result type
269 handler = APICommandHandler.parse(
270 SEARCH_CONTROL_ENTITIES_COMMAND, provider.search_control_entities
271 )
272 assert handler.type_hints["return"] is HassControlEntitySearchResult
273 unregister.assert_not_called()
274
275 unregister.assert_called_once_with()
276
277
278@pytest.mark.parametrize(
279 ("search", "expected"),
280 [
281 # entity id
282 ("kitchen_power", ["switch.kitchen_power"]),
283 # friendly name
284 ("amplifier", ["media_player.living_amp"]),
285 # device name (name_by_user wins over name)
286 ("kitchen speaker", ["switch.kitchen_power", "number.kitchen_volume"]),
287 # area name, inherited from the device
288 ("living room", ["media_player.living_amp", "input_boolean.standalone"]),
289 # area name of an entity that overrides its device's area
290 ("office", ["number.kitchen_volume"]),
291 ],
292)
293async def test_search_matches_every_searchable_field(search: str, expected: list[str]) -> None:
294 """Match the search text against entity ID, entity name, device name and area name."""
295 async with _start_provider() as (provider, _):
296 result = await provider.search_control_entities(search=search)
297
298 assert sorted(_entity_ids(result["groups"])) == sorted(expected)
299 assert result["truncated"] is False
300
301
302async def test_search_is_case_insensitive() -> None:
303 """Match regardless of the casing of the search text."""
304 async with _start_provider() as (provider, _):
305 result = await provider.search_control_entities(search="AMPLIFIER")
306
307 assert _entity_ids(result["groups"]) == ["media_player.living_amp"]
308
309
310@pytest.mark.parametrize(
311 ("search", "expected"),
312 [
313 # surrounding whitespace is not part of any field
314 (" kitchen ", ["switch.kitchen_power", "number.kitchen_volume"]),
315 # the words may match different fields: entity name and area name here
316 ("kitchen office", ["number.kitchen_volume"]),
317 # every word has to match something
318 ("kitchen hallway", []),
319 ],
320)
321async def test_search_matches_every_word_separately(search: str, expected: list[str]) -> None:
322 """Require each word of the search text to match a field, not the text as a whole."""
323 async with _start_provider() as (provider, _):
324 result = await provider.search_control_entities(search=search)
325
326 assert sorted(_entity_ids(result["groups"])) == sorted(expected)
327
328
329async def test_search_survives_an_entity_with_invalid_features() -> None:
330 """Skip an entity reporting an uninterpretable supported_features value."""
331 async with _start_provider() as (provider, _):
332 result = await provider.search_control_entities(search="receiver")
333 usable = await provider.search_control_entities(search="amplifier")
334
335 assert _entity_ids(result["groups"]) == []
336 assert _entity_ids(usable["groups"]) == ["media_player.living_amp"]
337
338
339async def test_search_without_text_returns_all_eligible_entities() -> None:
340 """Return every entity that can serve a control role when no search text is given."""
341 async with _start_provider() as (provider, _):
342 result = await provider.search_control_entities()
343
344 assert sorted(_entity_ids(result["groups"])) == [
345 "input_boolean.standalone",
346 "input_number.attic_volume",
347 "media_player.living_amp",
348 "number.kitchen_volume",
349 "switch.kitchen_power",
350 ]
351
352
353async def test_search_excludes_music_assistant_players() -> None:
354 """Never offer Music Assistant's own exposed players as a control."""
355 async with _start_provider() as (provider, _):
356 result = await provider.search_control_entities(search="player")
357
358 assert "media_player.mass_player" not in _entity_ids(result["groups"])
359 # the entity without any usable feature is left out too
360 assert "media_player.featureless" not in _entity_ids(result["groups"])
361
362
363async def test_control_type_filters_on_capability() -> None:
364 """Return only the entities that can serve the requested control role."""
365 async with _start_provider() as (provider, hass):
366 hass.state_requests.clear()
367 volume = await provider.search_control_entities(control_type=CONF_VOLUME_CONTROLS)
368 volume_requests = [entity_id for request in hass.state_requests for entity_id in request]
369 power = await provider.search_control_entities(control_type=CONF_POWER_CONTROLS)
370 mute = await provider.search_control_entities(control_type=CONF_MUTE_CONTROLS)
371
372 assert sorted(_entity_ids(volume["groups"])) == [
373 "input_number.attic_volume",
374 "media_player.living_amp",
375 "number.kitchen_volume",
376 ]
377 assert sorted(_entity_ids(power["groups"])) == [
378 "input_boolean.standalone",
379 "media_player.living_amp",
380 "switch.kitchen_power",
381 ]
382 assert sorted(_entity_ids(mute["groups"])) == [
383 "input_boolean.standalone",
384 "media_player.living_amp",
385 "switch.kitchen_power",
386 ]
387 # a volume search must not sweep the states of the on/off-only domains
388 assert volume_requests == [
389 "input_number.attic_volume",
390 "media_player.broken_features",
391 "media_player.featureless",
392 "media_player.living_amp",
393 "media_player.mass_player",
394 "number.kitchen_volume",
395 ]
396
397
398def test_control_type_domains_cover_every_qualifying_domain() -> None:
399 """Keep the per-role domain narrowing a superset of what can qualify for that role."""
400 logger = logging.getLogger("test.hass")
401 for control_type, domains in CONTROL_TYPE_DOMAINS.items():
402 assert set(domains) <= set(CONTROL_DOMAINS)
403 has_capability = CONTROL_TYPE_CAPABILITIES[control_type]
404 for domain in CONTROL_DOMAINS:
405 # the most capable entity the domain can hold, so no role is missed
406 state = cast(
407 "State",
408 {
409 "entity_id": f"{domain}.probe",
410 "attributes": {"supported_features": FULL_MEDIA_PLAYER_FEATURES},
411 },
412 )
413 capabilities = get_control_capabilities(state, logger)
414 entity = cast(
415 "HassControlEntity",
416 {"entity_id": state["entity_id"], "name": "", **capabilities._asdict()},
417 )
418 if has_capability(entity):
419 assert domain in domains, f"{domain} can serve {control_type} but is not swept"
420
421
422async def test_control_type_reports_every_supported_role() -> None:
423 """Report all roles an entity can serve, not just the one that was searched for."""
424 async with _start_provider() as (provider, _):
425 result = await provider.search_control_entities(
426 search="kitchen_power", control_type=CONF_POWER_CONTROLS
427 )
428
429 entity = result["groups"][0]["entities"][0]
430 assert (entity["power"], entity["volume"], entity["mute"]) == (True, False, True)
431
432
433async def test_unknown_control_type_is_rejected() -> None:
434 """Reject a control type that does not exist instead of returning everything."""
435 async with _start_provider() as (provider, _):
436 with pytest.raises(InvalidDataError, match="Invalid control type"):
437 await provider.search_control_entities(control_type="brightness_controls")
438
439
440@pytest.mark.parametrize("limit", [0, -1])
441async def test_limit_below_one_is_rejected(limit: int) -> None:
442 """Reject a limit that can never yield a result instead of returning nothing."""
443 async with _start_provider() as (provider, _):
444 with pytest.raises(InvalidDataError, match="Invalid limit"):
445 await provider.search_control_entities(limit=limit)
446
447
448async def test_results_are_grouped_by_device_and_area() -> None:
449 """Group the entities by device and effective area, ordered by area, device and name."""
450 async with _start_provider() as (provider, _):
451 result = await provider.search_control_entities()
452
453 assert [
454 (group["device_id"], group["device_name"], group["area_name"], _entity_ids([group]))
455 for group in result["groups"]
456 ] == [
457 ("dev_kitchen", "Kitchen Speaker", "Kitchen", ["switch.kitchen_power"]),
458 # the entity inherits the area of its device
459 ("dev_living", "Living Room Amp", "Living Room", ["media_player.living_amp"]),
460 # an entity without a device falls back to a group of its own area
461 (None, None, "Living Room", ["input_boolean.standalone"]),
462 # the entity overrides the area it would inherit from its (Kitchen) device
463 ("dev_kitchen", "Kitchen Speaker", "Office", ["number.kitchen_volume"]),
464 # a device without an area sorts last
465 ("dev_attic", "Attic Box", None, ["input_number.attic_volume"]),
466 ]
467
468
469async def test_limit_caps_entities_and_reports_truncation() -> None:
470 """Cap the number of entities and tell the caller that matches were left out."""
471 async with _start_provider() as (provider, _):
472 limited = await provider.search_control_entities(limit=2)
473 exact = await provider.search_control_entities(limit=5)
474
475 assert _entity_ids(limited["groups"]) == ["switch.kitchen_power", "media_player.living_amp"]
476 assert limited["truncated"] is True
477 assert len(_entity_ids(exact["groups"])) == 5
478 assert exact["truncated"] is False
479
480
481async def test_limit_cannot_be_raised_past_the_maximum() -> None:
482 """Cap what a caller can ask for, so no search can return an oversized response."""
483 with patch(
484 "music_assistant.providers.hass.control_entities.SEARCH_CONTROL_ENTITIES_MAX_LIMIT", 2
485 ):
486 async with _start_provider() as (provider, _):
487 result = await provider.search_control_entities(limit=1000)
488
489 assert _entity_ids(result["groups"]) == ["switch.kitchen_power", "media_player.living_amp"]
490 assert result["truncated"] is True
491
492
493async def test_result_does_not_alias_the_cached_candidates() -> None:
494 """Keep a caller that edits the response from corrupting what the next search sees."""
495 async with _start_provider() as (provider, _):
496 first = await provider.search_control_entities(search="kitchen_power")
497 first["groups"][0]["entities"][0]["name"] = "Mutated"
498 second = await provider.search_control_entities(search="kitchen_power")
499
500 assert second["groups"][0]["entities"][0]["name"] == "Kitchen Power"
501
502
503async def test_consecutive_searches_share_one_state_sweep() -> None:
504 """Sweep the Home Assistant states once and serve the next search from the cache."""
505 async with _start_provider() as (provider, hass):
506 hass.state_requests.clear()
507 await provider.search_control_entities(search="living")
508 await provider.search_control_entities(search="living room")
509 sweeps = len(hass.state_requests)
510
511 assert sweeps == 1
512
513
514async def test_concurrent_searches_share_one_state_sweep() -> None:
515 """Let searches that arrive together wait for a single state sweep."""
516 async with _start_provider() as (provider, hass):
517 hass.state_requests.clear()
518 results = await asyncio.gather(
519 provider.search_control_entities(search="kitchen"),
520 provider.search_control_entities(search="living"),
521 )
522 sweeps = len(hass.state_requests)
523
524 assert sweeps == 1
525 assert _entity_ids(results[0]["groups"]) == [
526 "switch.kitchen_power",
527 "number.kitchen_volume",
528 ]
529
530
531async def test_power_and_mute_searches_share_one_state_sweep() -> None:
532 """Reuse one sweep for the control roles that live in the same entity domains."""
533 async with _start_provider() as (provider, hass):
534 hass.state_requests.clear()
535 await provider.search_control_entities(control_type=CONF_POWER_CONTROLS)
536 await provider.search_control_entities(control_type=CONF_MUTE_CONTROLS)
537 after_power_and_mute = len(hass.state_requests)
538 await provider.search_control_entities(control_type=CONF_VOLUME_CONTROLS)
539 after_volume = len(hass.state_requests)
540
541 assert after_power_and_mute == 1
542 # the volume roles live in other domains, so they need a sweep of their own
543 assert after_volume == 2
544
545
546async def test_entity_registry_change_forces_a_fresh_state_sweep() -> None:
547 """Sweep again after Home Assistant reports an entity registry change."""
548 async with _start_provider() as (provider, hass):
549 hass.state_requests.clear()
550 await provider.search_control_entities()
551 hass.fire_event(
552 "entity_registry_updated",
553 {"action": "update", "entity_id": "media_player.living_amp"},
554 )
555 await provider.search_control_entities()
556 sweeps = len(hass.state_requests)
557
558 assert sweeps == 2
559
560
561async def test_cached_candidates_expire() -> None:
562 """Sweep again once the cached candidates have outlived their TTL."""
563 with patch("music_assistant.providers.hass.control_entities.CONTROL_ENTITY_CACHE_TTL", 0):
564 async with _start_provider() as (provider, hass):
565 hass.state_requests.clear()
566 await provider.search_control_entities()
567 await provider.search_control_entities()
568 sweeps = len(hass.state_requests)
569
570 assert sweeps == 2
571
572
573async def test_search_reuses_the_cached_registries() -> None:
574 """Consult Home Assistant for the registries once and reuse them on the next search."""
575 async with _start_provider() as (provider, hass):
576 registry_calls_after_startup = hass.registry_list_calls
577 await provider.search_control_entities()
578 after_first = (
579 hass.registry_list_calls,
580 hass.device_registry_calls,
581 hass.area_registry_calls,
582 )
583 await provider.search_control_entities(search="kitchen")
584 after_second = (
585 hass.registry_list_calls,
586 hass.device_registry_calls,
587 hass.area_registry_calls,
588 )
589
590 assert registry_calls_after_startup == 1
591 assert after_first == (1, 1, 1)
592 assert after_second == (1, 1, 1)
593
594
595async def test_search_is_refused_once_the_provider_unloaded() -> None:
596 """Refuse a search that arrives after the provider was unloaded."""
597 async with _start_provider() as (provider, _):
598 await provider.search_control_entities()
599
600 with pytest.raises(ProviderUnavailableError):
601 await provider.search_control_entities()
602
603
604async def test_search_in_flight_during_unload_leaves_no_cache_behind() -> None:
605 """Refuse a search that was suspended over the unload instead of refilling its cache."""
606 entered_sweep = asyncio.Event()
607 release_sweep = asyncio.Event()
608
609 async with _start_provider() as (provider, _):
610 original_get_states = provider.get_states
611
612 async def _blocked_get_states(**kwargs: Any) -> Any:
613 entered_sweep.set()
614 await release_sweep.wait()
615 return await original_get_states(**kwargs)
616
617 with patch.object(provider, "get_states", _blocked_get_states):
618 search = asyncio.create_task(provider.search_control_entities())
619 async with asyncio.timeout(5):
620 await entered_sweep.wait()
621
622 release_sweep.set()
623 with pytest.raises(ProviderUnavailableError):
624 await search
625 assert provider._control_entity_search._entries == {}
626