music-assistant-server

1"""Tests for the Home Assistant provider."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7from collections.abc import AsyncIterator, Callable
8from contextlib import asynccontextmanager
9from math import ceil
10from typing import Any, cast
11from unittest.mock import AsyncMock, MagicMock, patch
12
13import pytest
14from hass_client.exceptions import BaseHassClientError
15from music_assistant_models.enums import EventType, ProviderFeature
16from music_assistant_models.errors import SetupFailedError, UnsupportedFeaturedException
17
18from music_assistant.constants import CONF_LOG_LEVEL
19from music_assistant.providers.hass import (
20    CONF_AUTH_TOKEN,
21    CONF_URL,
22    CONF_VERIFY_SSL,
23    STATE_FETCH_BATCH_SIZE,
24    HassRegistryEntity,
25    HomeAssistantProvider,
26    setup,
27)
28from music_assistant.providers.hass.constants import (
29    CONF_MUTE_CONTROLS,
30    CONF_POWER_CONTROLS,
31    CONF_VOLUME_CONTROLS,
32)
33from tests.common import use_real_create_task
34
35LAST_CHANGED = 1683832716.072648
36LAST_CHANGED_ISO = "2023-05-11T19:18:36.072648+00:00"
37CONTEXT_ID = "01H0640ES8JCY1NGTNW3V41T5T"
38REGISTRY_LIST_COMMAND = "config/entity_registry/list_for_display"
39REGISTRY_ENTRIES_COMMAND = "config/entity_registry/get_entries"
40DEVICE_REGISTRY_LIST = "get_device_registry"
41
42
43def _state(entity_id: str, friendly_name: str) -> dict[str, Any]:
44    """Return a Home Assistant entity state."""
45    return {
46        "entity_id": entity_id,
47        "state": "idle",
48        "attributes": {"friendly_name": friendly_name},
49    }
50
51
52def _compressed(state: dict[str, Any]) -> dict[str, Any]:
53    """Return the compressed form Home Assistant sends for the given entity state."""
54    return {
55        "s": state["state"],
56        "a": state["attributes"],
57        "lc": LAST_CHANGED,
58        "c": CONTEXT_ID,
59    }
60
61
62def _config(**values: Any) -> MagicMock:
63    """Return a provider config exposing the given values via get_value (entry defaults)."""
64    persisted_values = {
65        CONF_URL: "http://homeassistant.local:8123",
66        CONF_AUTH_TOKEN: "token",
67        CONF_VERIFY_SSL: True,
68        CONF_LOG_LEVEL: "GLOBAL",
69        CONF_POWER_CONTROLS: [],
70        CONF_MUTE_CONTROLS: [],
71        CONF_VOLUME_CONTROLS: [],
72        **values,
73    }
74    config = MagicMock()
75    config.instance_id = "hass--test"
76    config.name = "Home Assistant"
77    config.get_value.side_effect = persisted_values.get
78    # get_setup_value falls through to config.values/get_value when setup_data is empty
79    config.values = {}
80    return config
81
82
83class _Cache:
84    """Provide the slice of the cache controller that @use_cache relies on."""
85
86    def __init__(self) -> None:
87        self.entries: dict[str, Any] = {}
88        self.fresh = True
89
90    async def get_with_freshness(self, key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
91        """Return the (data, is_fresh, found) triplet for the given key."""
92        # the real controller reads the cache database here, so yield like it does:
93        # @use_cache stores in the background, and only a yield lets that store land
94        await asyncio.sleep(0)
95        if key not in self.entries:
96            return None, False, False
97        if not self.fresh and not kwargs.get("include_expired"):
98            return None, False, False
99        return self.entries[key], self.fresh, True
100
101    async def set(self, key: str, data: Any, **kwargs: Any) -> None:
102        """Store data under the given key."""
103        # the real controller serializes on a worker thread and then writes the cache
104        # database, so a store lands well after the call that scheduled it returned
105        await asyncio.sleep(0)
106        await asyncio.sleep(0)
107        self.entries[key] = data
108
109    def expire_all(self) -> None:
110        """Mark every stored entry as no longer fresh."""
111        self.fresh = False
112
113
114def _mass() -> MagicMock:
115    """Return the Music Assistant dependencies used during provider startup."""
116    mass = MagicMock()
117    mass.cache = _Cache()
118    mass.http_session = MagicMock()
119    mass.http_session_no_ssl = MagicMock()
120    use_real_create_task(mass)
121    mass.players.register_or_update_player_control = AsyncMock()
122    # get_setup_value reads the (empty, here) live setup_data blob from the store, then
123    # falls through to the provider config mock's get_value for the persisted test values
124    mass.config.get = MagicMock(return_value={})
125    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
126    return mass
127
128
129class _HomeAssistantClient:
130    """Provide lifecycle-aware Home Assistant behavior for provider tests."""
131
132    def __init__(
133        self,
134        states: list[dict[str, Any]],
135        registry_error: Exception | None = None,
136        listener_error: Exception | None = None,
137        connect_error: Exception | None = None,
138        block_registry: bool = False,
139    ) -> None:
140        self.connected = False
141        self.disconnected = False
142        self.listener_started = asyncio.Event()
143        self.listener_cancelled = asyncio.Event()
144        self.listener_stopped = asyncio.Event()
145        self.registry_started = asyncio.Event()
146        self.registry_cancelled = asyncio.Event()
147        self.registry_stopped = asyncio.Event()
148        self.calls: list[str] = []
149        self.subscribed = asyncio.Event()
150        # entity_ids per subscribe_entities call and per invoked unsubscribe callable
151        self.subscriptions: list[list[str]] = []
152        self.unsubscribed: list[list[str]] = []
153        self.active_subscriptions = 0
154        # (event_type, callback) per subscribe_events call
155        self.event_subscriptions: list[tuple[str, Callable[[dict[str, Any]], None]]] = []
156        self.active_event_subscriptions = 0
157        # compressed states keyed by entity_id, as delivered over the websocket
158        self.compressed_states = {state["entity_id"]: _compressed(state) for state in states}
159        # entities Home Assistant reports as disabled, so leaves out of the registry listing
160        self.disabled_entities: set[str] = set()
161        # devices as returned in full by the device registry listing
162        self.devices: list[dict[str, Any]] = []
163        # events delivered ahead of a subscription's initial state message
164        self.leading_events: list[dict[str, Any]] = []
165        self.deliver_initial_states = True
166        self._registry_error = registry_error
167        self._listener_error = listener_error
168        self._connect_error = connect_error
169        self._registry_result = (
170            asyncio.get_running_loop().create_future() if block_registry else None
171        )
172        # resolve this to make an already running listener return, as a lost connection does
173        self.connection_lost: asyncio.Future[None] = asyncio.get_running_loop().create_future()
174        self.send_command = AsyncMock(side_effect=self._send_command)
175
176    async def connect(self) -> None:
177        """Connect the client."""
178        self.calls.append("connect")
179        if self._connect_error:
180            raise self._connect_error
181        self.connected = True
182
183    async def start_listening(self) -> None:
184        """Listen until the connection is lost or the provider stops the listener task."""
185        self.calls.append("start_listening")
186        self.listener_started.set()
187        try:
188            if self._listener_error:
189                raise self._listener_error
190            await self.connection_lost
191        except asyncio.CancelledError:
192            self.listener_cancelled.set()
193            raise
194        finally:
195            if self._registry_result and not self._registry_result.done():
196                self._registry_result.cancel()
197            self.listener_stopped.set()
198
199    async def subscribe_entities(
200        self, cb_func: Callable[[dict[str, Any]], None], entity_ids: list[str]
201    ) -> Callable[[], None]:
202        """Deliver the subscription's state messages and return the unsubscribe callable."""
203        self.calls.append("subscribe_entities")
204        self.subscriptions.append(list(entity_ids))
205        self.active_subscriptions += 1
206        self.subscribed.set()
207        loop = asyncio.get_running_loop()
208        for event in self.leading_events:
209            loop.call_soon(cb_func, event)
210        if self.deliver_initial_states:
211            initial = {
212                entity_id: self.compressed_states[entity_id]
213                for entity_id in entity_ids
214                if entity_id in self.compressed_states
215            }
216            loop.call_soon(cb_func, {"a": initial})
217
218        def _unsubscribe() -> None:
219            self.calls.append("unsubscribe_entities")
220            self.unsubscribed.append(list(entity_ids))
221            self.active_subscriptions -= 1
222
223        return _unsubscribe
224
225    async def subscribe_events(
226        self, cb_func: Callable[[dict[str, Any]], None], event_type: str
227    ) -> Callable[[], None]:
228        """Register the event callback after command responses can be received."""
229        await self.listener_started.wait()
230        self.calls.append("subscribe_events")
231        self.event_subscriptions.append((event_type, cb_func))
232        self.active_event_subscriptions += 1
233
234        def _unsubscribe() -> None:
235            self.calls.append("unsubscribe_events")
236            self.active_event_subscriptions -= 1
237
238        return _unsubscribe
239
240    async def get_device_registry(self) -> list[dict[str, Any]]:
241        """Return the full device registry listing."""
242        self.calls.append(DEVICE_REGISTRY_LIST)
243        return list(self.devices)
244
245    def fire_event(self, event_type: str, data: dict[str, Any]) -> None:
246        """Deliver an event to every subscriber of the given event type."""
247        for subscribed_type, cb_func in self.event_subscriptions:
248            if subscribed_type == event_type:
249                cb_func({"event_type": event_type, "data": data})
250
251    async def disconnect(self) -> None:
252        """Disconnect the client."""
253        self.calls.append("disconnect")
254        self.connected = False
255        self.disconnected = True
256
257    async def _send_command(self, command: str, **kwargs: Any) -> Any:
258        """Return the response Home Assistant sends for the given websocket command."""
259        if command == REGISTRY_LIST_COMMAND:
260            return await self._registry_for_display()
261        self.calls.append(command)
262        if command == REGISTRY_ENTRIES_COMMAND:
263            return self._registry_entries(cast("list[str]", kwargs["entity_ids"]))
264        return {"response": {"data": "answer"}}
265
266    async def _registry_for_display(self) -> dict[str, Any]:
267        """Return the entity registry listing after command responses can be received."""
268        await self.listener_started.wait()
269        self.calls.append(REGISTRY_LIST_COMMAND)
270        self.registry_started.set()
271        try:
272            if self._registry_result:
273                await self._registry_result
274            if self._registry_error:
275                raise self._registry_error
276            return {
277                "entity_categories": {},
278                "entities": [
279                    {"ei": entity_id, "pl": "test"}
280                    for entity_id in self.compressed_states
281                    if entity_id not in self.disabled_entities
282                ],
283            }
284        except asyncio.CancelledError:
285            self.registry_cancelled.set()
286            raise
287        finally:
288            self.registry_stopped.set()
289
290    def _registry_entries(self, entity_ids: list[str]) -> dict[str, dict[str, Any] | None]:
291        """Return the full registry entry of every requested entity, None when unknown."""
292        return {
293            entity_id: (
294                {
295                    "entity_id": entity_id,
296                    "id": f"registry_id_{entity_id}",
297                    "platform": "test",
298                    "device_id": "device_id",
299                    "config_entry_id": "config_entry_id",
300                }
301                if entity_id in self.compressed_states
302                else None
303            )
304            for entity_id in entity_ids
305        }
306
307
308@asynccontextmanager
309async def _start_provider(
310    states: list[dict[str, Any]], **config_values: Any
311) -> AsyncIterator[tuple[HomeAssistantProvider, _HomeAssistantClient]]:
312    """Start the provider with a connected mocked Home Assistant client."""
313    hass = _HomeAssistantClient(states)
314    mass = _mass()
315    manifest = MagicMock()
316    manifest.domain = "hass"
317    manifest.name = "Home Assistant"
318    with patch("music_assistant.providers.hass.HomeAssistantClient", return_value=hass):
319        provider = await setup(mass, manifest, _config(**config_values))
320        assert isinstance(provider, HomeAssistantProvider)
321        async with asyncio.timeout(1):
322            await provider.handle_async_init()
323    try:
324        yield provider, hass
325    finally:
326        await provider.unload()
327
328
329async def _wait_for_stored(provider: HomeAssistantProvider) -> None:
330    """Wait until the background store of the device listing has landed in the cache."""
331    cache = cast("_Cache", provider.mass.cache)
332    async with asyncio.timeout(1):
333        while not cache.entries:
334            await asyncio.sleep(0)
335
336
337def _hold_back_registry_fetch(hass: _HomeAssistantClient) -> asyncio.Future[None]:
338    """Hold back the next registry listing and return the future that releases it."""
339    registry_response: asyncio.Future[None] = asyncio.get_running_loop().create_future()
340    hass._registry_result = registry_response
341    # the startup fetch left the event set, so re-arm it for the fetch under test
342    hass.registry_started.clear()
343    return registry_response
344
345
346async def _wait_for_registry_fetch(hass: _HomeAssistantClient) -> None:
347    """Wait until the held back registry listing is in flight."""
348    async with asyncio.timeout(1):
349        await hass.registry_started.wait()
350
351
352def _registry_event(
353    entity_id: str, action: str = "update", changes: dict[str, Any] | None = None
354) -> dict[str, Any]:
355    """Return the data Home Assistant sends in an entity_registry_updated event."""
356    data: dict[str, Any] = {"action": action, "entity_id": entity_id}
357    if action == "update":
358        # an update carries the old value of every field it touched
359        data["changes"] = changes or {}
360    return data
361
362
363async def _fire_registry_update(
364    provider: HomeAssistantProvider,
365    hass: _HomeAssistantClient,
366    entity_id: str,
367    action: str,
368) -> None:
369    """Deliver an entity registry update and wait for the engine rebuild it schedules."""
370    with patch("music_assistant.providers.hass.ENGINE_REFRESH_DEBOUNCE", 0):
371        hass.fire_event("entity_registry_updated", _registry_event(entity_id, action))
372        assert provider._engine_refresh_task is not None
373        async with asyncio.timeout(1):
374            await provider._engine_refresh_task
375
376
377def _providers_updated_events(provider: HomeAssistantProvider) -> list[Any]:
378    """
379    Return the PROVIDERS_UPDATED events the provider signalled so far.
380
381    :param provider: The provider whose Music Assistant mock is inspected.
382    """
383    signal_event = cast("MagicMock", provider.mass.signal_event)
384    return [
385        call
386        for call in signal_event.call_args_list
387        if call.args[:1] == (EventType.PROVIDERS_UPDATED,)
388    ]
389
390
391async def test_feature_resolution_starts_listener_first() -> None:
392    """Resolve startup features only after the Home Assistant listener starts."""
393    states = [
394        _state("ai_task.default", "Default AI"),
395        _state("tts.default", "Default TTS"),
396    ]
397
398    async with _start_provider(states) as (provider, hass):
399        assert hass.calls[:4] == [
400            "connect",
401            "start_listening",
402            "subscribe_events",
403            REGISTRY_LIST_COMMAND,
404        ]
405        assert ProviderFeature.AI_QUERY in provider.supported_features
406        assert ProviderFeature.TTS in provider.supported_features
407
408
409async def test_feature_resolution_failure_cleans_up_connection() -> None:
410    """Clean up the listener and connection when feature resolution fails."""
411    hass = _HomeAssistantClient([], BaseHassClientError("Unable to load Home Assistant states"))
412    manifest = MagicMock()
413    manifest.domain = "hass"
414    manifest.name = "Home Assistant"
415    with patch("music_assistant.providers.hass.HomeAssistantClient", return_value=hass):
416        provider = await setup(_mass(), manifest, _config())
417        assert isinstance(provider, HomeAssistantProvider)
418
419        with pytest.raises(SetupFailedError, match="Unable to load Home Assistant states"):
420            async with asyncio.timeout(1):
421                await provider.handle_async_init()
422
423    assert hass.listener_started.is_set()
424    assert hass.listener_stopped.is_set()
425    assert hass.disconnected
426    assert hass.calls == [
427        "connect",
428        "start_listening",
429        "subscribe_events",
430        REGISTRY_LIST_COMMAND,
431        "unsubscribe_events",
432        "disconnect",
433    ]
434    assert provider._listen_task is None
435
436
437async def test_listener_failure_does_not_mask_feature_resolution_failure() -> None:
438    """Preserve the startup error when the listener also fails."""
439    hass = _HomeAssistantClient(
440        [],
441        BaseHassClientError("Unable to load Home Assistant states"),
442        RuntimeError("Listener failed"),
443    )
444    manifest = MagicMock()
445    manifest.domain = "hass"
446    manifest.name = "Home Assistant"
447    with patch("music_assistant.providers.hass.HomeAssistantClient", return_value=hass):
448        provider = await setup(_mass(), manifest, _config())
449        assert isinstance(provider, HomeAssistantProvider)
450
451        with pytest.raises(SetupFailedError, match="Unable to load Home Assistant states"):
452            async with asyncio.timeout(1):
453                await provider.handle_async_init()
454
455    assert hass.disconnected
456    assert provider._listen_task is None
457
458
459async def test_listener_exit_terminates_pending_feature_resolution() -> None:
460    """Fail startup when the listener exits while feature resolution is pending."""
461    hass = _HomeAssistantClient(
462        [],
463        listener_error=BaseHassClientError("Listener failed"),
464        block_registry=True,
465    )
466    mass = _mass()
467    manifest = MagicMock()
468    manifest.domain = "hass"
469    manifest.name = "Home Assistant"
470    with patch("music_assistant.providers.hass.HomeAssistantClient", return_value=hass):
471        provider = await setup(mass, manifest, _config())
472        assert isinstance(provider, HomeAssistantProvider)
473
474        with pytest.raises(SetupFailedError, match="listener stopped during startup"):
475            async with asyncio.timeout(1):
476                await provider.handle_async_init()
477
478    assert hass.listener_started.is_set()
479    assert hass.listener_stopped.is_set()
480    assert hass.registry_stopped.is_set()
481    assert hass.disconnected
482    assert provider._listen_task is None
483    mass.call_later.assert_not_called()
484
485
486async def test_feature_resolution_timeout_cleans_up_connection() -> None:
487    """Clean up startup when Home Assistant feature resolution times out."""
488    hass = _HomeAssistantClient([], block_registry=True)
489    mass = _mass()
490    manifest = MagicMock()
491    manifest.domain = "hass"
492    manifest.name = "Home Assistant"
493    with (
494        patch("music_assistant.providers.hass.HomeAssistantClient", return_value=hass),
495        patch("music_assistant.providers.hass.FEATURE_DISCOVERY_TIMEOUT", 0.1),
496    ):
497        provider = await setup(mass, manifest, _config())
498        assert isinstance(provider, HomeAssistantProvider)
499        init_task = asyncio.create_task(provider.handle_async_init())
500        async with asyncio.timeout(1):
501            await hass.registry_started.wait()
502
503        with pytest.raises(
504            SetupFailedError, match="Timed out while resolving Home Assistant feature entities"
505        ):
506            await init_task
507
508    assert hass.registry_stopped.is_set()
509    assert hass.registry_cancelled.is_set()
510    assert hass.listener_stopped.is_set()
511    assert hass.listener_cancelled.is_set()
512    assert hass.disconnected
513    assert provider._listen_task is None
514    mass.call_later.assert_not_called()
515
516
517async def test_feature_resolution_cancellation_cleans_up_connection() -> None:
518    """Clean up startup when Home Assistant initialization is cancelled."""
519    hass = _HomeAssistantClient([], block_registry=True)
520    mass = _mass()
521    manifest = MagicMock()
522    manifest.domain = "hass"
523    manifest.name = "Home Assistant"
524    with patch("music_assistant.providers.hass.HomeAssistantClient", return_value=hass):
525        provider = await setup(mass, manifest, _config())
526        assert isinstance(provider, HomeAssistantProvider)
527        init_task = asyncio.create_task(provider.handle_async_init())
528        async with asyncio.timeout(1):
529            await hass.registry_started.wait()
530        init_task.cancel()
531
532        with pytest.raises(asyncio.CancelledError):
533            await init_task
534
535    assert hass.registry_stopped.is_set()
536    assert hass.registry_cancelled.is_set()
537    assert hass.listener_stopped.is_set()
538    assert hass.listener_cancelled.is_set()
539    assert hass.disconnected
540    assert provider._listen_task is None
541    mass.call_later.assert_not_called()
542
543
544async def test_connection_failure_cleans_up_client() -> None:
545    """Clean up the client when connecting to Home Assistant fails."""
546    hass = _HomeAssistantClient([], connect_error=BaseHassClientError("Unable to connect"))
547    manifest = MagicMock()
548    manifest.domain = "hass"
549    manifest.name = "Home Assistant"
550    with patch("music_assistant.providers.hass.HomeAssistantClient", return_value=hass):
551        provider = await setup(_mass(), manifest, _config())
552        assert isinstance(provider, HomeAssistantProvider)
553
554        with pytest.raises(SetupFailedError, match="Unable to connect"):
555            await provider.handle_async_init()
556
557    assert hass.disconnected
558    assert provider._listen_task is None
559
560
561async def test_lost_connection_arms_the_reconnect_under_the_load_task_id() -> None:
562    """A lost connection arms the reconnect so a (re)load starting first cancels it."""
563    async with _start_provider([]) as (provider, hass):
564        mass = cast("MagicMock", provider.mass)
565        mass.call_later.reset_mock()
566        assert provider._listen_task is not None
567
568        hass.connection_lost.set_exception(BaseHassClientError("connection lost"))
569        async with asyncio.timeout(1):
570            await provider._listen_task
571
572        assert provider.available is False
573        retry = mass.call_later.call_args
574        assert retry.args == (5, mass.load_provider, "hass--test")
575        assert retry.kwargs == {"allow_retry": True, "task_id": "load_provider_hass--test"}
576
577
578async def test_engines_are_listed_for_every_feature_entity() -> None:
579    """Expose every Home Assistant TTS and AI Task entity as an engine."""
580    states = [
581        _state("tts.piper", "Piper"),
582        _state("tts.cloud", "Home Assistant Cloud"),
583        _state("ai_task.openai", "OpenAI"),
584        _state("sensor.example", "Example"),
585    ]
586
587    async with _start_provider(states) as (provider, hass):
588        calls_before = list(hass.calls)
589        tts_engines = await provider.get_tts_engines()
590        ai_engines = await provider.get_ai_engines()
591
592        # listing engines is a hot path for consumers: it must not touch Home Assistant
593        assert hass.calls == calls_before
594        assert [(engine.id, engine.name) for engine in tts_engines] == [
595            ("tts.cloud", "Home Assistant Cloud (tts.cloud)"),
596            ("tts.piper", "Piper (tts.piper)"),
597        ]
598        assert [(engine.id, engine.name) for engine in ai_engines] == [
599            ("ai_task.openai", "OpenAI (ai_task.openai)")
600        ]
601        assert tts_engines[0].uid == "hass--test/tts.cloud"
602
603
604async def test_engine_without_friendly_name_falls_back_to_entity_id() -> None:
605    """Name an engine after its entity_id when Home Assistant has no friendly name."""
606    unnamed = {"entity_id": "tts.unnamed", "state": "idle", "attributes": {}}
607
608    async with _start_provider([unnamed]) as (provider, _):
609        engines = await provider.get_tts_engines()
610
611        assert [engine.name for engine in engines] == ["tts.unnamed"]
612
613
614async def test_ai_query_uses_the_first_engine_by_default() -> None:
615    """Send an AI query to the first available engine when none is requested."""
616    states = [_state("ai_task.first", "First"), _state("ai_task.second", "Second")]
617
618    async with _start_provider(states) as (provider, hass):
619        result = await provider.ai_query("What is this song?")
620
621        assert ProviderFeature.AI_QUERY in provider.supported_features
622        assert result == "answer"
623        hass.send_command.assert_awaited_with(
624            "call_service",
625            domain="ai_task",
626            service="generate_data",
627            service_data={
628                "task_name": "music_assistant",
629                "instructions": "What is this song?",
630                "entity_id": "ai_task.first",
631            },
632            return_response=True,
633        )
634
635
636async def test_ai_query_uses_the_requested_engine() -> None:
637    """Send an AI query to the requested engine."""
638    states = [_state("ai_task.first", "First"), _state("ai_task.second", "Second")]
639
640    async with _start_provider(states) as (provider, hass):
641        await provider.ai_query("What is this song?", engine_id="ai_task.second")
642
643        service_data = hass.send_command.call_args.kwargs["service_data"]
644        assert service_data["entity_id"] == "ai_task.second"
645
646
647async def test_ai_query_not_advertised_without_entity() -> None:
648    """Do not advertise AI queries when Home Assistant has no AI Task entity."""
649    async with _start_provider([_state("sensor.example", "Example")]) as (provider, _):
650        assert ProviderFeature.AI_QUERY not in provider.supported_features
651
652        with pytest.raises(UnsupportedFeaturedException):
653            await provider.ai_query("What is this song?")
654
655
656def _mock_tts_response(provider: HomeAssistantProvider) -> MagicMock:
657    """Let the Home Assistant tts_get_url endpoint return a URL and return the post mock."""
658    response = AsyncMock()
659    response.raise_for_status = MagicMock()
660    response.json.return_value = {"url": "http://homeassistant.local/tts.mp3"}
661    post = cast("MagicMock", provider.mass.http_session.post)
662    post.return_value.__aenter__.return_value = response
663    return post
664
665
666async def test_tts_uses_the_first_engine_by_default() -> None:
667    """Render speech on the first available engine when none is requested."""
668    states = [_state("tts.first", "First"), _state("tts.second", "Second")]
669
670    async with _start_provider(states) as (provider, _):
671        post = _mock_tts_response(provider)
672
673        stream = await provider.get_tts_message("Hello")
674
675        assert ProviderFeature.TTS in provider.supported_features
676        assert stream.path == "http://homeassistant.local/tts.mp3"
677        post.assert_called_once()
678        request = post.call_args
679        assert request.args == ("http://homeassistant.local:8123/api/tts_get_url",)
680        assert request.kwargs["json"] == {"engine_id": "tts.first", "message": "Hello"}
681
682
683async def test_tts_uses_the_requested_engine() -> None:
684    """Render speech on the requested engine."""
685    states = [_state("tts.first", "First"), _state("tts.second", "Second")]
686
687    async with _start_provider(states) as (provider, _):
688        post = _mock_tts_response(provider)
689
690        await provider.get_tts_message("Hello", engine_id="tts.second")
691
692        assert post.call_args.kwargs["json"] == {"engine_id": "tts.second", "message": "Hello"}
693
694
695async def test_tts_not_advertised_without_entity() -> None:
696    """Do not advertise TTS when Home Assistant has no TTS entity."""
697    async with _start_provider([_state("sensor.example", "Example")]) as (provider, _):
698        assert ProviderFeature.TTS not in provider.supported_features
699
700        with pytest.raises(UnsupportedFeaturedException):
701            await provider.get_tts_message("Hello")
702
703
704async def test_registry_update_refreshes_the_engines() -> None:
705    """Pick up a feature entity that Home Assistant adds after startup."""
706    async with _start_provider([_state("sensor.example", "Example")]) as (provider, hass):
707        await provider.loaded_in_mass()
708        assert ProviderFeature.TTS not in provider.supported_features
709
710        hass.compressed_states["tts.new"] = _compressed(_state("tts.new", "New"))
711        await _fire_registry_update(provider, hass, "tts.new", "create")
712
713        assert [engine.id for engine in await provider.get_tts_engines()] == ["tts.new"]
714        assert ProviderFeature.TTS in provider.supported_features
715
716
717async def test_registry_update_discards_the_feature_of_a_removed_engine() -> None:
718    """Stop advertising a feature once its last Home Assistant entity is gone."""
719    states = [_state("tts.only", "Only"), _state("ai_task.only", "Only")]
720
721    async with _start_provider(states) as (provider, hass):
722        await provider.loaded_in_mass()
723        assert ProviderFeature.TTS in provider.supported_features
724
725        del hass.compressed_states["tts.only"]
726        await _fire_registry_update(provider, hass, "tts.only", "remove")
727
728        assert await provider.get_tts_engines() == []
729        assert ProviderFeature.TTS not in provider.supported_features
730        # the AI Task entity is untouched, so its feature stays declared
731        assert ProviderFeature.AI_QUERY in provider.supported_features
732
733
734async def test_changed_engines_notify_the_consumers_once() -> None:
735    """Announce a refresh that changed the engine lists with a single PROVIDERS_UPDATED."""
736    states = [_state("tts.only", "Only"), _state("ai_task.only", "Only")]
737
738    async with _start_provider(states) as (provider, hass):
739        await provider.loaded_in_mass()
740        mass = cast("MagicMock", provider.mass)
741        mass.signal_event.reset_mock()
742
743        del hass.compressed_states["tts.only"]
744        await _fire_registry_update(provider, hass, "tts.only", "remove")
745
746        events = _providers_updated_events(provider)
747        assert len(events) == 1
748        assert events[0].kwargs["data"] is mass.get_providers.return_value
749
750
751async def test_a_lost_ai_engine_notifies_the_consumers() -> None:
752    """Announce a vanished AI engine, the selection AI Radio depends on."""
753    states = [_state("tts.only", "Only"), _state("ai_task.only", "Only")]
754
755    async with _start_provider(states) as (provider, hass):
756        await provider.loaded_in_mass()
757        cast("MagicMock", provider.mass).signal_event.reset_mock()
758
759        del hass.compressed_states["ai_task.only"]
760        await _fire_registry_update(provider, hass, "ai_task.only", "remove")
761
762        assert await provider.get_ai_engines() == []
763        assert [engine.id for engine in await provider.get_tts_engines()] == ["tts.only"]
764        assert len(_providers_updated_events(provider)) == 1
765
766
767async def test_engines_are_in_place_before_the_consumers_are_told() -> None:
768    """Expose the rebuilt engine lists before signalling, as consumers read them at once."""
769    states = [_state("tts.only", "Only"), _state("ai_task.only", "Only")]
770
771    async with _start_provider(states) as (provider, hass):
772        await provider.loaded_in_mass()
773        signal_event = cast("MagicMock", provider.mass.signal_event)
774        signal_event.reset_mock()
775        engines_when_told: list[list[str]] = []
776        signal_event.side_effect = lambda *_args, **_kwargs: engines_when_told.append(
777            [engine.id for engine in provider._tts_engines]
778        )
779
780        del hass.compressed_states["tts.only"]
781        await _fire_registry_update(provider, hass, "tts.only", "remove")
782
783        assert engines_when_told == [[]]
784
785
786async def test_unchanged_engines_do_not_notify_the_consumers() -> None:
787    """Stay silent when a refresh rebuilds the very same engine lists."""
788    states = [_state("tts.only", "Only"), _state("ai_task.only", "Only")]
789
790    async with _start_provider(states) as (provider, hass):
791        await provider.loaded_in_mass()
792        cast("MagicMock", provider.mass).signal_event.reset_mock()
793
794        # registry churn that leaves the feature entities as they are, as in a rename of
795        # an entity that the engine name does not depend on
796        await _fire_registry_update(provider, hass, "tts.only", "update")
797
798        assert [engine.id for engine in await provider.get_tts_engines()] == ["tts.only"]
799        assert [engine.id for engine in await provider.get_ai_engines()] == ["ai_task.only"]
800        assert _providers_updated_events(provider) == []
801
802
803async def test_startup_refresh_does_not_notify_the_consumers() -> None:
804    """Leave the announcement of the engines found during startup to the load path."""
805    states = [_state("tts.only", "Only"), _state("ai_task.only", "Only")]
806
807    async with _start_provider(states) as (provider, _):
808        # the refresh that filled the empty lists ran before startup was marked complete
809        assert provider._startup_complete
810        assert [engine.id for engine in await provider.get_tts_engines()] == ["tts.only"]
811        assert _providers_updated_events(provider) == []
812
813
814async def test_refresh_tracks_engines_and_features_in_both_directions() -> None:
815    """Follow the engine lists and their features as feature entities appear and vanish."""
816    async with _start_provider([_state("sensor.example", "Example")]) as (provider, hass):
817        await provider.loaded_in_mass()
818        cast("MagicMock", provider.mass).signal_event.reset_mock()
819        assert ProviderFeature.TTS not in provider.supported_features
820        assert ProviderFeature.AI_QUERY not in provider.supported_features
821
822        hass.compressed_states["tts.new"] = _compressed(_state("tts.new", "New TTS"))
823        hass.compressed_states["ai_task.new"] = _compressed(_state("ai_task.new", "New AI"))
824        await _fire_registry_update(provider, hass, "tts.new", "create")
825
826        assert [engine.id for engine in await provider.get_tts_engines()] == ["tts.new"]
827        assert [engine.id for engine in await provider.get_ai_engines()] == ["ai_task.new"]
828        assert ProviderFeature.TTS in provider.supported_features
829        assert ProviderFeature.AI_QUERY in provider.supported_features
830
831        del hass.compressed_states["tts.new"]
832        del hass.compressed_states["ai_task.new"]
833        await _fire_registry_update(provider, hass, "tts.new", "remove")
834
835        assert await provider.get_tts_engines() == []
836        assert await provider.get_ai_engines() == []
837        assert ProviderFeature.TTS not in provider.supported_features
838        assert ProviderFeature.AI_QUERY not in provider.supported_features
839        assert len(_providers_updated_events(provider)) == 2
840
841
842async def test_registry_update_of_another_domain_is_ignored() -> None:
843    """Ignore registry updates for entities that cannot back a feature."""
844    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
845        await provider.loaded_in_mass()
846
847        hass.fire_event("entity_registry_updated", _registry_event("light.kitchen", "create"))
848
849        assert provider._engine_refresh_task is None
850
851
852async def test_registry_update_of_a_feature_entity_refreshes_without_a_refetch() -> None:
853    """Rebuild the engine lists for a renamed feature entity without refetching the registry."""
854    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
855        await provider.loaded_in_mass()
856        registry = await provider.get_entity_registry()
857        hass.compressed_states["tts.only"] = _compressed(_state("tts.only", "Renamed"))
858
859        with patch("music_assistant.providers.hass.ENGINE_REFRESH_DEBOUNCE", 0):
860            hass.fire_event(
861                "entity_registry_updated", _registry_event("tts.only", changes={"name": "Only"})
862            )
863            assert provider._engine_refresh_task is not None
864            async with asyncio.timeout(1):
865                await provider._engine_refresh_task
866
867        engines = await provider.get_tts_engines()
868        assert [engine.name for engine in engines] == ["Renamed (tts.only)"]
869        assert provider._entity_registry is registry
870
871
872async def test_burst_of_registry_updates_triggers_a_single_refresh() -> None:
873    """Collect a burst of registry updates into one rebuild of the engine lists."""
874    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
875        await provider.loaded_in_mass()
876        registry_fetches = hass.calls.count(REGISTRY_LIST_COMMAND)
877
878        with patch("music_assistant.providers.hass.ENGINE_REFRESH_DEBOUNCE", 0.05):
879            for index in range(3):
880                hass.fire_event(
881                    "entity_registry_updated", _registry_event(f"tts.new_{index}", "create")
882                )
883            assert provider._engine_refresh_task is not None
884            async with asyncio.timeout(1):
885                await provider._engine_refresh_task
886
887        assert hass.calls.count(REGISTRY_LIST_COMMAND) == registry_fetches + 1
888
889
890async def test_registry_update_of_another_domain_invalidates_the_registry() -> None:
891    """Refresh the cached registry for an entity that appeared in any domain."""
892    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
893        registry_fetches = hass.calls.count(REGISTRY_LIST_COMMAND)
894        hass.compressed_states["light.kitchen"] = _compressed(_state("light.kitchen", "Kitchen"))
895
896        hass.fire_event("entity_registry_updated", _registry_event("light.kitchen", "create"))
897
898        # an entity that can not back a feature must not trigger an engine rebuild
899        assert provider._engine_refresh_task is None
900        result = await provider.get_states(domains=("light",))
901        assert [state["entity_id"] for state in result] == ["light.kitchen"]
902        assert hass.calls.count(REGISTRY_LIST_COMMAND) == registry_fetches + 1
903
904
905@pytest.mark.parametrize(
906    "changes",
907    [
908        pytest.param({"name": "Old name"}, id="rename"),
909        pytest.param({"icon": None}, id="icon"),
910        pytest.param({"labels": []}, id="labels"),
911        pytest.param({"hidden_by": None}, id="hidden"),
912        # an integration reload re-registers its entities, which touches these
913        pytest.param({"capabilities": None, "supported_features": 0}, id="reload"),
914    ],
915)
916async def test_registry_update_of_unmirrored_fields_keeps_the_registry(
917    changes: dict[str, Any],
918) -> None:
919    """Keep the mirrored registry for an update that cannot change what it holds."""
920    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
921        registry = await provider.get_entity_registry()
922        registry_fetches = hass.calls.count(REGISTRY_LIST_COMMAND)
923
924        hass.fire_event(
925            "entity_registry_updated", _registry_event("light.kitchen", changes=changes)
926        )
927
928        assert await provider.get_entity_registry() is registry
929        assert hass.calls.count(REGISTRY_LIST_COMMAND) == registry_fetches
930
931
932@pytest.mark.parametrize(
933    "changes",
934    [
935        pytest.param({"entity_id": "light.old"}, id="entity_id"),
936        pytest.param({"platform": "old_platform"}, id="platform"),
937        pytest.param({"device_id": None}, id="device_id"),
938        pytest.param({"area_id": None}, id="area_id"),
939        # the listing omits disabled entities, so this one enters or leaves it
940        pytest.param({"disabled_by": None}, id="disabled_by"),
941        # a device rename reports no changed fields at all
942        pytest.param({}, id="changes_empty"),
943        # a move to another config entry can silently re-enable the entity
944        pytest.param({"config_entry_id": "other"}, id="config_entry_id"),
945    ],
946)
947async def test_registry_update_of_mirrored_fields_invalidates_the_registry(
948    changes: dict[str, Any],
949) -> None:
950    """Refetch the mirrored registry for an update that can change what it holds."""
951    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
952        await provider.get_entity_registry()
953        registry_fetches = hass.calls.count(REGISTRY_LIST_COMMAND)
954
955        hass.fire_event(
956            "entity_registry_updated", _registry_event("light.kitchen", changes=changes)
957        )
958
959        assert provider._entity_registry is None
960        await provider.get_entity_registry()
961        assert hass.calls.count(REGISTRY_LIST_COMMAND) == registry_fetches + 1
962
963
964async def test_unmirrored_change_during_the_fetch_is_still_cached() -> None:
965    """Cache a registry listing that only an irrelevant registry update raced."""
966    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
967        provider._entity_registry = None
968        # hold back the registry response until the update has been delivered
969        registry_response = _hold_back_registry_fetch(hass)
970        lookup = asyncio.ensure_future(provider.get_states(domains=("tts",)))
971        await _wait_for_registry_fetch(hass)
972
973        hass.fire_event(
974            "entity_registry_updated", _registry_event("light.kitchen", changes={"name": "Old"})
975        )
976        registry_response.set_result(None)
977        async with asyncio.timeout(1):
978            await lookup
979
980        assert provider._entity_registry is not None
981
982
983async def test_domains_are_resolved_through_the_display_registry() -> None:
984    """Resolve domains through the compact registry listing only."""
985    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
986        await provider.get_states(domains=("tts",))
987
988        assert REGISTRY_LIST_COMMAND in hass.calls
989        assert "config/entity_registry/list" not in hass.calls
990
991
992async def test_registry_is_fetched_once_per_connection() -> None:
993    """Serve repeated domain lookups from a registry that is fetched only once."""
994    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
995        registry_fetches = hass.calls.count(REGISTRY_LIST_COMMAND)
996
997        await provider.get_states(domains=("tts",))
998        await provider.get_states(domains=("media_player",))
999
1000        assert registry_fetches == 1
1001        assert hass.calls.count(REGISTRY_LIST_COMMAND) == registry_fetches
1002
1003
1004async def test_concurrent_domain_lookups_share_one_registry_fetch() -> None:
1005    """Let concurrent domain lookups share a single registry fetch."""
1006    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
1007        registry_fetches = hass.calls.count(REGISTRY_LIST_COMMAND)
1008        provider._entity_registry = None
1009        # hold back the registry response until both lookups are waiting for it
1010        registry_response = _hold_back_registry_fetch(hass)
1011
1012        lookups = asyncio.gather(
1013            provider.get_states(domains=("tts",)), provider.get_states(domains=("tts",))
1014        )
1015        await _wait_for_registry_fetch(hass)
1016        registry_response.set_result(None)
1017        async with asyncio.timeout(1):
1018            results = await lookups
1019
1020        assert all(len(result) == 1 for result in results)
1021        assert hass.calls.count(REGISTRY_LIST_COMMAND) == registry_fetches + 1
1022
1023
1024async def test_registry_changed_during_the_fetch_is_not_cached() -> None:
1025    """Keep a registry listing out of the cache when a registry update outdated it in flight."""
1026    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
1027        provider._entity_registry = None
1028        # hold back the registry response until the update has been delivered
1029        registry_response = _hold_back_registry_fetch(hass)
1030        lookup = asyncio.ensure_future(provider.get_states(domains=("tts",)))
1031        await _wait_for_registry_fetch(hass)
1032
1033        hass.fire_event("entity_registry_updated", _registry_event("light.kitchen", "create"))
1034        registry_response.set_result(None)
1035        async with asyncio.timeout(1):
1036            await lookup
1037
1038        assert provider._entity_registry is None
1039        registry_fetches = hass.calls.count(REGISTRY_LIST_COMMAND)
1040        await provider.get_states(domains=("tts",))
1041        assert hass.calls.count(REGISTRY_LIST_COMMAND) == registry_fetches + 1
1042
1043
1044async def test_shared_registry_rejects_writes() -> None:
1045    """Reject writes to the registry (and its entries) shared between all callers."""
1046    async with _start_provider([_state("tts.only", "Only")]) as (provider, _):
1047        registry = await provider.get_entity_registry()
1048
1049        with pytest.raises(TypeError):
1050            registry["light.kitchen"] = HassRegistryEntity(  # type: ignore[index]
1051                platform="test", device_id=None, area_id=None
1052            )
1053        with pytest.raises(AttributeError):
1054            registry["tts.only"].platform = "test"  # type: ignore[misc]
1055
1056
1057async def test_registry_reuses_repeated_strings() -> None:
1058    """Hold on to a single string object per distinct platform, device id and area id."""
1059    async with _start_provider([_state("tts.only", "Only")]) as (provider, _):
1060        # decode the listing like a real response, so the repeated platform, device id and
1061        # area id arrive as distinct string objects instead of shared literals
1062        entities = json.loads(
1063            json.dumps(
1064                [
1065                    {
1066                        "ei": f"light.lamp_{index}",
1067                        "pl": "esphome",
1068                        "di": "device",
1069                        "ai": "area",
1070                    }
1071                    for index in range(3)
1072                ]
1073            )
1074        )
1075        with patch.object(
1076            provider.hass, "send_command", AsyncMock(return_value={"entities": entities})
1077        ):
1078            registry = await provider._fetch_entity_registry()
1079
1080        assert len(registry) == 3
1081        assert registry["light.lamp_0"] == HassRegistryEntity(
1082            platform="esphome", device_id="device", area_id="area"
1083        )
1084        assert len({id(entry.platform) for entry in registry.values()}) == 1
1085        assert len({id(entry.device_id) for entry in registry.values()}) == 1
1086        assert len({id(entry.area_id) for entry in registry.values()}) == 1
1087
1088
1089async def test_registry_leaves_out_the_device_and_area_it_is_not_told_about() -> None:
1090    """Report no device or area for the entities whose listing entry omits them."""
1091    async with _start_provider([_state("tts.only", "Only")]) as (provider, _):
1092        entities = [
1093            {"ei": "light.full", "pl": "esphome", "di": "device", "ai": "area"},
1094            {"ei": "light.bare", "pl": "esphome"},
1095            {"ei": "light.device_only", "pl": "esphome", "di": "other_device"},
1096        ]
1097        with patch.object(
1098            provider.hass, "send_command", AsyncMock(return_value={"entities": entities})
1099        ):
1100            registry = await provider._fetch_entity_registry()
1101
1102        assert registry["light.bare"] == HassRegistryEntity(
1103            platform="esphome", device_id=None, area_id=None
1104        )
1105        assert registry["light.device_only"] == HassRegistryEntity(
1106            platform="esphome", device_id="other_device", area_id=None
1107        )
1108
1109
1110async def test_device_registry_is_reused_within_the_cache_window() -> None:
1111    """Serve a later device lookup from the cached listing."""
1112    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
1113        hass.devices = [{"id": "dev1"}]
1114
1115        assert await provider.get_device_registry() == {"dev1": {"id": "dev1"}}
1116        await _wait_for_stored(provider)
1117        await provider.get_device_registry()
1118
1119        assert hass.calls.count(DEVICE_REGISTRY_LIST) == 1
1120
1121
1122async def test_device_registry_is_refetched_once_the_cache_window_passed() -> None:
1123    """Fetch the device listing again once the cached entry is no longer fresh."""
1124    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
1125        hass.devices = [{"id": "dev1"}]
1126        await provider.get_device_registry()
1127        hass.devices = [{"id": "dev1"}, {"id": "dev2"}]
1128
1129        cast("_Cache", provider.mass.cache).expire_all()
1130
1131        assert list(await provider.get_device_registry()) == ["dev1", "dev2"]
1132        assert hass.calls.count(DEVICE_REGISTRY_LIST) == 2
1133
1134
1135async def test_device_entries_survive_the_cache_round_trip() -> None:
1136    """Return a cached device entry with all of its fields intact."""
1137    device = {
1138        "id": "dev1",
1139        "name": "Kitchen Speaker",
1140        "name_by_user": None,
1141        "connections": [["mac", "aa:bb:cc:dd:ee:ff"]],
1142        "manufacturer": "ESPHome",
1143    }
1144    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
1145        hass.devices = [device]
1146
1147        fetched = await provider.get_device_registry()
1148        await _wait_for_stored(provider)
1149        cached = await provider.get_device_registry()
1150
1151        assert hass.calls.count(DEVICE_REGISTRY_LIST) == 1
1152        # the cached read is reconstructed from the return annotation, so it must not
1153        # drop or reshape any of the device fields
1154        assert cached == fetched == {"dev1": device}
1155
1156
1157async def test_concurrent_device_lookups_do_not_fetch_per_caller() -> None:
1158    """Keep a burst of concurrent device lookups off a fetch-per-caller path."""
1159    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
1160        hass.devices = [{"id": "dev1"}]
1161
1162        async with asyncio.timeout(1):
1163            results = await asyncio.gather(*(provider.get_device_registry() for _ in range(20)))
1164
1165        assert all(list(result) == ["dev1"] for result in results)
1166        # the lock keeps the fetch count flat instead of growing with the burst; it does
1167        # not reach one, because use_cache stores in the background and the caller right
1168        # behind the first one still finds the cache cold
1169        assert hass.calls.count(DEVICE_REGISTRY_LIST) <= 2
1170
1171
1172async def test_disabled_entity_is_never_requested() -> None:
1173    """Leave an entity that Home Assistant disabled out of the state fetch."""
1174    states = [_state("media_player.kitchen", "Kitchen"), _state("media_player.spare", "Spare")]
1175
1176    async with _start_provider(states) as (provider, hass):
1177        # Home Assistant omits disabled entities from the registry, their state remains
1178        hass.disabled_entities.add("media_player.spare")
1179        hass.fire_event(
1180            "entity_registry_updated",
1181            _registry_event("media_player.spare", changes={"disabled_by": None}),
1182        )
1183        hass.subscriptions.clear()
1184
1185        result = await provider.get_states(domains=("media_player",))
1186
1187        assert [state["entity_id"] for state in result] == ["media_player.kitchen"]
1188        assert hass.subscriptions == [["media_player.kitchen"]]
1189
1190
1191async def test_registry_entries_are_fetched_for_the_given_entities_only() -> None:
1192    """Fetch the full registry entries of exactly the requested entities."""
1193    async with _start_provider([_state("media_player.kitchen", "Kitchen")]) as (provider, hass):
1194        entries = await provider.get_entity_registry_entries(
1195            ["media_player.kitchen", "media_player.unknown"]
1196        )
1197
1198        # an entity Home Assistant does not know is absent from the result
1199        assert list(entries) == ["media_player.kitchen"]
1200        assert entries["media_player.kitchen"]["config_entry_id"] == "config_entry_id"
1201        hass.send_command.assert_awaited_with(
1202            REGISTRY_ENTRIES_COMMAND,
1203            entity_ids=["media_player.kitchen", "media_player.unknown"],
1204        )
1205
1206
1207async def test_registry_entries_of_nothing_skips_the_round_trip() -> None:
1208    """Do not contact Home Assistant when no entity is requested."""
1209    async with _start_provider([_state("media_player.kitchen", "Kitchen")]) as (provider, hass):
1210        calls_before = list(hass.calls)
1211
1212        assert await provider.get_entity_registry_entries([]) == {}
1213        assert hass.calls == calls_before
1214
1215
1216async def test_entity_registry_subscription_is_replaced() -> None:
1217    """Replace the entity registry subscription instead of stacking a second one."""
1218    async with _start_provider([_state("tts.only", "Only")]) as (provider, hass):
1219        assert hass.active_event_subscriptions == 1
1220        assert hass.event_subscriptions[0][0] == "entity_registry_updated"
1221
1222        await provider._subscribe_entity_registry()
1223
1224        assert hass.active_event_subscriptions == 1
1225        assert hass.calls[-2:] == ["unsubscribe_events", "subscribe_events"]
1226
1227        await provider.unload()
1228
1229        assert hass.active_event_subscriptions == 0
1230
1231
1232async def test_config_entries_do_not_read_home_assistant() -> None:
1233    """Build the config entries without asking Home Assistant for anything."""
1234    async with _start_provider([_state("switch.example", "Example")]) as (provider, _):
1235        provider.available = True
1236        with patch.object(provider, "get_states", AsyncMock()) as get_states:
1237            entries = await provider.get_config_entries()
1238
1239    get_states.assert_not_awaited()
1240    assert CONF_POWER_CONTROLS in {entry.key for entry in entries}
1241
1242
1243async def test_config_entries_offer_the_control_lists_without_options() -> None:
1244    """Offer the control lists as plain entity id lists, filled in by the entity picker."""
1245    async with _start_provider([_state("switch.example", "Example")]) as (provider, _):
1246        provider.available = True
1247        entries = await provider.get_config_entries()
1248
1249    control_entries = [
1250        entry
1251        for entry in entries
1252        if entry.key in (CONF_POWER_CONTROLS, CONF_VOLUME_CONTROLS, CONF_MUTE_CONTROLS)
1253    ]
1254    assert len(control_entries) == 3
1255    for entry in control_entries:
1256        assert entry.options == []
1257        assert entry.multi_value is True
1258
1259
1260async def test_domain_states_use_a_single_subscription() -> None:
1261    """Fetch every entity of a domain in one websocket round-trip."""
1262    states = [_state(f"media_player.player_{index}", f"Player {index}") for index in range(50)]
1263
1264    async with _start_provider(states) as (provider, hass):
1265        result = await provider.get_states(domains=("media_player",))
1266
1267        assert len(result) == len(states)
1268        assert len(hass.subscriptions) == 1
1269        assert hass.subscriptions[0] == sorted(state["entity_id"] for state in states)
1270
1271
1272async def test_large_requests_are_split_into_batches() -> None:
1273    """Split a request that exceeds the batch size into bounded batches."""
1274    entity_ids = [
1275        f"media_player.player_{index:04d}" for index in range(STATE_FETCH_BATCH_SIZE * 2 + 1)
1276    ]
1277    states = [_state(entity_id, entity_id) for entity_id in entity_ids]
1278
1279    async with _start_provider(states) as (provider, hass):
1280        result = await provider.get_states(domains=("media_player",))
1281
1282        assert len(result) == len(entity_ids)
1283        assert len(hass.subscriptions) == ceil(len(entity_ids) / STATE_FETCH_BATCH_SIZE)
1284        assert all(len(batch) <= STATE_FETCH_BATCH_SIZE for batch in hass.subscriptions)
1285        requested = [entity_id for batch in hass.subscriptions for entity_id in batch]
1286        assert sorted(requested) == sorted(entity_ids)
1287        assert len(requested) == len(set(requested))
1288
1289
1290async def test_every_batch_is_unsubscribed() -> None:
1291    """Release the subscription of every batch once its states have been received."""
1292    entity_ids = [f"media_player.player_{index}" for index in range(5)]
1293    states = [_state(entity_id, entity_id) for entity_id in entity_ids]
1294
1295    async with _start_provider(states) as (provider, hass):
1296        with patch("music_assistant.providers.hass.STATE_FETCH_BATCH_SIZE", 2):
1297            await provider.get_states(domains=("media_player",))
1298
1299        assert len(hass.subscriptions) == 3
1300        assert hass.unsubscribed == hass.subscriptions
1301        assert hass.active_subscriptions == 0
1302
1303
1304async def test_timed_out_fetch_is_unsubscribed() -> None:
1305    """Release the subscription when Home Assistant never sends the states."""
1306    async with _start_provider([_state("media_player.kitchen", "Kitchen")]) as (provider, hass):
1307        hass.deliver_initial_states = False
1308
1309        with (
1310            patch("music_assistant.providers.hass.STATE_FETCH_TIMEOUT", 0.05),
1311            pytest.raises(TimeoutError),
1312        ):
1313            await provider.get_states(entity_ids=["media_player.kitchen"])
1314
1315        assert hass.unsubscribed == [["media_player.kitchen"]]
1316        assert hass.active_subscriptions == 0
1317
1318
1319async def test_cancelled_fetch_is_unsubscribed() -> None:
1320    """Release the subscription when the state fetch is cancelled."""
1321    async with _start_provider([_state("media_player.kitchen", "Kitchen")]) as (provider, hass):
1322        hass.deliver_initial_states = False
1323        fetch_task = asyncio.create_task(provider.get_states(entity_ids=["media_player.kitchen"]))
1324        async with asyncio.timeout(1):
1325            await hass.subscribed.wait()
1326        fetch_task.cancel()
1327
1328        with pytest.raises(asyncio.CancelledError):
1329            await fetch_task
1330
1331        assert hass.unsubscribed == [["media_player.kitchen"]]
1332        assert hass.active_subscriptions == 0
1333
1334
1335async def test_compressed_states_are_expanded() -> None:
1336    """Expand the compressed states of the initial message into full states."""
1337    async with _start_provider([]) as (provider, hass):
1338        hass.compressed_states = {
1339            "media_player.full": {
1340                "s": "playing",
1341                "a": {"friendly_name": "Full"},
1342                "lc": LAST_CHANGED,
1343                "lu": 1683838800.736819,
1344                "c": {"id": CONTEXT_ID, "parent_id": None, "user_id": "user"},
1345            },
1346            "media_player.unchanged": {"s": "idle", "lc": LAST_CHANGED, "c": CONTEXT_ID},
1347            "media_player.minimal": {},
1348        }
1349
1350        result = {
1351            state["entity_id"]: state
1352            for state in await provider.get_states(entity_ids=list(hass.compressed_states))
1353        }
1354
1355        assert result["media_player.full"]["state"] == "playing"
1356        assert result["media_player.full"]["attributes"] == {"friendly_name": "Full"}
1357        assert result["media_player.full"]["last_changed"] == LAST_CHANGED_ISO
1358        assert result["media_player.full"]["last_updated"] == "2023-05-11T21:00:00.736819+00:00"
1359        assert result["media_player.full"]["context"] == {
1360            "id": CONTEXT_ID,
1361            "parent_id": None,
1362            "user_id": "user",
1363        }
1364        # last_updated is omitted by HA when it is identical to last_changed
1365        assert result["media_player.unchanged"]["last_changed"] == LAST_CHANGED_ISO
1366        assert result["media_player.unchanged"]["last_updated"] == LAST_CHANGED_ISO
1367        assert result["media_player.unchanged"]["context"] == {
1368            "id": CONTEXT_ID,
1369            "parent_id": None,
1370            "user_id": None,
1371        }
1372        assert result["media_player.minimal"] == {
1373            "entity_id": "media_player.minimal",
1374            "state": "",
1375            "attributes": {},
1376            "last_changed": "",
1377            "last_updated": "",
1378            "context": {"id": "", "parent_id": None, "user_id": None},
1379        }
1380
1381
1382async def test_entity_without_state_is_absent() -> None:
1383    """Omit entities that Home Assistant has no state for."""
1384    async with _start_provider([_state("media_player.kitchen", "Kitchen")]) as (provider, hass):
1385        result = await provider.get_states(
1386            entity_ids=["media_player.kitchen", "media_player.removed"]
1387        )
1388
1389        assert [state["entity_id"] for state in result] == ["media_player.kitchen"]
1390        assert hass.subscriptions == [["media_player.kitchen", "media_player.removed"]]
1391
1392
1393async def test_state_change_does_not_complete_the_fetch() -> None:
1394    """Ignore a state change that arrives before the initial state message."""
1395    async with _start_provider([_state("media_player.kitchen", "Kitchen")]) as (provider, hass):
1396        hass.leading_events = [{"c": {"media_player.kitchen": {"+": {"s": "playing"}}}}]
1397
1398        result = await provider.get_states(entity_ids=["media_player.kitchen"])
1399
1400        assert [state["entity_id"] for state in result] == ["media_player.kitchen"]
1401        assert result[0]["state"] == "idle"
1402
1403
1404async def test_player_control_subscription_is_replaced() -> None:
1405    """Replace the player control subscription instead of stacking a second one."""
1406    states = [_state("media_player.kitchen", "Kitchen")]
1407    async with _start_provider(states, **{CONF_POWER_CONTROLS: ["media_player.kitchen"]}) as (
1408        provider,
1409        hass,
1410    ):
1411        await provider.loaded_in_mass()
1412        assert hass.active_subscriptions == 1
1413
1414        # only a changed selection results in a new subscription
1415        provider.config = _config(
1416            **{
1417                CONF_POWER_CONTROLS: ["media_player.kitchen"],
1418                CONF_MUTE_CONTROLS: ["switch.kitchen_amp"],
1419            }
1420        )
1421        await provider._register_player_controls()
1422
1423        assert len(hass.subscriptions) == 4
1424        assert hass.active_subscriptions == 1
1425        # the previous control subscription is only released once its replacement is live
1426        assert hass.calls[-2:] == ["subscribe_entities", "unsubscribe_entities"]
1427
1428        await provider.unload()
1429
1430        assert hass.active_subscriptions == 0
1431
1432
1433async def test_failed_control_subscription_keeps_the_previous_one() -> None:
1434    """A failed subscription attempt leaves the controls watched by the earlier one."""
1435    states = [_state("media_player.kitchen", "Kitchen"), _state("switch.amp", "Amp")]
1436    async with _start_provider(states, **{CONF_POWER_CONTROLS: ["media_player.kitchen"]}) as (
1437        provider,
1438        hass,
1439    ):
1440        await provider.loaded_in_mass()
1441        unsubscribe_before = provider._unsubscribe_controls
1442        assert hass.active_subscriptions == 1
1443
1444        async def _failing_subscribe(*_args: Any, **_kwargs: Any) -> Callable[[], None]:
1445            raise BaseHassClientError("connection lost")
1446
1447        hass.subscribe_entities = _failing_subscribe  # type: ignore[method-assign]
1448        provider.config = _config(
1449            **{
1450                CONF_POWER_CONTROLS: ["media_player.kitchen"],
1451                CONF_MUTE_CONTROLS: ["switch.amp"],
1452            }
1453        )
1454
1455        with pytest.raises(BaseHassClientError):
1456            await provider._register_player_controls()
1457
1458        assert provider._unsubscribe_controls is unsubscribe_before
1459        assert hass.active_subscriptions == 1
1460
1461
1462async def test_unchanged_control_selection_skips_home_assistant() -> None:
1463    """Reconciling an unchanged selection does not talk to Home Assistant at all."""
1464    states = [_state("media_player.kitchen", "Kitchen")]
1465    async with _start_provider(states, **{CONF_POWER_CONTROLS: ["media_player.kitchen"]}) as (
1466        provider,
1467        hass,
1468    ):
1469        await provider.loaded_in_mass()
1470        calls_after_load = list(hass.calls)
1471
1472        await provider._register_player_controls()
1473
1474        assert hass.calls == calls_after_load
1475
1476
1477async def test_control_list_change_reconciles_without_reload() -> None:
1478    """A change confined to the control lists is applied without reloading the provider."""
1479    states = [_state("media_player.kitchen", "Kitchen"), _state("switch.amp", "Amp")]
1480    async with _start_provider(states, **{CONF_POWER_CONTROLS: ["media_player.kitchen"]}) as (
1481        provider,
1482        _hass,
1483    ):
1484        mass = cast("MagicMock", provider.mass)
1485        await provider.loaded_in_mass()
1486        assert set(provider._player_controls or {}) == {"media_player.kitchen"}
1487        mass.call_later.reset_mock()
1488
1489        await provider.update_config(
1490            _config(**{CONF_POWER_CONTROLS: ["switch.amp"]}),
1491            {f"values/{CONF_POWER_CONTROLS}"},
1492        )
1493
1494        mass.call_later.assert_not_called()
1495        assert set(provider._player_controls or {}) == {"switch.amp"}
1496        mass.players.remove_player_control.assert_called_once_with("media_player.kitchen")
1497        registered = mass.players.register_or_update_player_control.call_args[0][0]
1498        assert registered.id == "switch.amp"
1499        assert registered.supports_power is True
1500
1501
1502async def test_control_role_change_is_applied() -> None:
1503    """Moving an already selected entity to another role re-registers it in that role."""
1504    states = [_state("media_player.kitchen", "Kitchen")]
1505    async with _start_provider(states, **{CONF_POWER_CONTROLS: ["media_player.kitchen"]}) as (
1506        provider,
1507        _hass,
1508    ):
1509        await provider.loaded_in_mass()
1510
1511        await provider.update_config(
1512            _config(**{CONF_VOLUME_CONTROLS: ["media_player.kitchen"]}),
1513            {f"values/{CONF_POWER_CONTROLS}", f"values/{CONF_VOLUME_CONTROLS}"},
1514        )
1515
1516        control = (provider._player_controls or {})["media_player.kitchen"]
1517        assert control.supports_volume is True
1518        assert control.supports_power is False
1519
1520
1521async def test_control_selection_drops_a_value_that_is_no_entity_id() -> None:
1522    """A stored selection that is no entity ID must not take the other controls down."""
1523    states = [_state("media_player.kitchen", "Kitchen")]
1524    async with _start_provider(
1525        states, **{CONF_POWER_CONTROLS: ["power_controls", "media_player.kitchen"]}
1526    ) as (provider, hass):
1527        await provider.loaded_in_mass()
1528
1529        assert set(provider._player_controls or {}) == {"media_player.kitchen"}
1530        # Home Assistant refuses the whole subscription when it carries the stray value
1531        assert hass.subscriptions[-1] == ["media_player.kitchen"]
1532
1533
1534@pytest.mark.parametrize(
1535    "changed_keys",
1536    [
1537        {f"values/{CONF_URL}"},
1538        # a control list changing alongside another value still needs the reload
1539        {f"values/{CONF_POWER_CONTROLS}", f"values/{CONF_URL}"},
1540    ],
1541)
1542async def test_other_config_change_reloads_provider(changed_keys: set[str]) -> None:
1543    """Any change beyond the control lists reloads the provider."""
1544    async with _start_provider([]) as (provider, _hass):
1545        mass = cast("MagicMock", provider.mass)
1546        await provider.loaded_in_mass()
1547        mass.call_later.reset_mock()
1548
1549        await provider.update_config(_config(**{CONF_POWER_CONTROLS: ["switch.amp"]}), changed_keys)
1550
1551        mass.call_later.assert_called_once()
1552        assert mass.call_later.call_args[0][1] is mass.load_provider_config
1553