/
/
/
1"""Tests for blocking the import of HA players that are natively supported in MA."""
2
3from __future__ import annotations
4
5import logging
6from types import SimpleNamespace
7from typing import TYPE_CHECKING, Any, cast
8from unittest.mock import AsyncMock
9
10from music_assistant_models.enums import IdentifierType
11
12from music_assistant.providers.hass_players.constants import (
13 CONF_PLAYERS,
14 DISABLED_REASON_NATIVE_DUPLICATE,
15 DISABLED_REASON_NATIVE_INTEGRATION,
16)
17from music_assistant.providers.hass_players.helpers import native_player_macs, normalized_mac
18from music_assistant.providers.hass_players.provider import HomeAssistantPlayerProvider
19
20if TYPE_CHECKING:
21 from music_assistant_models.config_entries import ConfigValueOption
22
23 from music_assistant.mass import MusicAssistant
24
25PLAY_MEDIA = 512
26MAC = "aa:bb:cc:dd:ee:ff"
27
28
29def _state(entity_id: str) -> dict[str, Any]:
30 return {
31 "entity_id": entity_id,
32 "state": "idle",
33 "attributes": {"friendly_name": entity_id, "supported_features": PLAY_MEDIA},
34 }
35
36
37def _entity(entity_id: str, platform: str, device_id: str | None = None) -> dict[str, Any]:
38 return {"entity_id": entity_id, "platform": platform, "device_id": device_id}
39
40
41def _native_player(mac: str | None = MAC, provider_domain: str = "sendspin") -> SimpleNamespace:
42 identifiers = {IdentifierType.MAC_ADDRESS: mac} if mac else {}
43 return SimpleNamespace(
44 provider=SimpleNamespace(domain=provider_domain),
45 device_info=SimpleNamespace(identifiers=identifiers),
46 )
47
48
49def _mass(
50 entities: list[dict[str, Any]],
51 devices: list[dict[str, Any]] | None = None,
52 native_players: list[SimpleNamespace] | None = None,
53) -> MusicAssistant:
54 entities_by_id = {entity["entity_id"]: entity for entity in entities}
55
56 async def _registry_entries(entity_ids: list[str]) -> dict[str, dict[str, Any]]:
57 return {entity_id: entities_by_id[entity_id] for entity_id in entity_ids}
58
59 hass_prov = SimpleNamespace(
60 hass=SimpleNamespace(connected=True),
61 get_device_registry=AsyncMock(
62 return_value={device["id"]: device for device in devices or []}
63 ),
64 get_entity_registry=AsyncMock(return_value=entities_by_id),
65 get_entity_registry_entries=AsyncMock(side_effect=_registry_entries),
66 get_states=AsyncMock(return_value=[_state(entity["entity_id"]) for entity in entities]),
67 logger=logging.getLogger("test.hass"),
68 )
69 return cast(
70 "MusicAssistant",
71 SimpleNamespace(
72 get_provider=lambda _domain: hass_prov,
73 players=native_players or [],
74 ),
75 )
76
77
78async def _picker_options(
79 mass: MusicAssistant, *, selected_players: list[str] | None = None
80) -> list[ConfigValueOption]:
81 prov = HomeAssistantPlayerProvider.__new__(HomeAssistantPlayerProvider)
82 prov.mass = mass
83
84 # entities already stored under CONF_PLAYERS must stay selectable on a config edit
85 def _stored_config_value(key: str, default: Any = None) -> Any:
86 return selected_players if key == CONF_PLAYERS else default
87
88 prov.get_config_value = _stored_config_value # type: ignore[method-assign, assignment]
89 entries = await prov.get_config_entries()
90 return entries[0].options
91
92
93async def test_native_integration_entity_is_disabled() -> None:
94 """An entity of an integration with a native MA provider can not be newly imported."""
95 mass = _mass([_entity("media_player.living_room_tv", "cast")])
96 (option,) = await _picker_options(mass)
97 assert option.disabled is True
98 assert option.disabled_reason == DISABLED_REASON_NATIVE_INTEGRATION
99
100
101async def test_native_duplicate_device_is_disabled() -> None:
102 """An entity whose device is already a native MA player can not be newly imported."""
103 mass = _mass(
104 [_entity("media_player.kitchen_speaker", "esphome", device_id="dev1")],
105 devices=[{"id": "dev1", "connections": [["mac", MAC.upper()]]}],
106 native_players=[_native_player()],
107 )
108 (option,) = await _picker_options(mass)
109 assert option.disabled is True
110 assert option.disabled_reason == DISABLED_REASON_NATIVE_DUPLICATE
111
112
113async def test_esphome_entity_without_native_player_is_selectable() -> None:
114 """An ESPHome device without a native MA player (no Sendspin) can still be imported."""
115 mass = _mass(
116 [_entity("media_player.diy_speaker", "esphome", device_id="dev1")],
117 devices=[{"id": "dev1", "connections": [["mac", "11:22:33:44:55:66"]]}],
118 native_players=[_native_player()],
119 )
120 (option,) = await _picker_options(mass)
121 assert option.disabled is False
122 assert option.disabled_reason is None
123
124
125async def test_already_imported_entity_stays_selectable() -> None:
126 """An already imported entity stays selectable so a config edit never drops it."""
127 mass = _mass([_entity("media_player.living_room_tv", "cast")])
128 (option,) = await _picker_options(mass, selected_players=["media_player.living_room_tv"])
129 assert option.disabled is False
130 assert option.disabled_reason is None
131
132
133async def test_regular_entity_is_selectable() -> None:
134 """An entity without native MA support is offered as before."""
135 mass = _mass([_entity("media_player.kodi_bedroom", "kodi")])
136 (option,) = await _picker_options(mass)
137 assert option.disabled is False
138 assert option.disabled_reason is None
139
140
141def test_native_player_macs_excludes_own_players() -> None:
142 """MACs of players imported through this provider itself are not counted."""
143 mass = cast(
144 "MusicAssistant",
145 SimpleNamespace(
146 players=[
147 _native_player(),
148 _native_player(mac="11:22:33:44:55:66", provider_domain="hass_players"),
149 _native_player(mac=None),
150 ]
151 ),
152 )
153 assert native_player_macs(mass) == {normalized_mac(MAC)}
154
155
156def test_normalized_mac() -> None:
157 """MAC addresses normalize regardless of separators and casing."""
158 assert normalized_mac("AA:BB:CC:DD:EE:FF") == "aabbccddeeff"
159 assert normalized_mac("aa-bb-cc-dd-ee-ff") == "aabbccddeeff"
160