/
/
/
1"""Tests for the Bose SoundTouch API client parsing helpers."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, Mock, patch
6
7from defusedxml import ElementTree as DefusedET
8
9from music_assistant.providers.bose_soundtouch.client import SoundtouchDevice
10from music_assistant.providers.bose_soundtouch.client.client import (
11 create_notification_xml,
12 create_zone_xml,
13)
14from music_assistant.providers.bose_soundtouch.client.schema.enums import PlayStatus, SourceStatus
15from music_assistant.providers.bose_soundtouch.client.schema.models import Zone, ZoneMember
16from music_assistant.providers.bose_soundtouch.helpers import extract_preset_id
17
18INFO_XML = """
19<info deviceID="ABC123">
20 <name>Living Room</name>
21 <type>SoundTouch 20</type>
22 <networkInfo type="SCM">
23 <macAddress>001122334455</macAddress>
24 <ipAddress>192.168.1.50</ipAddress>
25 </networkInfo>
26 <components>
27 <component>
28 <componentCategory>SCM</componentCategory>
29 <softwareVersion>27.0.6.46330 epdbuild</softwareVersion>
30 </component>
31 </components>
32</info>
33"""
34
35NOW_PLAYING_XML = """
36<nowPlaying deviceID="ABC123" source="SPOTIFY" sourceAccount="user1">
37 <ContentItem source="SPOTIFY"><itemName>My Playlist</itemName></ContentItem>
38 <track>Song Title</track>
39 <artist>The Artist</artist>
40 <album>The Album</album>
41 <art artImageStatus="IMAGE_PRESENT">http://192.168.1.50/art.jpg</art>
42 <time total="240">42</time>
43 <playStatus>PLAY_STATE</playStatus>
44</nowPlaying>
45"""
46
47
48def _get_client() -> SoundtouchDevice:
49 session_config = Mock()
50 session_config.logger = None
51 session_config.ip = "10.0.0.9"
52 return SoundtouchDevice(session_configuration=session_config)
53
54
55async def test_parse_info() -> None:
56 """Device info is parsed including identifiers and firmware version."""
57 client = _get_client()
58 with patch.object(client, "_get", new_callable=AsyncMock) as mock_get:
59 mock_get.return_value = DefusedET.fromstring(INFO_XML)
60
61 info = await client.get_info()
62
63 assert info.device_id == "ABC123"
64 assert info.name == "Living Room"
65 assert info.model == "SoundTouch 20"
66 assert isinstance(info.mac_addresses, set)
67 assert "001122334455" in info.mac_addresses
68 assert isinstance(info.ip_addresses, set)
69 assert "192.168.1.50" in info.ip_addresses
70 assert info.software_version == "27.0.6.46330"
71
72
73async def test_parse_info_falls_back_to_connection_ip() -> None:
74 """When the device reports no network IP, we still have the ip in our session config."""
75 client = _get_client()
76 with patch.object(client, "_get", new_callable=AsyncMock) as mock_get:
77 mock_get.return_value = DefusedET.fromstring('<info deviceID="X"><name>N</name></info>')
78 info = await client.get_info()
79
80 assert isinstance(info.ip_addresses, set)
81 assert "10.0.0.9" in info.ip_addresses
82
83
84async def test_parse_now_playing() -> None:
85 """A playing now_playing snapshot maps to the expected fields."""
86 client = _get_client()
87 with patch.object(client, "_get", new_callable=AsyncMock) as mock_get:
88 mock_get.return_value = DefusedET.fromstring(NOW_PLAYING_XML)
89 now_playing = await client.get_now_playing()
90
91 assert now_playing.content_item is not None
92 assert now_playing.content_item.source == "SPOTIFY"
93 # assert now_playing.content_item.source_account == "user1"
94 assert now_playing.track == "Song Title"
95 assert now_playing.artist == "The Artist"
96 assert now_playing.album == "The Album"
97 assert now_playing.art is not None
98 assert now_playing.art.url == "http://192.168.1.50/art.jpg"
99 assert now_playing.time_information is not None
100 assert now_playing.time_information.total == 240
101 assert now_playing.time_information.position == 42
102 assert now_playing.play_status in [PlayStatus.PLAY_STATE, PlayStatus.BUFFERING_STATE]
103
104
105async def test_parse_now_playing_standby() -> None:
106 """A standby snapshot reports the STANDBY source."""
107 client = _get_client()
108 with patch.object(client, "_get", new_callable=AsyncMock) as mock_get:
109 mock_get.return_value = DefusedET.fromstring(
110 '<nowPlaying deviceID="A" source="STANDBY"></nowPlaying>'
111 )
112 now_playing = await client.get_now_playing()
113 assert now_playing.content_item is None
114 assert now_playing.source == "STANDBY"
115 assert now_playing.track is None
116
117
118async def test_parse_volume() -> None:
119 """Volume level and mute state are parsed."""
120 client = _get_client()
121 with patch.object(client, "_get", new_callable=AsyncMock) as mock_get:
122 mock_get.return_value = DefusedET.fromstring(
123 "<volume><targetvolume>30</targetvolume>"
124 "<actualvolume>28</actualvolume><muteenabled>true</muteenabled></volume>"
125 )
126 volume = await client.get_volume()
127
128 assert volume.actual_volume == 28
129 assert volume.mute_enabled is True
130
131
132async def test_parse_sources() -> None:
133 """Sources are parsed with their ready state."""
134 client = _get_client()
135 with patch.object(client, "_get", new_callable=AsyncMock) as mock_get:
136 mock_get.return_value = DefusedET.fromstring(
137 '<sources deviceID="A">'
138 '<sourceItem source="AUX" sourceAccount="AUX" status="READY">AUX IN</sourceItem>'
139 '<sourceItem source="BLUETOOTH" status="UNAVAILABLE">Bluetooth</sourceItem>'
140 "</sources>"
141 )
142 sources = await client.get_sources()
143
144 assert len(sources.sources) == 2
145 assert sources.sources[0].source == "AUX"
146 assert sources.sources[0].source_account == "AUX"
147 assert sources.sources[0].source_name == "AUX IN"
148 assert sources.sources[0].status == SourceStatus.READY
149 assert sources.sources[1].status != SourceStatus.READY
150
151
152async def test_parse_zone_master() -> None:
153 """A zone master reports its members."""
154 client = _get_client()
155 with patch.object(client, "_get", new_callable=AsyncMock) as mock_get:
156 mock_get.return_value = DefusedET.fromstring(
157 '<zone master="ABC"><member ipaddress="192.168.1.51">DEF</member></zone>'
158 )
159 zone = await client.get_zone()
160
161 assert zone.leader is not None
162 assert zone.leader.mac == "ABC"
163 assert zone.members
164 assert ZoneMember(ip="192.168.1.51", mac="DEF") in zone.members
165
166
167async def test_parse_zone_empty() -> None:
168 """An empty zone response yields no master and no members."""
169 client = _get_client()
170 with patch.object(client, "_get", new_callable=AsyncMock) as mock_get:
171 mock_get.return_value = DefusedET.fromstring("<zone />")
172
173 zone = await client.get_zone()
174
175 if zone.leader is not None:
176 assert zone.leader.ip is None
177 assert zone.leader.mac is None
178 assert len(zone.members) == 0
179
180
181async def test_build_zone_xml() -> None:
182 """Zone request bodies are built with master and member entries."""
183 leader = ZoneMember(ip="1.2.3.4", mac="MASTER")
184 member = ZoneMember(ip="1.2.3.5", mac="SLAVE")
185 zone = Zone(leader=leader, members=[leader, member])
186
187 assert create_zone_xml(zone) == (
188 '<zone master="MASTER">'
189 '<member ipaddress="1.2.3.4">MASTER</member>'
190 '<member ipaddress="1.2.3.5">SLAVE</member>'
191 "</zone>"
192 )
193
194
195async def test_build_notification_xml() -> None:
196 """The notification body includes the app key, url and volume, escaping the url."""
197 xml = create_notification_xml("KEY", "http://host/stream?a=1&b=2", volume=25)
198 assert "<app_key>KEY</app_key>" in xml
199 assert "<url>http://host/stream?a=1&b=2</url>" in xml
200 assert "<volume>25</volume>" in xml
201
202
203def test_build_notification_xml_without_volume() -> None:
204 """The notification body omits the volume element when no volume is given."""
205 xml = create_notification_xml("KEY", "http://host/stream")
206 assert "<volume>" not in xml
207
208
209def test_extract_preset_id() -> None:
210 """A preset button press is detected from a websocket message."""
211 assert extract_preset_id('<updates deviceID="A"><preset id="3"/></updates>') == 3
212
213
214def test_extract_preset_id_no_preset() -> None:
215 """A message without a preset element returns None."""
216 assert extract_preset_id('<updates deviceID="A"><nowPlayingUpdated/></updates>') is None
217
218
219def test_extract_preset_id_invalid_xml() -> None:
220 """A non-XML message returns None instead of raising."""
221 assert extract_preset_id("not xml at all") is None
222
223
224def test_extract_preset_id_non_numeric() -> None:
225 """A non-numeric preset id returns None instead of raising."""
226 assert extract_preset_id('<preset id="abc"/>') is None
227