music-assistant-server

1"""
2Tests for the experimental Alice-playback intercept feature.
3
4When Alice (Yandex voice assistant) starts music on a Station, the intercept
5feature stops the Station's native player, resolves the track via the
6``yandex_music`` MA music provider, and starts playback on a configured target
7player.  Volume / seek / pause changes on the Station mirror to the target.
8
9The feature is gated by two switches: a provider-level master toggle
10(``intercept_feature_enabled``, default OFF) and a per-player toggle
11(``intercept_enabled``).  Both must be ON for any intercept action to happen.
12"""
13# Tests use MagicMock to stand in for MA core objects whose real types are
14# Callable / Player / etc.  Mypy strict-mode flags every ``assert_awaited_*`` as
15# attr-defined, every mock reassignment as method-assign, and the master-switch
16# branch in ``test_intercept_master_switch_off`` as unreachable (because
17# ``_intercept_enabled`` returns False there).  All three are expected here.
18# mypy: disable-error-code="attr-defined,method-assign,unreachable"
19
20from __future__ import annotations
21
22import asyncio
23import logging
24import time
25from typing import TYPE_CHECKING, Any
26from unittest.mock import AsyncMock, MagicMock
27
28from music_assistant_models.enums import PlaybackState, PlayerFeature, PlayerType
29from music_assistant_models.errors import UnsupportedFeaturedException
30
31if TYPE_CHECKING:
32    import pytest
33
34from music_assistant.providers.yandex_station.constants import (
35    CONF_INTERCEPT_ENABLED,
36    CONF_INTERCEPT_TARGET,
37)
38from music_assistant.providers.yandex_station.player import (
39    YandexStationPlayer,
40    _parse_yandex_track_id,
41)
42
43# ── Fixtures ──────────────────────────────────────────────────────────
44
45
46def _make_intercept_player(
47    *,
48    feature_enabled: bool = True,
49    per_player_enabled: bool = True,
50    target_player_id: str | None = "target_player",
51    yandex_music_present: bool = True,
52    external_playing: bool = False,
53) -> YandexStationPlayer:
54    """Build a player with intercept-related state and mocked mass."""
55    player = YandexStationPlayer.__new__(YandexStationPlayer)
56    player._player_id = "yandex_station_1"
57    player._external_playing = external_playing
58    player._external_media = None
59    player._intercept_active = False
60    player._last_intercepted_track_id = None
61    player._last_intercept_time = 0.0
62    player._last_mirrored_volume = None
63    player._last_progress = 0
64    player._last_progress_wall = 0.0
65    player._intercept_lock = asyncio.Lock()
66    player._alice_active_pause_sent = False
67    player._saved_station_volume = None
68    player._station_muted_by_intercept = False
69    player._prev_alice_state = ""
70    player._attr_volume_level = 50  # baseline for saved-volume capture
71
72    # Mock provider config (master switch) and player config (per-player toggle)
73    provider_config = MagicMock()
74    provider_config.get_value = MagicMock(return_value=feature_enabled)
75    provider = MagicMock()
76    provider.config = provider_config
77    player._provider = provider
78
79    def _player_cfg_get(key: str, default: object = None) -> object:
80        if key == CONF_INTERCEPT_ENABLED:
81            return per_player_enabled
82        if key == CONF_INTERCEPT_TARGET:
83            return target_player_id
84        return default
85
86    player_config = MagicMock()
87    player_config.get_value = MagicMock(side_effect=_player_cfg_get)
88    player._config = player_config
89
90    # Mock mass with the four touchpoints intercept uses
91    mass = MagicMock()
92    mass.get_provider = MagicMock(return_value=MagicMock() if yandex_music_present else None)
93    fake_track = MagicMock(name="resolved_track")
94    mass.music = MagicMock()
95    mass.music.get_item = AsyncMock(return_value=fake_track)
96    mass.player_queues = MagicMock()
97    mass.player_queues.play_media = AsyncMock()
98    mass.players = MagicMock()
99    mass.players.cmd_pause = AsyncMock()
100    mass.players.cmd_volume_set = AsyncMock()
101    mass.players.cmd_seek = AsyncMock()
102    # Default: target player exists (intercept pre-validation passes).
103    mass.players.get_player = MagicMock(return_value=MagicMock(name="target_player_obj"))
104    player.mass = mass
105
106    # Mock glagol with successful stop
107    player.glagol = MagicMock()
108    player.glagol.send = AsyncMock(return_value={"status": "SUCCESS"})
109
110    return player
111
112
113def _state(
114    *,
115    track_id: str = "12345",
116    playing: bool = True,
117    volume: float | None = 0.5,
118    progress: int = 0,
119    alice_state: str = "IDLE",
120) -> tuple[dict[str, Any], dict[str, Any], bool]:
121    """Build a (state, player_state, playing) tuple for _handle_intercept_tick."""
122    player_state = {"id": track_id, "progress": progress, "title": "Some Track"}
123    state: dict[str, Any] = {
124        "playerState": player_state,
125        "playing": playing,
126        "aliceState": alice_state,
127    }
128    if volume is not None:
129        state["volume"] = volume
130    return state, player_state, playing
131
132
133# ── Helper: track_id parser ───────────────────────────────────────────
134
135
136def test_parse_yandex_track_id_plain() -> None:
137    """Plain numeric ID passes through unchanged."""
138    assert _parse_yandex_track_id("12345") == "12345"
139
140
141def test_parse_yandex_track_id_with_album_suffix() -> None:
142    """`track:album` form drops the album suffix."""
143    assert _parse_yandex_track_id("12345:67890") == "12345"
144
145
146def test_parse_yandex_track_id_strips_whitespace() -> None:
147    """Surrounding whitespace is trimmed."""
148    assert _parse_yandex_track_id(" 12345 ") == "12345"
149
150
151def test_parse_yandex_track_id_empty() -> None:
152    """Empty input maps to empty string (callers must guard)."""
153    assert _parse_yandex_track_id("") == ""
154
155
156# ── Toggle / kill switch behaviour ────────────────────────────────────
157
158
159async def test_intercept_triggers_on_alice_play() -> None:
160    """Both switches ON, target set, yandex_music present → full intercept flow."""
161    player = _make_intercept_player()
162    state, player_state, playing = _state(track_id="12345")
163
164    await player._handle_intercept_tick(state, player_state, playing)
165
166    # Only mute(0) is sent — no `stop`, so the Station keeps emitting
167    # playerState ticks for each next track Alice queues (continuous handoff).
168    sent_payloads = [c.args[0] for c in player.glagol.send.await_args_list]
169    assert sent_payloads == [{"command": "setVolume", "volume": 0.0}]
170    player.mass.music.get_item.assert_awaited_once()
171    kwargs = player.mass.music.get_item.await_args.kwargs
172    assert kwargs["item_id"] == "12345"
173    assert kwargs["provider_instance_id_or_domain"] == "yandex_music"
174    player.mass.player_queues.play_media.assert_awaited_once()
175    play_kwargs = player.mass.player_queues.play_media.await_args.kwargs
176    assert play_kwargs["queue_id"] == "target_player"
177    assert player._intercept_active is True
178    assert player._last_intercepted_track_id == "12345"
179
180
181async def test_intercept_master_switch_off() -> None:
182    """Provider master toggle OFF → no action, even with per-player ON."""
183    player = _make_intercept_player(feature_enabled=False)
184    state, player_state, playing = _state()
185
186    # Real entrypoint guard is in _on_glagol_update, but verify _intercept_enabled
187    assert player._intercept_enabled is False
188
189    # Simulate the guard explicitly: tick should not be dispatched
190    if player._intercept_enabled and player._intercept_target_player_id:
191        await player._handle_intercept_tick(state, player_state, playing)
192
193    player.glagol.send.assert_not_awaited()
194    player.mass.music.get_item.assert_not_awaited()
195    player.mass.player_queues.play_media.assert_not_awaited()
196
197
198async def test_intercept_disabled_per_player() -> None:
199    """Master toggle ON, per-player OFF → no action."""
200    player = _make_intercept_player(per_player_enabled=False)
201
202    assert player._intercept_enabled is False
203
204
205# ── Failure paths ─────────────────────────────────────────────────────
206
207
208async def test_intercept_no_yandex_music_provider() -> None:
209    """Missing yandex_music provider → no stop, no play, just log."""
210    player = _make_intercept_player(yandex_music_present=False)
211    state, player_state, playing = _state()
212
213    await player._handle_intercept_tick(state, player_state, playing)
214
215    player.glagol.send.assert_not_awaited()
216    player.mass.music.get_item.assert_not_awaited()
217    player.mass.player_queues.play_media.assert_not_awaited()
218    assert player._intercept_active is False
219
220
221async def test_intercept_no_target_configured() -> None:
222    """Target player_id unset → no action."""
223    player = _make_intercept_player(target_player_id=None)
224    state, player_state, playing = _state()
225
226    await player._handle_intercept_tick(state, player_state, playing)
227
228    player.glagol.send.assert_not_awaited()
229    assert player._intercept_active is False
230
231
232async def test_intercept_during_external_playing() -> None:
233    """Our own bypass stream is playing → never intercept (anti-loop)."""
234    player = _make_intercept_player(external_playing=True)
235    state, player_state, playing = _state()
236
237    await player._handle_intercept_tick(state, player_state, playing)
238
239    player.glagol.send.assert_not_awaited()
240    player.mass.music.get_item.assert_not_awaited()
241
242
243async def test_intercept_dedup_same_track_within_window() -> None:
244    """Same track_id within 5s → second call is a no-op."""
245    player = _make_intercept_player()
246    state, player_state, playing = _state(track_id="999")
247
248    await player._handle_intercept_tick(state, player_state, playing)
249    await player._handle_intercept_tick(state, player_state, playing)
250
251    assert player.mass.player_queues.play_media.await_count == 1
252    # 1 send (mute(0) only — no stop) on the first tick; second debounced.
253    assert player.glagol.send.await_count == 1
254
255
256async def test_intercept_resolve_failure_does_not_silence_station() -> None:
257    """If get_item raises, the Station is left playing — never silenced."""
258    player = _make_intercept_player()
259    player.mass.music.get_item = AsyncMock(side_effect=RuntimeError("not found"))
260    state, player_state, playing = _state()
261
262    await player._handle_intercept_tick(state, player_state, playing)
263
264    # Resolve happens FIRST — failure means the Station stays playing.
265    player.glagol.send.assert_not_awaited()
266    player.mass.player_queues.play_media.assert_not_awaited()
267    assert player._intercept_active is False
268
269
270async def test_intercept_resolved_track_without_uri_skips_handoff() -> None:
271    """Resolved track with no uri → log warning, leave Station playing."""
272    player = _make_intercept_player()
273    bad_track = MagicMock(name="track_no_uri")
274    bad_track.uri = None
275    player.mass.music.get_item = AsyncMock(return_value=bad_track)
276    state, player_state, playing = _state()
277
278    await player._handle_intercept_tick(state, player_state, playing)
279
280    player.glagol.send.assert_not_awaited()
281    player.mass.player_queues.play_media.assert_not_awaited()
282    assert player._intercept_active is False
283
284
285# ── Mirroring ─────────────────────────────────────────────────────────
286
287
288async def test_volume_mirror_swallows_unsupported_feature() -> None:
289    """Targets without VOLUME_SET raise UnsupportedFeaturedException → log + no-op."""
290    player = _make_intercept_player()
291    player._intercept_active = True
292    player.mass.players.cmd_volume_set = AsyncMock(side_effect=UnsupportedFeaturedException("nope"))
293
294    # Should not raise
295    await player._maybe_mirror_volume(0.5)
296
297    player.mass.players.cmd_volume_set.assert_awaited_once()
298    # Stamp is still updated so we don't retry on every tick.
299    assert player._last_mirrored_volume == 50
300
301
302async def test_seek_mirror_swallows_unsupported_feature() -> None:
303    """Targets without SEEK raise UnsupportedFeaturedException → log + no-op."""
304    player = _make_intercept_player()
305    player._intercept_active = True
306    player._last_progress = 10
307    player._last_progress_wall = time.time() - 1  # 1s ago
308    player.mass.players.cmd_seek = AsyncMock(side_effect=UnsupportedFeaturedException("nope"))
309
310    # progress jump ~50s in 1s → would normally trigger cmd_seek
311    await player._maybe_mirror_seek(60)
312
313    player.mass.players.cmd_seek.assert_awaited_once()
314
315
316async def test_volume_mirror_after_intercept() -> None:
317    """Volume changes on the Station mirror to the target while active."""
318    player = _make_intercept_player()
319    state, player_state, _ = _state(track_id="1", volume=0.4)
320
321    # First tick triggers intercept
322    await player._handle_intercept_tick(state, player_state, True)
323    assert player._intercept_active is True
324    # Volume mirror happens in same tick
325    player.mass.players.cmd_volume_set.assert_awaited_with("target_player", 40)
326
327    # New tick with same track_id (within debounce) and different volume
328    state2, ps2, _ = _state(track_id="1", volume=0.7)
329    await player._handle_intercept_tick(state2, ps2, True)
330    player.mass.players.cmd_volume_set.assert_awaited_with("target_player", 70)
331
332
333async def test_volume_mirror_skipped_when_unchanged() -> None:
334    """Identical volume in consecutive ticks → only one cmd_volume_set call."""
335    player = _make_intercept_player()
336    state, player_state, _ = _state(track_id="1", volume=0.5)
337
338    await player._handle_intercept_tick(state, player_state, True)
339    await player._handle_intercept_tick(state, player_state, True)
340
341    # exactly one volume command for the value 50
342    calls = [c.args for c in player.mass.players.cmd_volume_set.await_args_list]
343    assert calls == [("target_player", 50)]
344
345
346async def test_seek_mirror_on_progress_jump() -> None:
347    """Progress jumps far ahead of wall-clock prediction → cmd_seek on target."""
348    player = _make_intercept_player()
349    # First tick establishes intercept and the progress baseline
350    state1, ps1, _ = _state(track_id="1", progress=10)
351    await player._handle_intercept_tick(state1, ps1, True)
352    assert player._intercept_active is True
353
354    # Same track, but progress jumped to 60 — must be detected as a seek
355    state2, ps2, _ = _state(track_id="1", progress=60)
356    await player._handle_intercept_tick(state2, ps2, True)
357
358    player.mass.players.cmd_seek.assert_awaited_with("target_player", 60)
359
360
361# ── Voice interrupt + intercept ───────────────────────────────────────
362
363
364async def test_alice_speaks_during_intercept_pauses_target_via_dispatcher() -> None:
365    """
366    Alice activity arrives via Glagol state — dispatcher pauses target.
367
368    This drives ``_handle_intercept_tick`` (the actual entry point), not the
369    bypass-only ``_handle_voice_interrupt`` helper.  The intercept session
370    must remain active so a follow-up Alice-initiated track resumes it.
371    """
372    player = _make_intercept_player()
373    player._intercept_active = True
374
375    # Same track_id as last_intercepted → debounce skips re-intercept;
376    # Alice activity must still pause the target.
377    player._last_intercepted_track_id = "12345"
378    player._last_intercept_time = time.time()
379    state, player_state, _ = _state(track_id="12345", alice_state="LISTENING")
380
381    await player._handle_intercept_tick(state, player_state, True)
382
383    player.mass.players.cmd_pause.assert_awaited_with("target_player")
384    # Session stays open so the next Alice track can resume it
385    assert player._intercept_active is True
386
387
388async def test_alice_idle_during_intercept_does_not_pause_target() -> None:
389    """No Alice activity → no spurious pause on the target."""
390    player = _make_intercept_player()
391    player._intercept_active = True
392    player._last_intercepted_track_id = "12345"
393    player._last_intercept_time = time.time()
394    state, player_state, _ = _state(track_id="12345", alice_state="IDLE")
395
396    await player._handle_intercept_tick(state, player_state, True)
397
398    player.mass.players.cmd_pause.assert_not_awaited()
399
400
401# ── Stale session / debounce on failure / serialisation ──────────────
402
403
404async def test_failed_intercept_debounces_to_avoid_log_spam() -> None:
405    """
406    Repeated WS ticks for the same failing track → only one resolve attempt.
407
408    Failed lookups must update the debounce timestamp; otherwise every Glagol
409    tick (~1Hz) would re-run get_item and emit a fresh warning.
410    """
411    player = _make_intercept_player()
412    player.mass.music.get_item = AsyncMock(side_effect=RuntimeError("not found"))
413    state, player_state, _ = _state(track_id="bad")
414
415    await player._handle_intercept_tick(state, player_state, True)
416    await player._handle_intercept_tick(state, player_state, True)
417    await player._handle_intercept_tick(state, player_state, True)
418
419    assert player.mass.music.get_item.await_count == 1
420
421
422async def test_target_player_unavailable_does_not_silence_station() -> None:
423    """Pre-validation: if get_player(target) returns None, no Glagol stop."""
424    player = _make_intercept_player()
425    player.mass.players.get_player = MagicMock(return_value=None)
426    state, player_state, _ = _state()
427
428    await player._handle_intercept_tick(state, player_state, True)
429
430    player.glagol.send.assert_not_awaited()
431    player.mass.player_queues.play_media.assert_not_awaited()
432
433
434async def test_failed_intercept_on_new_track_ends_stale_session() -> None:
435    """
436    A new track that fails to resolve must pause the target from the prior session.
437
438    Otherwise mirror updates from the Station's native fallback playback would
439    keep being forwarded to the target that's still on the previous track.
440    """
441    player = _make_intercept_player()
442    # Simulate a prior successful intercept on track A.
443    player._intercept_active = True
444    player._last_intercepted_track_id = "A"
445    player._last_intercept_time = 0.0  # well outside debounce
446    # New track B fails to resolve.
447    player.mass.music.get_item = AsyncMock(side_effect=RuntimeError("nope"))
448    state, player_state, _ = _state(track_id="B")
449
450    await player._handle_intercept_tick(state, player_state, True)
451
452    player.mass.players.cmd_pause.assert_awaited_with("target_player")
453    assert player._intercept_active is False
454
455
456async def test_handoff_failure_clears_intercept_active() -> None:
457    """
458    Failed handoff after mute must clear intercept_active.
459
460    Otherwise mirror code would forward state to a target that isn't playing.
461    Volume is also restored via _end_intercept_session so the Station isn't
462    stuck muted with no way for the user to recover.
463    """
464    player = _make_intercept_player()
465    player.mass.player_queues.play_media = AsyncMock(side_effect=RuntimeError("boom"))
466    state, player_state, _ = _state()
467
468    await player._handle_intercept_tick(state, player_state, True)
469
470    # mute(0) fired (resolve succeeded) + restore on session-end cleanup → 2 sends.
471    sent_payloads = [c.args[0] for c in player.glagol.send.await_args_list]
472    assert sent_payloads == [
473        {"command": "setVolume", "volume": 0.0},
474        {"command": "setVolume", "volume": 0.5},  # 50/100 from fixture baseline
475    ]
476    assert player._intercept_active is False
477
478
479async def test_session_end_clears_debounce_for_quick_resume() -> None:
480    """
481    End of session must clear the debounce for quick same-track resumes.
482
483    Otherwise a follow-up of the same track within 5s would be debounced and
484    left playing on the Station instead of being handed back to the target.
485    """
486    player = _make_intercept_player()
487    player._intercept_active = True
488    player._last_intercepted_track_id = "X"
489    player._last_intercept_time = time.time()
490
491    await player._pause_target(clear_session=True, clear_debounce=True)
492
493    assert player._intercept_active is False
494    assert player._last_intercepted_track_id is None
495    assert player._last_intercept_time == 0.0
496
497
498async def test_concurrent_ticks_do_not_double_handoff() -> None:
499    """
500    Two near-simultaneous WS ticks for the same track must only stop+play once.
501
502    Without the lock both tasks would pass the dedup check, both would call
503    glagol.send(stop) and play_media.  The lock + early debounce-mark serialise
504    them so the second one short-circuits.
505    """
506    player = _make_intercept_player()
507    state, player_state, _ = _state(track_id="X")
508
509    # Make get_item slow so the second tick definitely arrives mid-handoff.
510    resolve_started = asyncio.Event()
511    resolve_release = asyncio.Event()
512
513    async def slow_get_item(**_kwargs: Any) -> Any:
514        resolve_started.set()
515        await resolve_release.wait()
516        return MagicMock(uri="yandex_music://track/X")
517
518    player.mass.music.get_item = AsyncMock(side_effect=slow_get_item)
519
520    t1 = asyncio.create_task(player._handle_intercept_tick(state, player_state, True))
521    await resolve_started.wait()
522    # Second tick fires while first is still inside _maybe_intercept's lock.
523    t2 = asyncio.create_task(player._handle_intercept_tick(state, player_state, True))
524    # Give t2 a chance to acquire the lock and hit the dedup check.
525    await asyncio.sleep(0)
526    resolve_release.set()
527    await asyncio.gather(t1, t2)
528
529    # First tick: mute(0) only = 1 send (no stop in continuous-playback mode).
530    # Second tick: short-circuits via debounce → still 1 total.
531    assert player.glagol.send.await_count == 1
532    assert player.mass.player_queues.play_media.await_count == 1
533
534
535# ── Continuous playback (v1.4.14) ────────────────────────────────────
536
537
538async def test_intercept_does_not_send_stop() -> None:
539    """
540    Continuous-playback contract: a handoff must never send {"command":"stop"}.
541
542    Sending stop pauses the Station's queue → no more playerState ticks → no
543    next-track handoff.  We only ever mute via setVolume(0).
544    """
545    player = _make_intercept_player()
546    state, player_state, _ = _state(track_id="42")
547
548    await player._handle_intercept_tick(state, player_state, True)
549
550    sent = [c.args[0] for c in player.glagol.send.await_args_list]
551    assert all(p.get("command") != "stop" for p in sent)
552    assert {"command": "setVolume", "volume": 0.0} in sent
553
554
555async def test_continuous_handoff_on_track_id_change() -> None:
556    """Subsequent track_id → second handoff. Station muted ONCE per session."""
557    player = _make_intercept_player()
558
559    state_a, ps_a, _ = _state(track_id="trackA")
560    await player._handle_intercept_tick(state_a, ps_a, True)
561    # Move beyond the 5s same-track debounce by rewinding _last_intercept_time.
562    player._last_intercept_time -= 10
563    state_b, ps_b, _ = _state(track_id="trackB")
564    await player._handle_intercept_tick(state_b, ps_b, True)
565
566    # Two handoffs, one per track.
567    assert player.mass.player_queues.play_media.await_count == 2
568    # Mute(0) is sent only ONCE — at session start.  Subsequent handoffs
569    # don't re-mute (Station already at vol=0).
570    mute_sends = [
571        c.args[0]
572        for c in player.glagol.send.await_args_list
573        if c.args[0] == {"command": "setVolume", "volume": 0.0}
574    ]
575    assert len(mute_sends) == 1
576    assert player._last_intercepted_track_id == "trackB"
577
578
579async def test_same_track_during_active_session_is_no_op() -> None:
580    """
581    Same playerState.id on every WS tick must NOT re-trigger handoff.
582
583    Regression guard for the live-station bug where the target's audio
584    stuttered every ~5s.  Glagol emits ``playerState`` once per second
585    for the entire track duration (3-5min) carrying the same ``id``;
586    the original 5-second failure-debounce expired mid-track and let
587    every subsequent tick fire a fresh ``play_media(REPLACE)``.  Once a
588    track is handed off (``_intercept_active=True``), the same id must
589    short-circuit regardless of how much time has passed.
590    """
591    player = _make_intercept_player()
592    # Establish a session with track X already handed off.
593    player._intercept_active = True
594    player._last_intercepted_track_id = "X"
595    player._last_intercept_time = time.time() - 100  # well past 5s debounce
596
597    state, player_state, _ = _state(track_id="X")
598    await player._handle_intercept_tick(state, player_state, True)
599
600    # No new handoff, no API churn.
601    player.mass.music.get_item.assert_not_awaited()
602    player.mass.player_queues.play_media.assert_not_awaited()
603    player.glagol.send.assert_not_awaited()
604
605
606async def test_session_end_restores_station_volume() -> None:
607    """_end_intercept_session must send setVolume(saved/100) back to the Station."""
608    player = _make_intercept_player()
609    player._intercept_active = True
610    player._saved_station_volume = 70
611    player._station_muted_by_intercept = True
612
613    await player._end_intercept_session(clear_debounce=True)
614
615    sent = [c.args[0] for c in player.glagol.send.await_args_list]
616    assert {"command": "setVolume", "volume": 0.7} in sent
617    assert player._intercept_active is False
618    assert player._saved_station_volume is None
619    assert player._station_muted_by_intercept is False
620    assert player._last_intercepted_track_id is None
621
622
623async def test_session_end_when_not_muted_does_not_send_volume() -> None:
624    """If we never muted (e.g. user beat us to it via app), don't send volume."""
625    player = _make_intercept_player()
626    player._intercept_active = True
627    player._saved_station_volume = 70
628    player._station_muted_by_intercept = False
629
630    await player._end_intercept_session(clear_debounce=False)
631
632    player.glagol.send.assert_not_awaited()
633    assert player._intercept_active is False
634
635
636async def test_volume_mirror_skips_zero_during_session() -> None:
637    """Self-induced mute must not propagate to target."""
638    player = _make_intercept_player()
639    player._intercept_active = True
640    player._station_muted_by_intercept = True
641    player._saved_station_volume = 60
642
643    await player._maybe_mirror_volume(0.0)
644
645    player.mass.players.cmd_volume_set.assert_not_awaited()
646
647
648async def test_volume_mirror_allows_zero_when_no_session() -> None:
649    """Outside a session, vol=0 must mirror normally."""
650    player = _make_intercept_player()
651    player._intercept_active = False
652
653    await player._maybe_mirror_volume(0.0)
654
655    player.mass.players.cmd_volume_set.assert_awaited_once_with("target_player", 0)
656
657
658async def test_user_unmute_via_yandex_app_clears_self_mute_flag() -> None:
659    """Station vol > 0 mid-session: clear self-mute flag, update saved baseline."""
660    player = _make_intercept_player()
661    player._intercept_active = True
662    player._station_muted_by_intercept = True
663    player._saved_station_volume = 50
664
665    await player._maybe_mirror_volume(0.8)
666
667    player.mass.players.cmd_volume_set.assert_awaited_once_with("target_player", 80)
668    assert player._station_muted_by_intercept is False
669    assert player._saved_station_volume == 80
670
671
672async def test_alice_active_unmutes_station() -> None:
673    """LISTENING/SPEAKING during a session → restore Station volume for Alice TTS."""
674    player = _make_intercept_player()
675    player._intercept_active = True
676    player._saved_station_volume = 70
677    player._station_muted_by_intercept = True
678    state, player_state, _ = _state(track_id="X", alice_state="LISTENING")
679
680    await player._handle_intercept_tick(state, player_state, True)
681
682    sent = [c.args[0] for c in player.glagol.send.await_args_list]
683    assert {"command": "setVolume", "volume": 0.7} in sent
684    assert player._station_muted_by_intercept is False
685    # mirror baseline pre-set so the next vol-tick doesn't bounce to target
686    assert player._last_mirrored_volume == 70
687    # target paused once
688    player.mass.players.cmd_pause.assert_awaited_once_with("target_player")
689
690
691async def test_alice_idle_remutes_station() -> None:
692    """
693    LISTENING → IDLE edge: re-mute Station now that Alice is done.
694
695    Previous alice state is threaded in as a parameter (snapshot taken
696    *before* the dispatcher overwrote ``_prev_alice_state``), since the
697    dispatcher schedules the tick via ``mass.create_task`` and the field
698    would otherwise read the post-assignment current state by the time
699    the tick runs.
700    """
701    player = _make_intercept_player()
702    player._intercept_active = True
703    player._saved_station_volume = 70
704    player._station_muted_by_intercept = False  # currently unmuted (alice was active)
705    # Prevent the playing=False session-end branch from firing in this test —
706    # without an established track id, the early-return short-circuits.
707    player._last_intercepted_track_id = None
708    state, player_state, _ = _state(track_id="X", alice_state="IDLE", playing=False)
709
710    await player._handle_intercept_tick(state, player_state, False, prev_alice_state="SPEAKING")
711
712    sent = [c.args[0] for c in player.glagol.send.await_args_list]
713    assert {"command": "setVolume", "volume": 0.0} in sent
714    assert player._station_muted_by_intercept is True
715
716
717async def test_playing_false_during_session_ends_session_and_pauses_target() -> None:
718    """
719    Physical pause / 'Алиса, пауза' / end-of-queue → end session.
720
721    We can't reliably distinguish a transient user pause from end-of-queue
722    on a single ``playing=False`` event, so always end the session — that
723    way Station volume is restored even when the queue ends for good.
724    Cost: ~one WS round-trip of native audio on quick resume before the
725    new session's mute(0) lands.
726    """
727    player = _make_intercept_player()
728    player._intercept_active = True
729    player._last_intercepted_track_id = "X"  # established session
730    player._last_intercept_time = time.time()
731    player._saved_station_volume = 50
732    player._station_muted_by_intercept = True
733    state, player_state, _ = _state(track_id="X", playing=False, alice_state="IDLE")
734
735    await player._handle_intercept_tick(state, player_state, False)
736
737    player.mass.players.cmd_pause.assert_awaited_once_with("target_player")
738    # Session ended → Station volume restored, flags cleared.
739    sent = [c.args[0] for c in player.glagol.send.await_args_list]
740    assert {"command": "setVolume", "volume": 0.5} in sent
741    assert player._intercept_active is False
742    assert player._last_intercepted_track_id is None
743    assert player._saved_station_volume is None
744    assert player._station_muted_by_intercept is False
745
746
747async def test_playing_false_without_established_session_does_not_pause() -> None:
748    """
749    Lingering playing=False before any track was intercepted is a no-op.
750
751    Replaces the old test_session_survives_lingering_playing_false but with the
752    correct invariant: we only treat playing=False as 'user paused' when a
753    session has actually established a track (debounce non-None).
754    """
755    player = _make_intercept_player()
756    player._intercept_active = True
757    player._last_intercepted_track_id = None  # no track yet
758    state, player_state, _ = _state(track_id="X", playing=False, alice_state="IDLE")
759
760    await player._handle_intercept_tick(state, player_state, False)
761
762    player.mass.players.cmd_pause.assert_not_awaited()
763    assert player._intercept_active is True
764
765
766async def test_pause_target_clear_session_restores_station_volume() -> None:
767    """_pause_target(clear_session=True) funnels through _end_intercept_session."""
768    player = _make_intercept_player()
769    player._intercept_active = True
770    player._saved_station_volume = 40
771    player._station_muted_by_intercept = True
772
773    await player._pause_target(clear_session=True, clear_debounce=False)
774
775    player.mass.players.cmd_pause.assert_awaited_once_with("target_player")
776    sent = [c.args[0] for c in player.glagol.send.await_args_list]
777    assert {"command": "setVolume", "volume": 0.4} in sent
778    assert player._intercept_active is False
779    assert player._saved_station_volume is None
780
781
782# ── glagol.send() result validation (PR #3605 review) ────────────────
783
784
785async def test_handoff_aborts_on_mute_send_error() -> None:
786    """
787    Mute-send transport error must abort the handoff, not silently proceed.
788
789    glagol.send returns {"error": ...} for transport failures rather than
790    raising — without explicit validation we'd flip _station_muted_by_intercept
791    to True and start the target while the Station is still audible.
792    _raise_if_failed must convert the error into PlayerCommandFailed so the
793    surrounding handoff try/except runs the session-end cleanup path.
794    """
795    player = _make_intercept_player()
796    # Mute send fails with the standard transport-error envelope
797    player.glagol.send = AsyncMock(return_value={"error": "timeout"})
798    state, player_state, _ = _state(track_id="42")
799
800    await player._handle_intercept_tick(state, player_state, True)
801
802    # Mute call attempted; play_media never reached because mute failed
803    player.glagol.send.assert_awaited_once_with({"command": "setVolume", "volume": 0.0})
804    player.mass.player_queues.play_media.assert_not_awaited()
805    # Session not active, mute flag not set, debounce still recorded
806    assert player._intercept_active is False
807    assert player._station_muted_by_intercept is False
808
809
810async def test_alice_unmute_keeps_flag_when_send_errors() -> None:
811    """
812    Alice activates and our setVolume(saved/100) returns {"error": ...}.
813
814    The flag must NOT flip — an inconsistent flag would prevent the
815    edge-IDLE re-mute branch from firing later (because it gates on
816    `not _station_muted_by_intercept`).
817    """
818    player = _make_intercept_player()
819    player._intercept_active = True
820    player._saved_station_volume = 70
821    player._station_muted_by_intercept = True
822    player._last_intercepted_track_id = None  # avoid playing=False session-end branch
823    player.glagol.send = AsyncMock(return_value={"error": "not_connected"})
824    state, player_state, _ = _state(track_id="X", alice_state="LISTENING")
825
826    await player._handle_intercept_tick(state, player_state, True)
827
828    player.glagol.send.assert_awaited_once_with({"command": "setVolume", "volume": 0.7})
829    # Flag preserved — Station is still (presumably) muted; next attempt
830    # to unmute can happen on the next alice tick.
831    assert player._station_muted_by_intercept is True
832
833
834async def test_alice_remute_keeps_flag_when_send_errors() -> None:
835    """Edge LISTENING/SPEAKING → IDLE: re-mute send fails → flag stays False."""
836    player = _make_intercept_player()
837    player._intercept_active = True
838    player._saved_station_volume = 70
839    player._station_muted_by_intercept = False
840    player._last_intercepted_track_id = None  # avoid playing=False session-end branch
841    player.glagol.send = AsyncMock(return_value={"error": "timeout"})
842    state, player_state, _ = _state(track_id="X", alice_state="IDLE", playing=False)
843
844    await player._handle_intercept_tick(state, player_state, False, prev_alice_state="SPEAKING")
845
846    player.glagol.send.assert_awaited_once_with({"command": "setVolume", "volume": 0.0})
847    # Re-mute didn't actually land → flag stays False so the next IDLE-edge
848    # tick can attempt it again (the prev_alice_state parameter would no
849    # longer be LISTENING/SPEAKING, but a future alice activation will reset
850    # the cycle).
851    assert player._station_muted_by_intercept is False
852
853
854async def test_restore_station_volume_logs_warning_on_send_error(
855    caplog: pytest.LogCaptureFixture,
856) -> None:
857    """
858    Volume-restore transport error must log at WARNING for operator visibility.
859
860    A stuck-muted Station is user-visible, so we surface the failure loudly
861    (not DEBUG).  Validates that _restore_station_volume catches the error
862    envelope from glagol.send and emits a WARNING-level log line.
863    """
864    player = _make_intercept_player()
865    player.glagol.send = AsyncMock(return_value={"error": "not_connected"})
866
867    with caplog.at_level(logging.WARNING):
868        await player._restore_station_volume(50)
869
870    # The send was attempted; the transport error surfaced in a WARNING.
871    player.glagol.send.assert_awaited_once_with({"command": "setVolume", "volume": 0.5})
872    assert any("failed to restore Station volume" in r.message for r in caplog.records)
873
874
875# ── Real entrypoint ──────────────────────────────────────────────────
876
877
878async def test_on_glagol_update_dispatches_intercept_tick_via_create_task() -> None:
879    """
880    _on_glagol_update must hand intercept work off through mass.create_task.
881
882    This covers the integration boundary the other tests bypass by calling
883    _handle_intercept_tick directly.
884    """
885    player = _make_intercept_player()
886    # Stub out _update_playback_state and friends so we only observe the
887    # intercept dispatch.  ``update_state`` / ``set_current_media`` are
888    # declared @final on Player, so use setattr() to dodge mypy's [misc]
889    # error in upstream's strict-mode CI.
890    player._update_playback_state = MagicMock()
891    setattr(player, "update_state", MagicMock())  # noqa: B010
892    setattr(player, "set_current_media", MagicMock())  # noqa: B010
893    player._attr_available = False
894    player._attr_powered = True
895    player._attr_volume_level = 0
896    player._attr_playback_state = PlaybackState.IDLE
897    player._attr_elapsed_time = 0
898    player._attr_elapsed_time_last_updated = 0.0
899    player._attr_current_media = None
900    player._prev_alice_state = ""
901    player._voice_resume_task = None
902    player._voice_control_enabled_cache = False  # not the real attr but harmless
903
904    captured: list[Any] = []
905    player.mass.create_task = MagicMock(side_effect=captured.append)
906
907    raw_state = {
908        "state": {
909            "playerState": {"id": "12345", "title": "X", "progress": 0, "duration": 60},
910            "playing": True,
911            "volume": 0.5,
912            "aliceState": "IDLE",
913        }
914    }
915    player._on_glagol_update(raw_state)
916
917    # At least one create_task call must be the intercept tick coroutine.
918    # Coroutine objects expose `__name__` only on Py3.8+; `cr_code.co_name` is
919    # the portable introspection point for any coroutine across Python versions.
920    coro_names = [getattr(getattr(c, "cr_code", None), "co_name", "") for c in captured]
921    assert "_handle_intercept_tick" in coro_names, coro_names
922    # Cleanup never-awaited coroutines so pytest doesn't warn.
923    for coro in captured:
924        if hasattr(coro, "close"):
925            coro.close()
926
927
928async def test_dispatcher_threads_prev_alice_state_snapshot() -> None:
929    """
930    _on_glagol_update must pass the *pre-assignment* alice state to the tick.
931
932    The dispatcher overwrites self._prev_alice_state with the current
933    aliceState before scheduling the tick coroutine.  If the tick read
934    self._prev_alice_state directly, the LISTENING/SPEAKING → IDLE edge
935    re-mute branch would be dead code in production (always reading the
936    current state).  This test pins down the snapshot mechanism: when
937    prev was SPEAKING and current is IDLE, the tick must receive
938    prev=SPEAKING via parameter — the fix for Copilot's #57 review.
939    """
940    player = _make_intercept_player()
941    player._update_playback_state = MagicMock()
942    setattr(player, "update_state", MagicMock())  # noqa: B010
943    setattr(player, "set_current_media", MagicMock())  # noqa: B010
944    player._attr_available = False
945    player._attr_powered = True
946    player._attr_volume_level = 0
947    player._attr_playback_state = PlaybackState.IDLE
948    player._attr_elapsed_time = 0
949    player._attr_elapsed_time_last_updated = 0.0
950    player._attr_current_media = None
951    player._prev_alice_state = "SPEAKING"  # pre-assignment value
952    player._voice_resume_task = None
953    player._voice_control_enabled_cache = False
954
955    captured: list[Any] = []
956    player.mass.create_task = MagicMock(side_effect=captured.append)
957
958    raw_state = {
959        "state": {
960            "playerState": {"id": "12345", "title": "X", "progress": 0, "duration": 60},
961            "playing": False,  # alice IDLE, not playing → idle-edge scenario
962            "volume": 0.5,
963            "aliceState": "IDLE",  # current
964        }
965    }
966    player._on_glagol_update(raw_state)
967
968    # The coroutine has frozen its arguments; cr_frame.f_locals exposes them.
969    intercept_coros = [
970        c
971        for c in captured
972        if getattr(getattr(c, "cr_code", None), "co_name", "") == "_handle_intercept_tick"
973    ]
974    assert intercept_coros, "intercept tick was not scheduled"
975    locals_dict = intercept_coros[0].cr_frame.f_locals
976    assert locals_dict["prev_alice_state"] == "SPEAKING", (
977        f"snapshot leaked: got {locals_dict['prev_alice_state']!r}"
978    )
979    # Field itself was overwritten with current state, as expected.
980    assert player._prev_alice_state == "IDLE"
981    for coro in captured:
982        if hasattr(coro, "close"):
983            coro.close()
984
985
986# ── Round 3: alice handling, target.available, debounce preservation ─
987
988
989async def test_alice_pause_is_idempotent_across_ticks() -> None:
990    """Alice talks for several WS ticks → only one cmd_pause to the target."""
991    player = _make_intercept_player()
992    player._intercept_active = True
993    state, player_state, _ = _state(track_id="X", alice_state="LISTENING")
994
995    await player._handle_intercept_tick(state, player_state, True)
996    await player._handle_intercept_tick(state, player_state, True)
997    await player._handle_intercept_tick(state, player_state, True)
998
999    assert player.mass.players.cmd_pause.await_count == 1
1000    # Session stays open so the next Alice track resumes it
1001    assert player._intercept_active is True
1002    assert player._alice_active_pause_sent is True
1003
1004
1005async def test_alice_pause_flag_clears_on_idle() -> None:
1006    """After alice goes IDLE, the pause-flag resets so the next interaction repauses."""
1007    player = _make_intercept_player()
1008    player._intercept_active = True
1009    player._alice_active_pause_sent = True
1010
1011    state_idle, ps_idle, _ = _state(track_id="X", alice_state="IDLE")
1012    await player._handle_intercept_tick(state_idle, ps_idle, True)
1013    assert player._alice_active_pause_sent is False
1014
1015
1016async def test_alice_voice_clears_debounce() -> None:
1017    """After voice-pause, a same-track resume must not be blocked by debounce."""
1018    player = _make_intercept_player()
1019    player._intercept_active = True
1020    player._last_intercepted_track_id = "Y"
1021    player._last_intercept_time = time.time()
1022    state, player_state, _ = _state(track_id="Y", alice_state="LISTENING")
1023
1024    await player._handle_intercept_tick(state, player_state, True)
1025
1026    assert player._last_intercepted_track_id is None
1027    assert player._last_intercept_time == 0.0
1028
1029
1030async def test_alice_active_blocks_new_handoff_in_same_tick() -> None:
1031    """A fresh playerState.id arriving alongside alice activity must not start a handoff."""
1032    player = _make_intercept_player()
1033    player._intercept_active = True
1034    # Same tick: alice listening AND a new track id appeared.
1035    state, player_state, _ = _state(track_id="NEW", alice_state="LISTENING")
1036
1037    await player._handle_intercept_tick(state, player_state, True)
1038
1039    # Target paused for alice, but no handoff started for the new track.
1040    player.mass.players.cmd_pause.assert_awaited()
1041    player.glagol.send.assert_not_awaited()
1042    player.mass.music.get_item.assert_not_awaited()
1043    player.mass.player_queues.play_media.assert_not_awaited()
1044
1045
1046async def test_target_with_available_false_is_rejected() -> None:
1047    """get_player can return an object with available=False — must not silence Station."""
1048    player = _make_intercept_player()
1049    unavailable = MagicMock()
1050    unavailable.available = False
1051    player.mass.players.get_player = MagicMock(return_value=unavailable)
1052    state, player_state, _ = _state()
1053
1054    await player._handle_intercept_tick(state, player_state, True)
1055
1056    player.glagol.send.assert_not_awaited()
1057    player.mass.player_queues.play_media.assert_not_awaited()
1058
1059
1060async def test_failed_intercept_on_new_track_preserves_debounce() -> None:
1061    """New-track failure pauses old session but keeps the new track's debounce."""
1062    player = _make_intercept_player()
1063    player._intercept_active = True
1064    player._last_intercepted_track_id = "OLD"
1065    player._last_intercept_time = 0.0  # well outside the 5s window
1066    player.mass.music.get_item = AsyncMock(side_effect=RuntimeError("nope"))
1067    state, player_state, _ = _state(track_id="NEW")
1068
1069    await player._handle_intercept_tick(state, player_state, True)
1070    # Old session ended
1071    player.mass.players.cmd_pause.assert_awaited()
1072    assert player._intercept_active is False
1073    # NEW track's debounce stamp survives so the next tick is a no-op
1074    assert player._last_intercepted_track_id == "NEW"
1075
1076    await player._handle_intercept_tick(state, player_state, True)
1077    # Still only one resolve attempt — second tick was debounced
1078    assert player.mass.music.get_item.await_count == 1
1079
1080
1081async def test_pause_target_helper_flag_combinations() -> None:
1082    """The two flags on _pause_target are independent."""
1083    player = _make_intercept_player()
1084    player._intercept_active = True
1085    player._last_intercepted_track_id = "X"
1086    player._last_intercept_time = time.time()
1087
1088    # clear_session=False, clear_debounce=True → keeps active, clears debounce
1089    await player._pause_target(clear_session=False, clear_debounce=True)
1090    assert player._intercept_active is True
1091    assert player._last_intercepted_track_id is None
1092
1093    # Re-establish state
1094    player._last_intercepted_track_id = "Y"
1095    player._last_intercept_time = time.time()
1096
1097    # clear_session=True, clear_debounce=False → clears active, keeps debounce
1098    await player._pause_target(clear_session=True, clear_debounce=False)
1099    assert player._intercept_active is False
1100    assert player._last_intercepted_track_id == "Y"
1101
1102
1103# ── Round 4: serialisation, fault tolerance, dropdown filter ──────────
1104
1105
1106async def test_concurrent_alice_ticks_send_one_pause() -> None:
1107    """
1108    Two parallel LISTENING ticks → only one cmd_pause to the target.
1109
1110    Without the tick-level lock, both tasks would see
1111    `_alice_active_pause_sent=False` before either await completes and both
1112    would issue cmd_pause.  With the lock + flag-set-before-await, the second
1113    task sees the flag set and short-circuits.
1114    """
1115    player = _make_intercept_player()
1116    player._intercept_active = True
1117    state, player_state, _ = _state(track_id="X", alice_state="LISTENING")
1118
1119    pause_started = asyncio.Event()
1120    pause_release = asyncio.Event()
1121
1122    async def slow_pause(*_args: Any, **_kwargs: Any) -> None:
1123        pause_started.set()
1124        await pause_release.wait()
1125
1126    player.mass.players.cmd_pause = AsyncMock(side_effect=slow_pause)
1127
1128    t1 = asyncio.create_task(player._handle_intercept_tick(state, player_state, True))
1129    await pause_started.wait()
1130    # T2 fires while T1 is still inside cmd_pause holding the lock.
1131    t2 = asyncio.create_task(player._handle_intercept_tick(state, player_state, True))
1132    await asyncio.sleep(0)  # let t2 try to acquire the lock
1133    pause_release.set()
1134    await asyncio.gather(t1, t2)
1135
1136    assert player.mass.players.cmd_pause.await_count == 1
1137
1138
1139async def test_pause_target_cleanup_runs_when_cmd_pause_raises() -> None:
1140    """
1141    If cmd_pause raises, the state-cleanup must still happen.
1142
1143    Otherwise _intercept_active stays stale and every later WS update retries
1144    the failing path forever.
1145    """
1146    player = _make_intercept_player()
1147    player._intercept_active = True
1148    player._last_intercepted_track_id = "X"
1149    player._last_intercept_time = time.time()
1150    player.mass.players.cmd_pause = AsyncMock(side_effect=RuntimeError("gone"))
1151
1152    await player._pause_target(clear_session=True, clear_debounce=True)
1153
1154    # Despite the raise, both flags were cleared in `finally`.
1155    assert player._intercept_active is False
1156    assert player._last_intercepted_track_id is None
1157
1158
1159async def test_target_dropdown_lists_all_players_except_self() -> None:
1160    """
1161    Every registered player except the Station itself shows in the dropdown.
1162
1163    Intercept dispatches via ``mass.player_queues.play_media(queue_id=...)``
1164    which routes through the per-player queue, so any registered player is
1165    a valid target regardless of which playback features it advertises.
1166    A feature filter here only ends up hiding legitimate targets (AirPlay /
1167    DLNA / BT bridges that don't expose ``PLAY_MEDIA`` directly).  Mirror
1168    helpers (volume / pause / seek) gracefully no-op via
1169    ``UnsupportedFeaturedException`` when the chosen target lacks them.
1170    Non-audio player types (capture-only sources, lights, displays) are
1171    the exception: they can never render a track, so they are excluded.
1172    The list is sorted by display name for predictable UX.
1173    """
1174    player = _make_intercept_player()
1175
1176    full = MagicMock()
1177    full.player_id = "full"
1178    full.display_name = "Full"
1179    full.type = PlayerType.PLAYER
1180    full.supported_features = {
1181        PlayerFeature.PLAY_MEDIA,
1182        PlayerFeature.PAUSE,
1183        PlayerFeature.VOLUME_SET,
1184        PlayerFeature.SEEK,
1185    }
1186    play_media_only = MagicMock()
1187    play_media_only.player_id = "minimal"
1188    play_media_only.display_name = "Minimal"
1189    play_media_only.type = PlayerType.PLAYER
1190    play_media_only.supported_features = {PlayerFeature.PLAY_MEDIA}
1191    no_play_media = MagicMock()
1192    no_play_media.player_id = "no_play_media"
1193    no_play_media.display_name = "No Play Media"
1194    no_play_media.type = PlayerType.PLAYER
1195    no_play_media.supported_features = {PlayerFeature.PAUSE, PlayerFeature.VOLUME_SET}
1196    self_player = MagicMock()
1197    self_player.player_id = player.player_id
1198    self_player.display_name = "Self"
1199    self_player.type = PlayerType.PLAYER
1200    self_player.supported_features = {PlayerFeature.PLAY_MEDIA}
1201    source_player = MagicMock()
1202    source_player.player_id = "turntable"
1203    source_player.display_name = "Turntable"
1204    source_player.type = PlayerType.SOURCE
1205    source_player.supported_features = set()
1206    player.mass.players.all_players = MagicMock(
1207        return_value=[full, play_media_only, no_play_media, self_player, source_player]
1208    )
1209
1210    entries = await YandexStationPlayer.get_config_entries(player)
1211    target_entry = next(e for e in entries if getattr(e, "key", None) == CONF_INTERCEPT_TARGET)
1212    listed_ids = [opt.value for opt in target_entry.options]
1213
1214    # Every non-self player appears, regardless of supported_features,
1215    # sorted alphabetically by display name (Full, Minimal, No Play Media).
1216    # The capture-only source player is excluded by type.
1217    assert listed_ids == ["full", "minimal", "no_play_media"]
1218
1219
1220async def test_concurrent_mirror_volume_serialised() -> None:
1221    """
1222    Back-to-back volume updates must be applied in order.
1223
1224    Without the tick-level lock, an older volume task could finish after a
1225    newer one and leave the target stale.  With the lock, the second tick
1226    blocks until the first finishes — guaranteeing in-order application.
1227    """
1228    player = _make_intercept_player()
1229    player._intercept_active = True
1230    player._last_intercepted_track_id = "X"
1231    player._last_intercept_time = time.time()
1232
1233    applied: list[int] = []
1234    first_started = asyncio.Event()
1235    first_release = asyncio.Event()
1236
1237    async def slow_first(*_args: Any, **kwargs: Any) -> None:  # noqa: ARG001
1238        applied.append(_args[1])
1239        first_started.set()
1240        await first_release.wait()
1241
1242    async def fast(*_args: Any, **kwargs: Any) -> None:  # noqa: ARG001
1243        applied.append(_args[1])
1244
1245    cmds = AsyncMock(side_effect=slow_first)
1246    player.mass.players.cmd_volume_set = cmds
1247
1248    state1, ps1, _ = _state(track_id="X", volume=0.3)
1249    t1 = asyncio.create_task(player._handle_intercept_tick(state1, ps1, True))
1250    await first_started.wait()
1251    cmds.side_effect = fast  # next call uses the fast handler
1252    state2, ps2, _ = _state(track_id="X", volume=0.6)
1253    t2 = asyncio.create_task(player._handle_intercept_tick(state2, ps2, True))
1254    await asyncio.sleep(0)
1255    first_release.set()
1256    await asyncio.gather(t1, t2)
1257
1258    # In-order application: 30 then 60 (not the reverse).
1259    assert applied == [30, 60]
1260