music-assistant-server

1"""Unit tests for the Sonos S1 player."""
2
3from __future__ import annotations
4
5import asyncio
6import threading
7from functools import partial
8from typing import TYPE_CHECKING, cast
9from unittest.mock import AsyncMock, MagicMock, patch
10
11import pytest
12from music_assistant_models.enums import IdentifierType, MediaType, PlaybackState
13from music_assistant_models.errors import PlayerCommandFailed
14from soco.exceptions import SoCoException
15
16from music_assistant.mass import MusicAssistant
17from music_assistant.models.player import PlayerMedia
18from music_assistant.providers.sonos_s1 import player as player_module
19from music_assistant.providers.sonos_s1.constants import (
20    POLL_INTERVAL,
21    SUBSCRIPTION_SERVICES,
22    TRANSITION_POLL_INTERVAL,
23)
24from music_assistant.providers.sonos_s1.player import SonosPlayer
25
26if TYPE_CHECKING:
27    from collections.abc import AsyncGenerator
28
29STREAM_URL = "http://192.168.1.2:8097/single/sessionabc/queue1/item1/player1.flac"
30
31
32def _make_soco(uid: str = "RINCON_000E58AAAAAA01400", name: str = "Test Sonos") -> MagicMock:
33    """Create a mocked soco device."""
34    soco = MagicMock()
35    soco.uid = uid
36    soco.household_id = "Sonos_household"
37    soco.player_name = name
38    soco._player_name = name
39    soco.ip_address = "127.0.0.1"
40    soco.speaker_info = {"model_name": "Sonos Play:1"}
41    return soco
42
43
44@pytest.fixture
45def sonos_player() -> SonosPlayer:
46    """Create a SonosPlayer with a mocked soco device and provider."""
47    provider = MagicMock()
48    provider.mass.streams.resolve_stream_url = AsyncMock(return_value=STREAM_URL)
49    return SonosPlayer(provider=provider, soco=_make_soco())
50
51
52@pytest.fixture
53async def timer_mass() -> AsyncGenerator[MusicAssistant]:
54    """Create a bare MusicAssistant exposing the real call_later/cancel_timer machinery."""
55    mass = object.__new__(MusicAssistant)
56    mass.loop = asyncio.get_running_loop()
57    mass.loop_thread_id = threading.get_ident()
58    mass._tracked_timers = {}
59    mass._tracked_tasks = {}
60    mass.config = MagicMock()
61    mass.players = MagicMock()
62    mass.players.all_players.return_value = []
63    yield mass
64    for handle in mass._tracked_timers.values():
65        handle.cancel()
66    for task in mass._tracked_tasks.values():
67        task.cancel()
68
69
70def _make_player(mass: MusicAssistant, uid: str, name: str) -> SonosPlayer:
71    """Create a SonosPlayer bound to the given MusicAssistant."""
72    provider = MagicMock()
73    provider.mass = mass
74    provider.topology_condition = asyncio.Condition()
75    return SonosPlayer(provider=provider, soco=_make_soco(uid, name))
76
77
78def _poll_id(player: SonosPlayer) -> str:
79    """Return the task id that debounces the follow-up poll of the given speaker."""
80    return f"sonos_poll_{player.player_id}"
81
82
83def _pending_polls(mass: MusicAssistant) -> list[str]:
84    """Return the task ids of all pending speaker polls."""
85    return sorted(task_id for task_id in mass._tracked_timers if task_id.startswith("sonos_poll_"))
86
87
88def _make_media() -> PlayerMedia:
89    """Return PlayerMedia as built by the queue controller, with an MA media uri."""
90    return PlayerMedia(
91        uri="library://track/123",
92        media_type=MediaType.TRACK,
93        title="Test Track",
94        artist="Test Artist",
95        album="Test Album",
96        duration=180,
97        source_id="queue1",
98        queue_item_id="item1",
99    )
100
101
102async def test_enqueue_next_media_builds_didl_from_stream_url(
103    sonos_player: SonosPlayer,
104) -> None:
105    """The enqueue metadata res element must contain the stream url, not the MA media uri."""
106    await sonos_player.enqueue_next_media(_make_media())
107    call_args = sonos_player.soco.avTransport.SetNextAVTransportURI.call_args
108    args = dict(call_args.args[0])
109    assert args["NextURI"] == STREAM_URL
110    assert STREAM_URL in args["NextURIMetaData"]
111    assert "library://track/123" not in args["NextURIMetaData"]
112
113
114async def test_play_media_builds_didl_from_stream_url(sonos_player: SonosPlayer) -> None:
115    """The play metadata res element must contain the stream url, not the MA media uri."""
116    await sonos_player.play_media(_make_media())
117    call_args = sonos_player.soco.play_uri.call_args
118    assert call_args.args[0] == STREAM_URL
119    assert STREAM_URL in call_args.kwargs["meta"]
120    assert "library://track/123" not in call_args.kwargs["meta"]
121
122
123class _RecordingSoco:
124    """Minimal soco stand-in that records the thread its speaker query ran on."""
125
126    def __init__(self, actions: list[str]) -> None:
127        self._actions = actions
128        self.query_threads: list[int] = []
129        self.paused = False
130
131    @property
132    def available_actions(self) -> list[str]:
133        """Return the transport actions the speaker currently offers."""
134        self.query_threads.append(threading.get_ident())
135        return self._actions
136
137    def pause(self) -> None:
138        """Pause playback on the speaker."""
139        self.paused = True
140
141
142async def test_pause_queries_the_speaker_off_the_event_loop(sonos_player: SonosPlayer) -> None:
143    """Asking a speaker whether it can pause must not stall the event loop."""
144    soco = _RecordingSoco(["Play", "Stop", "Pause"])
145    sonos_player.soco = soco
146
147    await sonos_player.pause()
148
149    assert len(soco.query_threads) == 1
150    assert soco.query_threads[0] != threading.get_ident()
151    assert soco.paused
152
153
154async def test_pause_falls_back_to_stop_when_the_speaker_cannot_pause(
155    sonos_player: SonosPlayer,
156) -> None:
157    """A speaker that offers no pause action is stopped instead."""
158    soco = _RecordingSoco(["Play", "Stop"])
159    sonos_player.soco = soco
160
161    with patch.object(sonos_player, "stop", AsyncMock()) as stop:
162        await sonos_player.pause()
163
164    stop.assert_awaited_once()
165    assert not soco.paused
166
167
168def _set_transport_state(sonos_player: SonosPlayer, state: str) -> None:
169    """Make the mocked speaker report the given transport state."""
170    sonos_player.soco.get_current_transport_info.return_value = {"current_transport_state": state}
171
172
173def test_transitional_state_shortens_the_poll_interval(sonos_player: SonosPlayer) -> None:
174    """A transitional transport state keeps the last known state and is watched closely."""
175    sonos_player._attr_playback_state = PlaybackState.IDLE
176    _set_transport_state(sonos_player, "TRANSITIONING")
177
178    sonos_player.poll_media()
179
180    assert sonos_player._attr_playback_state == PlaybackState.IDLE
181    assert sonos_player.poll_interval == TRANSITION_POLL_INTERVAL
182
183
184def test_settled_state_restores_the_poll_interval(sonos_player: SonosPlayer) -> None:
185    """A usable transport state returns the speaker to the regular poll interval."""
186    sonos_player._attr_poll_interval = TRANSITION_POLL_INTERVAL
187
188    _set_transport_state(sonos_player, "PLAYING")
189    with (
190        patch.object(sonos_player, "_set_basic_track_info"),
191        patch.object(sonos_player, "update_player"),
192    ):
193        sonos_player.poll_media()
194
195    assert sonos_player.poll_interval == POLL_INTERVAL
196
197
198def test_transitional_event_shortens_the_poll_interval(sonos_player: SonosPlayer) -> None:
199    """A transitional state delivered by subscription event is watched closely too."""
200    event = MagicMock()
201    event.variables = {"transport_state": "TRANSITIONING"}
202
203    sonos_player._handle_avtransport_event(event)
204
205    assert sonos_player.poll_interval == TRANSITION_POLL_INTERVAL
206
207
208async def test_repeated_commands_collapse_to_one_pending_poll(
209    timer_mass: MusicAssistant,
210) -> None:
211    """A burst of commands leaves a single pending poll instead of one poll per command."""
212    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
213
214    await player.volume_set(30)
215    first_handle = timer_mass._tracked_timers[_poll_id(player)]
216    await player.volume_set(40)
217    await player.volume_mute(True)
218    await player.play()
219
220    assert _pending_polls(timer_mass) == [_poll_id(player)]
221    assert first_handle.cancelled()
222
223
224async def test_each_speaker_keeps_its_own_pending_poll(timer_mass: MusicAssistant) -> None:
225    """Commands to different speakers are polled independently."""
226    kitchen = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
227    study = _make_player(timer_mass, "RINCON_000E58BBBBBB01400", "Study")
228
229    await kitchen.volume_set(30)
230    await study.volume_set(40)
231
232    assert _pending_polls(timer_mass) == sorted([_poll_id(kitchen), _poll_id(study)])
233
234
235async def test_set_members_polls_the_speakers_it_regrouped(timer_mass: MusicAssistant) -> None:
236    """Grouping polls each speaker that was joined or unjoined, not the coordinator."""
237    kitchen = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
238    study = _make_player(timer_mass, "RINCON_000E58BBBBBB01400", "Study")
239    hallway = _make_player(timer_mass, "RINCON_000E58CCCCCC01400", "Hallway")
240    cast("MagicMock", timer_mass.players).get_player.side_effect = {
241        study.player_id: study,
242        hallway.player_id: hallway,
243    }.get
244
245    await kitchen.set_members(player_ids_to_add=[study.player_id, hallway.player_id])
246
247    assert _pending_polls(timer_mass) == sorted([_poll_id(study), _poll_id(hallway)])
248
249
250async def test_grouping_failure_is_reported_as_a_player_command_failure(
251    timer_mass: MusicAssistant,
252) -> None:
253    """A speaker that refuses to join surfaces as a typed player command failure."""
254    kitchen = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
255    study = _make_player(timer_mass, "RINCON_000E58BBBBBB01400", "Study")
256    cast("MagicMock", timer_mass.players).get_player.side_effect = {study.player_id: study}.get
257    study.soco.join.side_effect = SoCoException("the speaker refused to join")
258
259    with pytest.raises(PlayerCommandFailed, match="Kitchen"):
260        await kitchen.set_members(player_ids_to_add=[study.player_id])
261
262
263async def test_unload_cancels_the_pending_poll(timer_mass: MusicAssistant) -> None:
264    """An unloaded speaker is not polled by a command that was still in flight."""
265    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
266    await player.volume_set(30)
267    handle = timer_mass._tracked_timers[_poll_id(player)]
268
269    await player.on_unload()
270
271    assert _pending_polls(timer_mass) == []
272    assert handle.cancelled()
273
274
275async def test_unload_cancels_a_poll_that_already_started(
276    timer_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
277) -> None:
278    """A poll that started just before the speaker was unloaded is aborted."""
279    monkeypatch.setattr(player_module, "COMMAND_POLL_DELAY", 0)
280    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
281    polling = asyncio.Event()
282
283    async def _slow_poll() -> None:
284        polling.set()
285        await asyncio.sleep(5)
286
287    monkeypatch.setattr(player, "poll", _slow_poll)
288    player.schedule_poll()
289    await polling.wait()
290    task = timer_mass._tracked_tasks[_poll_id(player)]
291
292    await player.on_unload()
293
294    with pytest.raises(asyncio.CancelledError):
295        await task
296
297
298async def test_unloaded_player_ignores_results_from_a_running_poll(
299    timer_mass: MusicAssistant,
300) -> None:
301    """A poll left running in its worker thread cannot update an unloaded speaker."""
302    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
303    await player.on_unload()
304
305    with patch.object(player, "_update_attributes") as update_attributes:
306        player.update_player()
307
308    update_attributes.assert_not_called()
309
310
311async def test_unloaded_player_ignores_group_topology_updates(
312    timer_mass: MusicAssistant,
313) -> None:
314    """A group update raised by a running poll cannot update an unloaded speaker."""
315    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
316    # let the speaker report itself as coordinator of a two-speaker group, so a
317    # topology update would regroup and signal the change if it were not unloaded
318    member = MagicMock(uid="RINCON_000E58BBBBBB01400", is_visible=True)
319    player.soco.group.coordinator.uid = player.player_id
320    player.soco.group.members = [player.soco.group.coordinator, member]
321    await player.on_unload()
322
323    with patch.object(player, "update_state") as update_state:
324        await player.create_update_groups_coro()
325        await asyncio.sleep(0)
326
327    assert player.group_members == []
328    update_state.assert_not_called()
329
330
331async def test_on_unload_unsubscribes_from_soco_events(timer_mass: MusicAssistant) -> None:
332    """Unloading a speaker also tears down its soco event subscriptions."""
333    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
334    subscription = MagicMock()
335    subscription.unsubscribe = AsyncMock()
336    player._subscriptions = [subscription]
337
338    await player.on_unload()
339
340    subscription.unsubscribe.assert_awaited_once()
341    assert player._subscriptions == []
342
343
344async def test_on_unload_unsubscribes_even_when_already_unavailable(
345    timer_mass: MusicAssistant,
346) -> None:
347    """Unlike offline(), on_unload() must still tear down an unavailable speaker's events."""
348    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
349    subscription = MagicMock()
350    subscription.unsubscribe = AsyncMock()
351    player._subscriptions = [subscription]
352    player._attr_available = False
353
354    await player.on_unload()
355
356    subscription.unsubscribe.assert_awaited_once()
357    assert player._subscriptions == []
358
359
360async def test_unloaded_player_is_not_resubscribed_by_a_late_poll(
361    timer_mass: MusicAssistant,
362) -> None:
363    """A poll that reaches an unloaded speaker must not subscribe it to events again."""
364    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
365    # the speaker was marked unavailable earlier, so it carries no subscriptions and
366    # unloading it leaves that unavailable state behind for a late poll to act on
367    player._attr_available = False
368    await player.on_unload()
369
370    with patch.object(player, "subscribe", AsyncMock()) as subscribe:
371        await player.poll()
372
373    subscribe.assert_not_called()
374    assert player._attr_available is False
375
376
377async def test_speaker_answering_a_poll_again_is_resubscribed(
378    timer_mass: MusicAssistant,
379) -> None:
380    """A speaker that answers a poll after being unavailable is subscribed to events again."""
381    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
382    player.soco.group.coordinator.uid = player.player_id
383    player.soco.group.members = [player.soco.group.coordinator]
384    player._attr_available = False
385
386    with patch.object(player, "subscribe", AsyncMock()) as subscribe:
387        await player.poll()
388
389    subscribe.assert_called_once()
390    assert player._attr_available is True
391
392
393async def test_unloading_mid_subscribe_keeps_no_subscriptions(
394    timer_mass: MusicAssistant,
395) -> None:
396    """A speaker unloaded while subscribing must not keep the subscriptions it created."""
397    player = _make_player(timer_mass, "RINCON_000E58AAAAAA01400", "Kitchen")
398    subscribing = asyncio.Event()
399    speaker_responds = asyncio.Event()
400
401    async def _slow_subscribe_target(_target: object, _callback: object) -> None:
402        subscribing.set()
403        await speaker_responds.wait()
404        player._subscriptions.append(MagicMock(unsubscribe=AsyncMock()))
405
406    with patch.object(player, "_subscribe_target", _slow_subscribe_target):
407        subscribe_task = asyncio.create_task(player.subscribe())
408        await subscribing.wait()
409
410        unload_task = asyncio.create_task(player.on_unload())
411        await asyncio.sleep(0)
412        assert not unload_task.done()
413
414        speaker_responds.set()
415        async with asyncio.timeout(5):
416            await asyncio.gather(subscribe_task, unload_task)
417
418    assert player._subscriptions == []
419
420
421def _make_rediscovered_soco(ip_address: str = "127.0.0.2") -> MagicMock:
422    """Create the mocked soco device that discovery hands over for a speaker that moved."""
423    soco = _make_soco()
424    soco.ip_address = ip_address
425    return soco
426
427
428async def test_update_ip_talks_to_the_rediscovered_speaker(sonos_player: SonosPlayer) -> None:
429    """A speaker found at another address is reached through the newly discovered device."""
430    sonos_player._attr_available = False
431    new_soco = _make_rediscovered_soco()
432
433    with patch.object(sonos_player, "setup", AsyncMock()) as setup:
434        await sonos_player.update_ip(new_soco)
435
436    assert sonos_player.soco is new_soco
437    setup.assert_awaited_once()
438    assert sonos_player.device_info.identifiers[IdentifierType.IP_ADDRESS] == "127.0.0.2"
439
440
441async def test_update_ip_probes_the_speaker_off_the_event_loop(sonos_player: SonosPlayer) -> None:
442    """Probing the rediscovered speaker must not stall the event loop."""
443    sonos_player._attr_available = False
444    new_soco = _make_rediscovered_soco()
445    probing = threading.Event()
446    probe_may_finish = threading.Event()
447
448    def _blocking_probe(*_args: object, **_kwargs: object) -> None:
449        probing.set()
450        probe_may_finish.wait(5)
451
452    new_soco.renderingControl.GetVolume.side_effect = _blocking_probe
453
454    # a probe held on the event loop would starve this block until it hits the timeout
455    with patch.object(sonos_player, "setup", AsyncMock()) as setup:
456        async with asyncio.timeout(2):
457            update = asyncio.create_task(sonos_player.update_ip(new_soco))
458            await asyncio.to_thread(probing.wait, 5)
459            probe_may_finish.set()
460            await update
461
462    setup.assert_awaited_once()
463
464
465async def test_update_ip_marks_the_recovered_speaker_available(sonos_player: SonosPlayer) -> None:
466    """A speaker that answers at its new address counts as reachable again."""
467    sonos_player._attr_available = False
468    new_soco = _make_rediscovered_soco()
469
470    with patch.object(sonos_player, "setup", AsyncMock()):
471        await sonos_player.update_ip(new_soco)
472
473    assert sonos_player.available
474
475
476async def test_update_ip_skips_setup_when_the_new_address_stays_silent(
477    sonos_player: SonosPlayer,
478) -> None:
479    """An unanswered probe leaves the speaker for the regular poll to pick up."""
480    sonos_player._attr_available = False
481    new_soco = _make_rediscovered_soco()
482    new_soco.renderingControl.GetVolume.side_effect = SoCoException("no answer")
483
484    with patch.object(sonos_player, "setup", AsyncMock()) as setup:
485        await sonos_player.update_ip(new_soco)
486
487    setup.assert_not_awaited()
488    assert sonos_player.soco is new_soco
489
490
491async def test_update_ip_leaves_an_unloaded_speaker_alone(sonos_player: SonosPlayer) -> None:
492    """An unloaded speaker must not be reconnected, its subscriptions would leak."""
493    sonos_player._attr_available = False
494    sonos_player._unloaded = True
495    original_soco = sonos_player.soco
496
497    with patch.object(sonos_player, "setup", AsyncMock()) as setup:
498        await sonos_player.update_ip(_make_rediscovered_soco())
499
500    assert sonos_player.soco is original_soco
501    setup.assert_not_awaited()
502
503
504async def test_update_ip_leaves_a_responding_speaker_alone(sonos_player: SonosPlayer) -> None:
505    """A speaker that is still reachable keeps the device it is already talking to."""
506    original_soco = sonos_player.soco
507
508    with patch.object(sonos_player, "setup", AsyncMock()) as setup:
509        await sonos_player.update_ip(_make_rediscovered_soco())
510
511    assert sonos_player.soco is original_soco
512    setup.assert_not_awaited()
513
514
515async def test_setup_reads_the_speaker_off_the_event_loop(sonos_player: SonosPlayer) -> None:
516    """Reading the initial state of a speaker must not stall the event loop."""
517    cast("MagicMock", sonos_player.mass.players).register_or_update = AsyncMock()
518    reading_threads: list[int] = []
519
520    with (
521        patch.object(
522            sonos_player,
523            "update_groups",
524            lambda: reading_threads.append(threading.get_ident()),
525        ),
526        patch.object(sonos_player, "poll_media"),
527        patch.object(sonos_player, "subscribe", AsyncMock()),
528    ):
529        await sonos_player.setup()
530
531    assert len(reading_threads) == 1
532    assert reading_threads[0] != threading.get_ident()
533
534
535async def test_unsubscribe_drops_subscriptions_even_when_cancelled() -> None:
536    """A cancelled unsubscribe must not leave stale entries that block resubscribing."""
537    provider = MagicMock()
538    player = SonosPlayer(provider=provider, soco=_make_soco())
539    subscription = MagicMock()
540    subscription.unsubscribe = AsyncMock(side_effect=partial(asyncio.sleep, 5))
541    player._subscriptions = [subscription]
542
543    task = asyncio.create_task(player.unsubscribe())
544    await asyncio.sleep(0)
545    task.cancel()
546    with pytest.raises(asyncio.CancelledError):
547        await task
548
549    assert player._subscriptions == []
550    assert player.missing_subscriptions == SUBSCRIPTION_SERVICES
551
552
553async def _subscribe_with_failing_speaker(player: SonosPlayer) -> None:
554    """Let the given player attempt to subscribe to a speaker that cannot be reached."""
555    with (
556        patch.object(player, "_subscribe_target", AsyncMock(side_effect=OSError("unreachable"))),
557        patch.object(player, "update_state"),
558    ):
559        async with asyncio.timeout(5):
560            await player.subscribe()
561
562
563async def test_failed_subscribe_marks_the_speaker_offline() -> None:
564    """A failed subscription must take the speaker offline and release the lock."""
565    player = SonosPlayer(provider=MagicMock(), soco=_make_soco())
566
567    await _subscribe_with_failing_speaker(player)
568
569    assert player.available is False
570    assert not player._subscription_lock.locked()
571
572
573async def test_speaker_can_resubscribe_after_a_failed_subscribe() -> None:
574    """A speaker that failed to subscribe must still be able to subscribe later."""
575    player = SonosPlayer(provider=MagicMock(), soco=_make_soco())
576    await _subscribe_with_failing_speaker(player)
577
578    with patch.object(player, "_subscribe_target", AsyncMock()) as subscribe_target:
579        async with asyncio.timeout(5):
580            await player.subscribe()
581
582    assert subscribe_target.await_count == len(SUBSCRIPTION_SERVICES)
583
584
585async def test_speaker_taken_offline_mid_subscribe_keeps_no_subscriptions() -> None:
586    """A speaker that goes offline while subscribing must not keep the subscriptions it created."""
587    player = SonosPlayer(provider=MagicMock(), soco=_make_soco())
588    subscribing = asyncio.Event()
589    speaker_responds = asyncio.Event()
590
591    async def _slow_subscribe_target(_target: object, _callback: object) -> None:
592        subscribing.set()
593        await speaker_responds.wait()
594        player._subscriptions.append(MagicMock(unsubscribe=AsyncMock()))
595
596    with (
597        patch.object(player, "_subscribe_target", _slow_subscribe_target),
598        patch.object(player, "update_state"),
599    ):
600        subscribe_task = asyncio.create_task(player.subscribe())
601        await subscribing.wait()
602
603        offline_task = asyncio.create_task(player.offline())
604        await asyncio.sleep(0)
605        assert not offline_task.done()
606
607        speaker_responds.set()
608        async with asyncio.timeout(5):
609            await asyncio.gather(subscribe_task, offline_task)
610
611    assert player.available is False
612    assert player._subscriptions == []
613    assert player.missing_subscriptions == SUBSCRIPTION_SERVICES
614
615
616async def test_speaker_going_offline_is_not_resubscribed_halfway() -> None:
617    """No new subscriptions may be created while a speaker is still going offline."""
618    player = SonosPlayer(provider=MagicMock(), soco=_make_soco())
619    unsubscribing = asyncio.Event()
620    speaker_responds = asyncio.Event()
621
622    async def _slow_unsubscribe() -> None:
623        unsubscribing.set()
624        await speaker_responds.wait()
625
626    player._subscriptions = [MagicMock(unsubscribe=_slow_unsubscribe)]
627
628    with (
629        patch.object(player, "_subscribe_target", AsyncMock()) as subscribe_target,
630        patch.object(player, "update_state"),
631    ):
632        offline_task = asyncio.create_task(player.offline())
633        await unsubscribing.wait()
634
635        subscribe_task = asyncio.create_task(player.subscribe())
636        await asyncio.sleep(0)
637        subscribe_target.assert_not_called()
638
639        speaker_responds.set()
640        async with asyncio.timeout(5):
641            await asyncio.gather(offline_task, subscribe_task)
642
643    assert subscribe_target.await_count == len(SUBSCRIPTION_SERVICES)
644