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