music-assistant-server

1# mypy: disable-error-code="attr-defined,unreachable,method-assign,misc,assignment,unused-ignore"
2"""Tests for the YandexYnisonProvider."""
3
4from __future__ import annotations
5
6import asyncio
7from contextlib import suppress
8from typing import Any
9from unittest.mock import AsyncMock, MagicMock, patch
10
11import pytest
12from music_assistant_models.enums import (
13    ContentType,
14    PlaybackState,
15    ProviderFeature,
16    ProviderType,
17)
18from music_assistant_models.errors import (
19    LoginFailed,
20    PlayerCommandFailed,
21    ResourceTemporarilyUnavailable,
22    UnsupportedFeaturedException,
23)
24from music_assistant_models.media_items import AudioFormat, AudioSource
25from ya_passport_auth import SecretStr
26from ya_passport_auth.ma import BorrowedCredentialSource, list_yandex_music_instances
27
28from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER, ThrottlerManager
29from music_assistant.providers.yandex_ynison.constants import (
30    CONF_ALLOW_PLAYER_SWITCH,
31    CONF_DEVICE_ID,
32    CONF_MASS_PLAYER_ID,
33    CONF_PUBLISH_NAME,
34    CONF_TOKEN,
35    CONF_X_TOKEN,
36    CONF_YM_INSTANCE,
37    DEFAULT_DISPLAY_NAME,
38    OUTPUT_AUTO,
39    PLAYER_ID_AUTO,
40    YM_INSTANCE_OWN,
41)
42from music_assistant.providers.yandex_ynison.provider import (
43    _API_MAX_RETRIES,
44    _COMMAND_IDEMPOTENCY_TTL,
45    AUDIO_SOURCE_ID,
46    YandexYnisonProvider,
47)
48from music_assistant.providers.yandex_ynison.streaming import (
49    PCM_LOSSLESS_PARAMS,
50    PCM_LOSSY_PARAMS,
51    make_pcm_format,
52)
53from music_assistant.providers.yandex_ynison.ynison_client import YnisonSendError, YnisonState
54
55
56def _arm_play_media_recorder(provider: YandexYnisonProvider) -> list[tuple[str, str]]:
57    """
58    Replace `play_media` with a recorder and run `create_task` coros inline.
59
60    Returns the list of (target_id, uri) tuples captured during the test.
61    The inline create_task lets the scheduled `play_media` coroutine
62    actually execute against the recorder.
63    """
64    calls: list[tuple[str, str]] = []
65
66    async def _record(target_id: str, uri: str) -> None:
67        calls.append((target_id, uri))
68
69    provider.mass.player_queues.play_media = _record
70    provider.mass.create_task = MagicMock(
71        side_effect=lambda coro, *_a, **_kw: asyncio.get_event_loop().create_task(coro)
72    )
73    return calls
74
75
76def _stub_attr(obj: object, name: str, value: Any) -> None:
77    """Setattr that bypasses mypy method-assign and ruff B010."""
78    setattr(obj, name, value)
79
80
81def _make_mock_config(values: dict[str, Any] | None = None) -> MagicMock:
82    """Create a mock ProviderConfig."""
83    defaults: dict[str, Any] = {
84        CONF_TOKEN: "test-music-token",
85        CONF_YM_INSTANCE: YM_INSTANCE_OWN,
86        CONF_MASS_PLAYER_ID: PLAYER_ID_AUTO,
87        CONF_ALLOW_PLAYER_SWITCH: True,
88        CONF_PUBLISH_NAME: DEFAULT_DISPLAY_NAME,
89        CONF_DEVICE_ID: "test-device-uuid",
90        "log_level": "GLOBAL",
91    }
92    if values:
93        defaults.update(values)
94    config = MagicMock()
95    config.get_value.side_effect = defaults.get
96    # Provider.__init__ now caches the AudioSource which serialises name into a
97    # uri/sort_name — both expect real strings, not MagicMock attribute access.
98    config.instance_id = "yandex_ynison_test"
99    config.name = "Yandex Music Connect"
100    return config
101
102
103def _make_mock_mass() -> MagicMock:
104    """Create a mock MusicAssistant instance."""
105    mass = MagicMock()
106    mass.cache_path = "/var/cache/test-cache"
107
108    def _create_task(coro: object) -> MagicMock:
109        if asyncio.iscoroutine(coro):
110            coro.close()  # prevent RuntimeWarning for unawaited coroutine
111        return MagicMock()
112
113    mass.create_task = MagicMock(side_effect=_create_task)
114    mass.subscribe = MagicMock(return_value=MagicMock())
115    mass.get_providers = MagicMock(return_value=[])
116    mass.config.set_raw_provider_config_value = MagicMock()
117    # Auth values now live in setup_data; the provider reads them via
118    # get_setup_value. Empty setup_data routes those reads through to
119    # config.get_value (via get_config_value; the seeded stub above).
120    mass.config.get = MagicMock(return_value={})
121    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
122
123    # Cache — return None (miss) by default
124    mass.cache.get = AsyncMock(return_value=None)
125    mass.cache.set = AsyncMock()
126    mass.cache.delete = AsyncMock()
127
128    # Players
129    mass.players.all_players = MagicMock(return_value=[])
130    mass.players.get_player = MagicMock(return_value=None)
131    mass.players.cmd_stop = AsyncMock()
132    mass.players.cmd_pause = AsyncMock()
133    mass.players.cmd_play = AsyncMock()
134    mass.players.cmd_volume_set = AsyncMock()
135    mass.players.trigger_player_update = MagicMock()
136
137    # Player queues
138    mass.player_queues.play_media = AsyncMock()
139    mass.player_queues.pause = AsyncMock()
140    mass.player_queues.play = AsyncMock()
141
142    # Streams — live metadata updates flow through update_stream_metadata
143    mass.streams.update_stream_metadata = MagicMock()
144
145    return mass
146
147
148def _make_mock_manifest() -> MagicMock:
149    """Create a mock ProviderManifest."""
150    manifest = MagicMock()
151    manifest.domain = "yandex_ynison"
152    return manifest
153
154
155def _make_provider(player_id: str = PLAYER_ID_AUTO) -> YandexYnisonProvider:
156    """Create a YandexYnisonProvider with mock dependencies."""
157    mass = _make_mock_mass()
158    config = _make_mock_config({CONF_MASS_PLAYER_ID: player_id})
159    manifest = _make_mock_manifest()
160    provider = YandexYnisonProvider(mass, manifest, config, {ProviderFeature.AUDIO_SOURCE})
161    provider._api_throttler = ThrottlerManager(rate_limit=1000)
162    return provider
163
164
165# ------------------------------------------------------------------
166# Provider init
167# ------------------------------------------------------------------
168
169
170class TestProviderInit:
171    """Tests for provider initialization."""
172
173    def test_audio_source_details(self) -> None:
174        """AudioSource should be configured correctly."""
175        provider = _make_provider()
176
177        source = provider._audio_source
178        # provider_mapping carries the audio_format in the new model
179        mapping = next(iter(source.provider_mappings))
180        assert mapping.audio_format.content_type == ContentType.PCM_S16LE
181        assert mapping.audio_format.sample_rate == 44100
182        assert mapping.audio_format.bit_depth == 16
183        assert mapping.audio_format.channels == 2
184        # capabilities default off until a matching Yandex Music provider links
185        assert source.can_play_pause is False
186        assert source.can_seek is False
187        assert source.can_next_previous is False
188        assert source.exclusive is True
189
190    def test_device_id_persisted(self) -> None:
191        """When no device_id in config, should generate and persist."""
192        mass = _make_mock_mass()
193        config = _make_mock_config({CONF_DEVICE_ID: None})
194        manifest = _make_mock_manifest()
195
196        provider = YandexYnisonProvider(mass, manifest, config, {ProviderFeature.AUDIO_SOURCE})
197
198        # Should have generated a device ID and saved it
199        mass.config.set_raw_provider_config_value.assert_called()
200        assert provider._device_id  # non-empty
201
202    def test_existing_device_id_used(self) -> None:
203        """When device_id exists in config, should use it."""
204        mass = _make_mock_mass()
205        config = _make_mock_config({CONF_DEVICE_ID: "existing-uuid"})
206        manifest = _make_mock_manifest()
207
208        provider = YandexYnisonProvider(mass, manifest, config, {ProviderFeature.AUDIO_SOURCE})
209
210        assert provider._device_id == "existing-uuid"
211
212
213# ------------------------------------------------------------------
214# Player selection
215# ------------------------------------------------------------------
216
217
218class TestPlayerSelection:
219    """Tests for _get_target_player_id."""
220
221    def test_auto_no_players(self) -> None:
222        """Auto mode returns None when no players available."""
223        provider = _make_provider()
224        assert provider._get_target_player_id() is None
225
226    def test_auto_with_playing_player(self) -> None:
227        """Auto mode selects the currently playing player."""
228        provider = _make_provider()
229
230        player1 = MagicMock()
231        player1.player_id = "player1"
232        player1.display_name = "Player 1"
233        player1.state.playback_state = PlaybackState.IDLE
234
235        player2 = MagicMock()
236        player2.player_id = "player2"
237        player2.display_name = "Player 2"
238        player2.state.playback_state = PlaybackState.PLAYING
239
240        provider.mass.players.all_players.return_value = [player1, player2]  # type: ignore[attr-defined]
241
242        assert provider._get_target_player_id() == "player2"
243
244    def test_specific_player_exists(self) -> None:
245        """Returns configured player when it exists."""
246        provider = _make_provider("my-player")
247        provider.mass.players.get_player.return_value = MagicMock()  # type: ignore[attr-defined]
248
249        assert provider._get_target_player_id() == "my-player"
250
251    def test_specific_player_missing(self) -> None:
252        """Returns None when configured player no longer exists."""
253        provider = _make_provider("gone-player")
254        provider.mass.players.get_player.return_value = None  # type: ignore[attr-defined]
255
256        assert provider._get_target_player_id() is None
257
258    def test_active_player_takes_priority(self) -> None:
259        """Active player takes priority over auto selection."""
260        provider = _make_provider()
261        provider._active_player_id = "active-one"
262        provider.mass.players.get_player.return_value = MagicMock()  # type: ignore[attr-defined]
263
264        assert provider._get_target_player_id() == "active-one"
265
266
267# ------------------------------------------------------------------
268# Source selection
269# ------------------------------------------------------------------
270
271
272class TestSourceSelection:
273    """Tests for on_source_selected (the new PluginProvider hook)."""
274
275    async def test_on_source_selected_sets_active(self) -> None:
276        """Selecting source sets the active player and records the session id."""
277        provider = _make_provider()
278
279        await provider.on_source_selected("main", "new-player", "new-player", "session_1")
280        assert provider._active_player_id == "new-player"
281        assert provider._active_session_id == "session_1"
282
283    async def test_on_source_selected_switching_disabled(self) -> None:
284        """Rejects source selection when player switching is disabled."""
285        mass = _make_mock_mass()
286        config = _make_mock_config({CONF_ALLOW_PLAYER_SWITCH: False})
287        manifest = _make_mock_manifest()
288        provider = YandexYnisonProvider(mass, manifest, config, {ProviderFeature.AUDIO_SOURCE})
289
290        # Set default player
291        provider._default_player_id = "default-player"
292        mass.players.get_player.return_value = MagicMock()
293
294        with pytest.raises(RuntimeError, match="Player switching is disabled"):
295            await provider.on_source_selected("main", "other-player", "other-player", "session_1")
296
297        # Should have redirected to the configured default via play_media
298        mass.player_queues.play_media.assert_awaited()
299        assert provider._active_player_id is None
300
301    async def test_on_source_selected_disabled_redirect_not_repeated(self) -> None:
302        """
303        Repeated rejected selections must not re-issue the redirect play_media.
304
305        When player switching is disabled and a non-target player keeps having
306        the source selected (sendspin bridge / sync-group indirection re-triggers
307        the stream under a player id that never equals the configured target),
308        the redirect ``play_media`` must fire at most once per idempotency window.
309        Otherwise every rejection re-issues the redirect, which re-triggers
310        selection, producing an unbounded ``AudioError`` storm.
311        """
312        mass = _make_mock_mass()
313        config = _make_mock_config({CONF_ALLOW_PLAYER_SWITCH: False})
314        manifest = _make_mock_manifest()
315        provider = YandexYnisonProvider(mass, manifest, config, {ProviderFeature.AUDIO_SOURCE})
316
317        provider._default_player_id = "default-player"
318        mass.players.get_player.return_value = MagicMock()
319
320        for _ in range(3):
321            with pytest.raises(RuntimeError, match="Player switching is disabled"):
322                await provider.on_source_selected(
323                    "main", "other-player", "other-player", "session_1"
324                )
325
326        # Three rejected selections, but the redirect fired only once.
327        assert mass.player_queues.play_media.await_count == 1
328
329
330# ------------------------------------------------------------------
331# Clear active player
332# ------------------------------------------------------------------
333
334
335class TestClearActivePlayer:
336    """Tests for _clear_active_player."""
337
338    def test_clears_state(self) -> None:
339        """Clearing active player resets state and triggers update."""
340        provider = _make_provider()
341
342        provider._active_player_id = "some-player"
343        provider._in_use_by_queue = "some-player"
344
345        provider._clear_active_player()
346
347        assert provider._active_player_id is None
348        assert provider._in_use_by_queue is None  # type: ignore[unreachable]
349        provider.mass.players.trigger_player_update.assert_called_with("some-player")
350
351
352# ------------------------------------------------------------------
353# Provider matching
354# ------------------------------------------------------------------
355
356
357class TestProviderMatching:
358    """Tests for _check_yandex_provider_match."""
359
360    async def test_finds_yandex_music_provider(self) -> None:
361        """Links to Yandex Music provider and enables playback control."""
362        provider = _make_provider()
363
364        mock_ym = MagicMock()
365        mock_ym.domain = "yandex_music"
366        mock_ym.type = ProviderType.MUSIC
367        provider.mass.get_providers.return_value = [mock_ym]  # type: ignore[attr-defined]
368
369        await provider._check_yandex_provider_match()
370
371        assert provider._yandex_provider is mock_ym
372        # Capability flags rebuilt on the AudioSource when a matching provider links
373        assert provider._audio_source.can_play_pause is True
374        assert provider._audio_source.can_seek is True
375        assert provider._audio_source.can_next_previous is True
376
377    async def test_no_matching_provider(self) -> None:
378        """No linked provider disables playback control."""
379        provider = _make_provider()
380
381        provider.mass.get_providers.return_value = []  # type: ignore[attr-defined]
382        await provider._check_yandex_provider_match()
383
384        assert provider._yandex_provider is None
385        assert provider._audio_source.can_play_pause is False
386
387
388# ------------------------------------------------------------------
389# Ynison state handling
390# ------------------------------------------------------------------
391
392
393class TestYnisonStateHandling:
394    """Tests for _handle_ynison_state."""
395
396    async def test_activates_on_our_device(self) -> None:
397        """Activates playback when Ynison reports our device as active."""
398        provider = _make_provider()
399
400        # Setup a target player
401        player = MagicMock()
402        player.player_id = "player1"
403        player.display_name = "Player 1"
404        provider.mass.players.all_players.return_value = [player]  # type: ignore[attr-defined]
405        provider.mass.players.get_player.return_value = player  # type: ignore[attr-defined]
406
407        state = YnisonState(
408            active_device_id=provider._device_id,
409            player_state={
410                "status": {"paused": False, "progress_ms": 5000, "duration_ms": 200000},
411                "player_queue": {
412                    "current_playable_index": 0,
413                    "playable_list": [{"playable_id": "track1"}],
414                },
415            },
416        )
417
418        await provider._handle_ynison_state(state)
419
420        assert provider._active_player_id == "player1"
421
422    async def test_clears_on_device_switch(self) -> None:
423        """Clears active player when device switches away."""
424        provider = _make_provider()
425
426        provider._active_player_id = "player1"
427        provider._in_use_by_queue = "player1"
428
429        state = YnisonState(active_device_id="other-device-id")
430        await provider._handle_ynison_state(state)
431
432        assert provider._active_player_id is None
433        assert provider._in_use_by_queue is None  # type: ignore[unreachable]
434
435    async def test_seek_detected_from_ynison(self) -> None:
436        """Detects seek from Yandex app via progress drift."""
437        provider = _make_provider()
438
439        player = MagicMock()
440        player.player_id = "player1"
441        provider.mass.players.all_players.return_value = [player]  # type: ignore[attr-defined]
442        provider.mass.players.get_player.return_value = player  # type: ignore[attr-defined]
443
444        def _make_state(progress_ms: int) -> YnisonState:
445            return YnisonState(
446                active_device_id=provider._device_id,
447                player_state={
448                    "status": {
449                        "paused": False,
450                        "progress_ms": progress_ms,
451                        "duration_ms": 200000,
452                    },
453                    "player_queue": {
454                        "current_playable_index": 0,
455                        "playable_list": [{"playable_id": "track1"}],
456                    },
457                },
458            )
459
460        # First state — track starts at 0ms
461        await provider._handle_ynison_state(_make_state(0))
462        assert provider._current_streaming_track_id == "track1"  # set eagerly on detection
463
464        # Expire the grace period so the seek detection isn't suppressed
465        provider._seek_grace_until = 0.0
466
467        # Second state — seek to 60s (drift 60000ms > 2000ms)
468        await provider._handle_ynison_state(_make_state(60000))
469        assert provider._seek_position_ms == 60000
470        assert provider._track_changed_event.is_set()
471
472        # Verify force_update=True was used so the server sends a full
473        # PLAYER_UPDATED event (not just a lightweight elapsed-time one)
474        provider.mass.players.trigger_player_update.assert_called_with("player1", force_update=True)  # type: ignore[attr-defined]
475
476    async def test_seek_grace_period_after_track_change(self) -> None:
477        """Seek detection is suppressed during grace period after track change."""
478        provider = _make_provider()
479
480        player = MagicMock()
481        player.player_id = "player1"
482        provider.mass.players.all_players.return_value = [player]  # type: ignore[attr-defined]
483        provider.mass.players.get_player.return_value = player  # type: ignore[attr-defined]
484
485        def _make_state(progress_ms: int) -> YnisonState:
486            return YnisonState(
487                active_device_id=provider._device_id,
488                player_state={
489                    "status": {
490                        "paused": False,
491                        "progress_ms": progress_ms,
492                        "duration_ms": 200000,
493                    },
494                    "player_queue": {
495                        "current_playable_index": 0,
496                        "playable_list": [{"playable_id": "track1"}],
497                    },
498                },
499            )
500
501        # Track starts — sets grace period
502        await provider._handle_ynison_state(_make_state(0))
503        assert provider._seek_grace_until > 0
504
505        # Echo with progress=0 arrives during grace period — should NOT
506        # trigger seek even though drift calculation would exceed threshold
507        provider._track_changed_event.clear()
508        await provider._handle_ynison_state(_make_state(0))
509        assert provider._seek_position_ms == 0  # unchanged
510        assert not provider._track_changed_event.is_set()  # no false seek
511
512    async def test_progress_throttled_update(self) -> None:
513        """Regular progress updates trigger player update with throttling."""
514        provider = _make_provider()
515
516        player = MagicMock()
517        player.player_id = "player1"
518        provider.mass.players.all_players.return_value = [player]  # type: ignore[attr-defined]
519        provider.mass.players.get_player.return_value = player  # type: ignore[attr-defined]
520
521        state = YnisonState(
522            active_device_id=provider._device_id,
523            player_state={
524                "status": {
525                    "paused": False,
526                    "progress_ms": 5000,
527                    "duration_ms": 200000,
528                },
529                "player_queue": {
530                    "current_playable_index": 0,
531                    "playable_list": [{"playable_id": "track1"}],
532                },
533            },
534        )
535
536        # First call — significant (new track) → always triggers
537        await provider._handle_ynison_state(state)
538        call_count_1 = provider.mass.players.trigger_player_update.call_count  # type: ignore[attr-defined]
539
540        # Simulate same track still playing (no seek, no track change).
541        # Mark as echo so the seek-detection branch stays quiet.
542        state2 = YnisonState(
543            active_device_id=provider._device_id,
544            player_state={
545                "status": {
546                    "paused": False,
547                    "progress_ms": 6000,
548                    "duration_ms": 200000,
549                },
550                "player_queue": {
551                    "current_playable_index": 0,
552                    "playable_list": [{"playable_id": "track1"}],
553                },
554            },
555            last_update_is_echo=True,
556        )
557
558        # Second call shortly after — throttled, no trigger
559        await provider._handle_ynison_state(state2)
560        call_count_2 = provider.mass.players.trigger_player_update.call_count  # type: ignore[attr-defined]
561
562        # Force the throttle to expire
563        provider._last_player_update_time = 0.0
564        await provider._handle_ynison_state(state2)
565        call_count_3 = provider.mass.players.trigger_player_update.call_count  # type: ignore[attr-defined]
566
567        # First call triggered, second was throttled, third triggered
568        assert call_count_1 >= 1
569        assert call_count_2 == call_count_1
570        assert call_count_3 > call_count_2
571
572        # Regular (non-seek) updates should NOT use force_update
573        provider.mass.players.trigger_player_update.assert_called_with(  # type: ignore[attr-defined]
574            "player1", force_update=False
575        )
576
577    async def test_duration_updated_from_stream_details(self) -> None:
578        """Duration is updated from stream_details and pushed to Ynison."""
579        provider = _make_provider()
580        # trigger_player_update needs the actual player id (bridge), not the
581        # queue id — bridge players (`spb_*`) wrap the bare ALSA UUID and the
582        # MA UI's state machine lives on the bridge.
583        provider._active_player_id = "spb_bridge1"
584        provider._in_use_by_queue = "player1"
585        mock_ynison = MagicMock()
586        mock_ynison.update_playing_status = AsyncMock()
587        mock_ynison.state.is_paused = False
588        provider._ynison = mock_ynison
589
590        stream_details = MagicMock()
591        stream_details.duration = 185  # seconds
592
593        await provider._update_metadata_from_stream(stream_details, seek_ms=30000)
594
595        # live track-change info lives on _stream_metadata (pushed through
596        # streamdetails.stream_metadata), not on the AudioSource MediaItem
597        meta = provider._stream_metadata
598        assert meta.duration == 185
599        assert meta.elapsed_time == 30  # 30000ms → 30s
600        assert provider._actual_duration_ms == 185000
601        provider.mass.players.trigger_player_update.assert_called_once_with(  # type: ignore[attr-defined]
602            "spb_bridge1", force_update=True
603        )
604        # Real duration pushed to Ynison (heartbeat — no `strict`).
605        mock_ynison.update_playing_status.assert_awaited_once_with(
606            progress_ms=30000, duration_ms=185000, paused=False, strict=False
607        )
608
609    async def test_signal_track_completion_advances_index(self) -> None:
610        """Track completion advances index and reports status."""
611        provider = _make_provider()
612        mock_ynison = MagicMock()
613        mock_ynison.state = YnisonState(
614            active_device_id=provider._device_id,
615            player_state={
616                "status": {"paused": False, "progress_ms": 180000, "duration_ms": 200000},
617                "player_queue": {
618                    "current_playable_index": 0,
619                    "playable_list": [{"playable_id": "t1"}, {"playable_id": "t2"}],
620                    "entity_id": "playlist:123",
621                    "entity_type": "PLAYLIST",
622                },
623            },
624        )
625        mock_ynison.update_playing_status = AsyncMock()
626        mock_ynison.update_player_state = AsyncMock()
627        provider._ynison = mock_ynison
628
629        await provider._signal_track_completion()
630
631        # 1. Reports progress=duration (`strict=True` — end-of-track signal).
632        mock_ynison.update_playing_status.assert_awaited_once_with(
633            progress_ms=200000, duration_ms=200000, paused=False, strict=True
634        )
635        # 2. Advances current_playable_index by 1
636        call_args = mock_ynison.update_player_state.call_args
637        sent_state = call_args.kwargs["player_state"]
638        assert sent_state["player_queue"]["current_playable_index"] == 1
639        assert sent_state["status"]["progress_ms"] == "0"
640        assert sent_state["status"]["paused"] is False
641        # Resets actual duration for next track
642        assert provider._actual_duration_ms == 0
643
644    async def test_signal_track_completion_no_send_full_state(self) -> None:
645        """Track completion never sends full state reset."""
646        provider = _make_provider()
647        mock_ynison = MagicMock()
648        mock_ynison.state = YnisonState(
649            active_device_id=provider._device_id,
650            player_state={
651                "status": {"paused": False, "progress_ms": 180000, "duration_ms": 200000},
652                "player_queue": {
653                    "current_playable_index": 0,
654                    "playable_list": [{"playable_id": "t1"}, {"playable_id": "t2"}],
655                    "entity_id": "playlist:123",
656                },
657            },
658        )
659        mock_ynison.update_playing_status = AsyncMock()
660        mock_ynison.update_player_state = AsyncMock()
661        mock_ynison.send_full_state = AsyncMock()
662        provider._ynison = mock_ynison
663
664        await provider._signal_track_completion()
665
666        # Must NOT send full state reset
667        mock_ynison.send_full_state.assert_not_called()
668
669    async def test_signal_track_completion_uses_actual_duration(self) -> None:
670        """Track completion prefers _actual_duration_ms over stale state.duration_ms."""
671        provider = _make_provider()
672        provider._actual_duration_ms = 300000
673        mock_ynison = MagicMock()
674        mock_ynison.state = YnisonState(
675            active_device_id=provider._device_id,
676            player_state={
677                "status": {"paused": False, "progress_ms": 180000, "duration_ms": 200000},
678                "player_queue": {
679                    "current_playable_index": 0,
680                    "playable_list": [{"playable_id": "t1"}, {"playable_id": "t2"}],
681                },
682            },
683        )
684        mock_ynison.update_playing_status = AsyncMock()
685        mock_ynison.update_player_state = AsyncMock()
686        provider._ynison = mock_ynison
687
688        await provider._signal_track_completion()
689
690        mock_ynison.update_playing_status.assert_awaited_once_with(
691            progress_ms=300000, duration_ms=300000, paused=False, strict=True
692        )
693
694    async def test_signal_track_completion_radio_replenishes_queue(self) -> None:
695        """At end of RADIO queue, fetches more tracks via YM API and advances."""
696        provider = _make_provider()
697        mock_ynison = MagicMock()
698        mock_ynison.state = YnisonState(
699            active_device_id=provider._device_id,
700            player_state={
701                "status": {"paused": False, "progress_ms": 200000, "duration_ms": 215000},
702                "player_queue": {
703                    "current_playable_index": 1,
704                    "playable_list": [
705                        {"playable_id": "t1", "from": "radio-src"},
706                        {"playable_id": "t2", "from": "radio-src"},
707                    ],
708                    "entity_id": "user:onyourwave",
709                    "entity_type": "RADIO",
710                },
711            },
712        )
713        mock_ynison.update_playing_status = AsyncMock()
714        mock_ynison.update_player_state = AsyncMock()
715        provider._ynison = mock_ynison
716
717        # Mock YM provider returning new tracks
718        mock_track = MagicMock()
719        mock_track.id = "t3"
720        mock_track.title = "New Track"
721        mock_track.albums = [MagicMock(id="a3")]
722        mock_track.cover_uri = "cover3.jpg"
723
724        mock_ym_provider = MagicMock()
725        mock_ym_provider.get_rotor_station_tracks = AsyncMock(
726            return_value=([mock_track], "batch-123")
727        )
728        provider._yandex_provider = mock_ym_provider
729
730        await provider._signal_track_completion()
731
732        # Fetched tracks from station
733        mock_ym_provider.get_rotor_station_tracks.assert_awaited_once_with(
734            "user:onyourwave", queue="t2"
735        )
736        # Advanced index to 2 with expanded playable_list
737        call_args = mock_ynison.update_player_state.call_args
738        sent_state = call_args.kwargs["player_state"]
739        assert sent_state["player_queue"]["current_playable_index"] == 2
740        expanded = sent_state["player_queue"]["playable_list"]
741        assert len(expanded) == 3
742        assert expanded[2]["playable_id"] == "t3"
743        assert expanded[2]["title"] == "New Track"
744        assert expanded[2]["from"] == "radio-src"
745
746    async def test_signal_track_completion_radio_no_provider(self) -> None:
747        """At end of queue without YM provider, does not crash."""
748        provider = _make_provider()
749        mock_ynison = MagicMock()
750        mock_ynison.state = YnisonState(
751            active_device_id=provider._device_id,
752            player_state={
753                "status": {"paused": False, "progress_ms": 200000, "duration_ms": 215000},
754                "player_queue": {
755                    "current_playable_index": 1,
756                    "playable_list": [
757                        {"playable_id": "t1"},
758                        {"playable_id": "t2"},
759                    ],
760                    "entity_id": "user:onyourwave",
761                    "entity_type": "RADIO",
762                },
763            },
764        )
765        mock_ynison.update_playing_status = AsyncMock()
766        mock_ynison.update_player_state = AsyncMock()
767        provider._ynison = mock_ynison
768        provider._yandex_provider = None
769
770        await provider._signal_track_completion()
771
772        # Status reported
773        mock_ynison.update_playing_status.assert_awaited_once()
774        # Cannot advance — no provider to fetch tracks
775        mock_ynison.update_player_state.assert_not_called()
776
777    async def test_prefetch_on_second_to_last_track(self) -> None:
778        """Pre-fetches tracks when playing second-to-last item in queue."""
779        provider = _make_provider()
780        mock_ynison = MagicMock()
781        mock_ynison.connected = True
782        mock_ynison.update_player_state = AsyncMock()
783        # 4 tracks, currently at index 2 (second-to-last)
784        mock_ynison.state = YnisonState(
785            active_device_id=provider._device_id,
786            player_state={
787                "status": {"paused": False, "progress_ms": 10000, "duration_ms": 200000},
788                "player_queue": {
789                    "current_playable_index": 2,
790                    "playable_list": [
791                        {"playable_id": "t1", "from": "src"},
792                        {"playable_id": "t2", "from": "src"},
793                        {"playable_id": "t3", "from": "src"},
794                        {"playable_id": "t4", "from": "src"},
795                    ],
796                    "entity_id": "user:onyourwave",
797                    "entity_type": "RADIO",
798                },
799            },
800        )
801        provider._ynison = mock_ynison
802
803        mock_track = MagicMock()
804        mock_track.id = "t5"
805        mock_track.title = "Prefetched"
806        mock_track.albums = [MagicMock(id="a5")]
807        mock_track.cover_uri = "cover5.jpg"
808
809        mock_ym_provider = MagicMock()
810        mock_ym_provider.get_rotor_station_tracks = AsyncMock(
811            return_value=([mock_track], "batch-pfx")
812        )
813        provider._yandex_provider = mock_ym_provider
814
815        # Use real create_task so prefetch coroutine actually runs
816        provider.mass.create_task = lambda coro: asyncio.get_event_loop().create_task(coro)  # type: ignore[method-assign, assignment, misc]
817
818        # Trigger prefetch
819        provider._maybe_prefetch(
820            2,
821            mock_ynison.state.player_state["player_queue"]["playable_list"],
822            "user:onyourwave",
823            "RADIO",
824        )
825        assert provider._prefetch_task is not None
826        await provider._prefetch_task
827
828        # Prefetched list should contain old + new
829        assert provider._prefetched_list is not None
830        assert len(provider._prefetched_list) == 5
831        assert provider._prefetched_list[4]["playable_id"] == "t5"
832
833    async def test_signal_completion_uses_prefetched(self) -> None:
834        """Track completion uses pre-fetched data instead of making API call."""
835        provider = _make_provider()
836        mock_ynison = MagicMock()
837        mock_ynison.state = YnisonState(
838            active_device_id=provider._device_id,
839            player_state={
840                "status": {"paused": False, "progress_ms": 200000, "duration_ms": 215000},
841                "player_queue": {
842                    "current_playable_index": 3,
843                    "playable_list": [
844                        {"playable_id": "t1"},
845                        {"playable_id": "t2"},
846                        {"playable_id": "t3"},
847                        {"playable_id": "t4"},
848                    ],
849                    "entity_id": "user:onyourwave",
850                    "entity_type": "RADIO",
851                },
852            },
853        )
854        mock_ynison.update_playing_status = AsyncMock()
855        mock_ynison.update_player_state = AsyncMock()
856        provider._ynison = mock_ynison
857
858        # Simulate pre-fetched data
859        prefetched = [
860            {"playable_id": "t1"},
861            {"playable_id": "t2"},
862            {"playable_id": "t3"},
863            {"playable_id": "t4"},
864            {"playable_id": "t5"},
865        ]
866        provider._prefetched_list = prefetched
867
868        mock_ym_provider = MagicMock()
869        mock_ym_provider.get_rotor_station_tracks = AsyncMock()
870        provider._yandex_provider = mock_ym_provider
871
872        await provider._signal_track_completion()
873
874        # Should NOT have called API — used prefetched
875        mock_ym_provider.get_rotor_station_tracks.assert_not_awaited()
876        # Advanced with prefetched list
877        call_args = mock_ynison.update_player_state.call_args
878        sent_state = call_args.kwargs["player_state"]
879        assert sent_state["player_queue"]["current_playable_index"] == 4
880        assert len(sent_state["player_queue"]["playable_list"]) == 5
881        # Prefetch consumed
882        assert provider._prefetched_list is None
883
884    async def test_best_duration_prefers_actual(self) -> None:
885        """_best_duration_ms prefers _actual_duration_ms over state.duration_ms."""
886        provider = _make_provider()
887        mock_ynison = MagicMock()
888        mock_ynison.state = YnisonState(
889            active_device_id=provider._device_id,
890            player_state={
891                "status": {"duration_ms": 200000},
892            },
893        )
894        provider._ynison = mock_ynison
895
896        # Fallback to state when actual is 0
897        assert provider._best_duration_ms() == 200000
898
899        # Prefer actual when set
900        provider._actual_duration_ms = 300000
901        assert provider._best_duration_ms() == 300000
902
903        # Without ynison, only actual
904        provider._ynison = None
905        assert provider._best_duration_ms() == 300000
906        provider._actual_duration_ms = 0
907        assert provider._best_duration_ms() == 0
908
909    async def test_wait_for_track_change_ignores_echo(self) -> None:
910        """_wait_for_track_change should ignore echoes and wait for actual change."""
911        provider = _make_provider()
912        mock_ynison = MagicMock()
913        mock_ynison.state = YnisonState(
914            active_device_id=provider._device_id,
915            player_state={
916                "status": {"progress_ms": 248000, "duration_ms": 248000},
917                "player_queue": {
918                    "current_playable_index": 0,
919                    "playable_list": [{"playable_id": "old_track"}],
920                    "entity_id": "user:onyourwave",
921                    "entity_type": "RADIO",
922                },
923            },
924        )
925        provider._ynison = mock_ynison
926
927        async def simulate_echo_then_change() -> None:
928            await asyncio.sleep(0.01)
929            # First event: echo with same track (should be ignored)
930            provider._track_changed_event.set()
931            await asyncio.sleep(0.01)
932            # Second event: actual track change
933            mock_ynison.state = YnisonState(
934                active_device_id=provider._device_id,
935                player_state={
936                    "status": {"progress_ms": 0, "duration_ms": 0},
937                    "player_queue": {
938                        "current_playable_index": 0,
939                        "playable_list": [{"playable_id": "new_track"}],
940                        "entity_id": "user:onyourwave",
941                        "entity_type": "RADIO",
942                    },
943                },
944            )
945            provider._track_changed_event.set()
946
947        task = asyncio.create_task(simulate_echo_then_change())
948        result = await provider._wait_for_track_change("old_track", timeout=5.0)
949        assert result is True
950        await task
951
952    async def test_wait_for_track_change_returns_immediately_if_already_advanced(
953        self,
954    ) -> None:
955        """
956        If Ynison already advanced before the call, return True without waiting.
957
958        Regression: _wait_for_track_change used to clear _track_changed_event
959        before checking state, so a state update that arrived between
960        _signal_track_completion() and this method losing the signal and
961        stalled for the full 30s timeout.
962        """
963        provider = _make_provider()
964        mock_ynison = MagicMock()
965        # State already shows the NEW track at the time of entry
966        mock_ynison.state = YnisonState(
967            active_device_id=provider._device_id,
968            player_state={
969                "status": {"progress_ms": 0, "duration_ms": 0},
970                "player_queue": {
971                    "current_playable_index": 0,
972                    "playable_list": [{"playable_id": "new_track"}],
973                    "entity_id": "user:onyourwave",
974                    "entity_type": "RADIO",
975                },
976            },
977        )
978        provider._ynison = mock_ynison
979        # Event is already set (from the _activate_playback that ran before us)
980        # but pre-check in _wait_for_track_change should catch this regardless.
981        provider._track_changed_event.set()
982
983        # Tight timeout would fail if pre-check were absent — state check must
984        # happen before clear()+wait().
985        result = await provider._wait_for_track_change("old_track", timeout=0.1)
986        assert result is True
987
988    async def test_wait_for_track_change_timeout(self) -> None:
989        """_wait_for_track_change returns False on timeout."""
990        provider = _make_provider()
991        mock_ynison = MagicMock()
992        mock_ynison.state = YnisonState(
993            active_device_id=provider._device_id,
994            player_state={
995                "status": {"progress_ms": 248000},
996                "player_queue": {
997                    "current_playable_index": 5,
998                    "playable_list": [{"playable_id": "old_track"}],
999                    "entity_id": "user:onyourwave",
1000                    "entity_type": "RADIO",
1001                },
1002            },
1003        )
1004        provider._ynison = mock_ynison
1005
1006        result = await provider._wait_for_track_change("old_track", timeout=0.1)
1007        assert result is False
1008
1009
1010# ------------------------------------------------------------------
1011# ------------------------------------------------------------------
1012# PCM normalization (per-track ffmpeg → adaptive PCM)
1013# ------------------------------------------------------------------
1014
1015
1016class TestPCMNormalization:
1017    """Tests for per-track ffmpeg normalization to PCM."""
1018
1019    async def test_stream_track_always_uses_ffmpeg(self) -> None:
1020        """_stream_track always normalizes through ffmpeg, even without seek."""
1021        provider = _make_provider()
1022        provider._in_use_by_queue = "player1"
1023
1024        mock_yandex = MagicMock()
1025        sd = MagicMock()
1026        sd.expiration = 600
1027        sd.duration = 200
1028        sd.audio_format = MagicMock()
1029        mock_yandex.get_stream_details = AsyncMock(return_value=sd)
1030
1031        async def _fake_audio_stream(_details: object) -> Any:
1032            yield b"raw-cdn-data"
1033
1034        mock_yandex.get_audio_stream = _fake_audio_stream
1035        provider._yandex_provider = mock_yandex
1036
1037        mock_ynison = MagicMock()
1038        mock_ynison.update_playing_status = AsyncMock()
1039        mock_ynison.state.is_paused = False
1040        provider._ynison = mock_ynison
1041
1042        async def _fake_ffmpeg(**_kwargs: object) -> Any:
1043            yield b"pcm-normalized"
1044
1045        with patch(
1046            "music_assistant.providers.yandex_ynison.provider.get_ffmpeg_stream",
1047            side_effect=_fake_ffmpeg,
1048        ) as mock_ffmpeg:
1049            collected: list[bytes] = []
1050            async for chunk in provider._stream_track("track:123"):
1051                collected.append(chunk)
1052
1053        assert collected == [b"pcm-normalized"]
1054        mock_ffmpeg.assert_called_once()
1055        call_kwargs = mock_ffmpeg.call_args
1056        # Default (no YM provider linked) → lossy profile
1057        assert call_kwargs.kwargs["output_format"] == provider._normalized_format
1058        assert call_kwargs.kwargs["output_format"].content_type == ContentType.PCM_S16LE
1059        # No seek args when seek_ms=0. The per-track decode no longer paces
1060        # itself with -re (spec 0006): MA's realtime pacer is the single pacing
1061        # authority, and dropping -re lets a small read-ahead absorb CDN jitter.
1062        args = call_kwargs.kwargs.get("extra_input_args", [])
1063        assert "-re" not in args
1064        assert "-ss" not in args
1065
1066    async def test_stream_track_seek_adds_ss_arg(self) -> None:
1067        """With seek > 0, _stream_track adds -ss to ffmpeg args."""
1068        provider = _make_provider()
1069        provider._in_use_by_queue = "player1"
1070
1071        mock_yandex = MagicMock()
1072        sd = MagicMock()
1073        sd.expiration = 600
1074        sd.duration = 200
1075        sd.audio_format = MagicMock()
1076        mock_yandex.get_stream_details = AsyncMock(return_value=sd)
1077
1078        async def _fake_audio_stream(_details: object) -> Any:
1079            yield b"raw-data"
1080
1081        mock_yandex.get_audio_stream = _fake_audio_stream
1082        provider._yandex_provider = mock_yandex
1083
1084        mock_ynison = MagicMock()
1085        mock_ynison.update_playing_status = AsyncMock()
1086        mock_ynison.state.is_paused = False
1087        provider._ynison = mock_ynison
1088
1089        async def _fake_ffmpeg(**_kwargs: object) -> Any:
1090            yield b"pcm-seeked"
1091
1092        with patch(
1093            "music_assistant.providers.yandex_ynison.provider.get_ffmpeg_stream",
1094            side_effect=_fake_ffmpeg,
1095        ) as mock_ffmpeg:
1096            collected: list[bytes] = []
1097            async for chunk in provider._stream_track("track:123", seek_ms=5000):
1098                collected.append(chunk)
1099
1100        assert collected == [b"pcm-seeked"]
1101        mock_ffmpeg.assert_called_once()
1102        call_kwargs = mock_ffmpeg.call_args
1103        args = call_kwargs.kwargs.get("extra_input_args", [])
1104        assert "-ss" in args
1105        # -re removed in spec 0006 — realtime pacer is the only pacing authority.
1106        assert "-re" not in args
1107
1108    async def test_stream_track_logs_output_rate_and_bit_depth(
1109        self, caplog: pytest.LogCaptureFixture
1110    ) -> None:
1111        """
1112        The per-track stream log carries output sample rate AND bit depth.
1113
1114        Spec 0006 AC6: with the passthrough fast path the declared PCM format
1115        IS the delivered audio, so an operator must read the output rate and
1116        bit depth — not just the content type — from the one stream log line
1117        to tell rate passthrough from a resample.
1118        """
1119        provider = _make_provider()
1120        provider._in_use_by_queue = "player1"
1121        provider._normalized_params = dict(PCM_LOSSLESS_PARAMS)
1122
1123        mock_yandex = MagicMock()
1124        sd = MagicMock()
1125        sd.expiration = 600
1126        sd.duration = 200
1127        sd.audio_format = MagicMock()
1128        mock_yandex.get_stream_details = AsyncMock(return_value=sd)
1129
1130        async def _fake_audio_stream(_details: object) -> Any:
1131            yield b"raw"
1132
1133        mock_yandex.get_audio_stream = _fake_audio_stream
1134        provider._yandex_provider = mock_yandex
1135
1136        mock_ynison = MagicMock()
1137        mock_ynison.update_playing_status = AsyncMock()
1138        mock_ynison.state.is_paused = False
1139        provider._ynison = mock_ynison
1140
1141        async def _fake_ffmpeg(**_kwargs: object) -> Any:
1142            yield b"pcm"
1143
1144        with (
1145            patch(
1146                "music_assistant.providers.yandex_ynison.provider.get_ffmpeg_stream",
1147                side_effect=_fake_ffmpeg,
1148            ),
1149            caplog.at_level("INFO"),
1150        ):
1151            async for _ in provider._stream_track("track:123"):
1152                pass
1153
1154        stream_lines = [
1155            r.getMessage() for r in caplog.records if "Streaming track" in r.getMessage()
1156        ]
1157        assert stream_lines, "expected a 'Streaming track' log line"
1158        line = stream_lines[0]
1159        assert "44100Hz" in line
1160        assert "24bit" in line
1161
1162    async def test_default_format_is_pcm_s16le(self) -> None:
1163        """Default AudioSource audio_format is PCM s16le (lossy profile)."""
1164        provider = _make_provider()
1165        mapping = next(iter(provider._audio_source.provider_mappings))
1166        assert mapping.audio_format.content_type == ContentType.PCM_S16LE
1167        assert mapping.audio_format.sample_rate == 44100
1168        assert mapping.audio_format.bit_depth == 16
1169        assert mapping.audio_format.channels == 2
1170
1171    async def test_superb_quality_uses_lossless_profile(self) -> None:
1172        """
1173        When YM quality=superb and no hint, format is PCM s24le/44.1kHz.
1174
1175        Spec 0006: the no-hint lossless floor is CD-rate (44.1 kHz), not
1176        48 kHz — a missing format hint must not upsample the common case.
1177        """
1178        provider = _make_provider()
1179
1180        mock_yandex = MagicMock()
1181        mock_yandex.domain = "yandex_music"
1182        mock_yandex.type = ProviderType.MUSIC
1183        mock_yandex.config.get_value = MagicMock(return_value="superb")
1184        provider._yandex_provider = mock_yandex
1185        provider._update_normalized_format()
1186
1187        mock_yandex.config.get_value.assert_called_with("quality")
1188        assert provider._normalized_format.content_type == ContentType.PCM_S24LE
1189        assert provider._normalized_format.sample_rate == 44100
1190        assert provider._normalized_format.bit_depth == 24
1191        # AudioSource is rebuilt with the new audio_format on the provider_mapping
1192        mapping = next(iter(provider._audio_source.provider_mappings))
1193        assert mapping.audio_format == provider._normalized_format
1194
1195    async def test_balanced_quality_uses_lossy_profile(self) -> None:
1196        """When YM quality=balanced, format stays PCM s16le/44.1kHz."""
1197        provider = _make_provider()
1198
1199        mock_yandex = MagicMock()
1200        mock_yandex.domain = "yandex_music"
1201        mock_yandex.type = ProviderType.MUSIC
1202        mock_yandex.config.get_value = MagicMock(return_value="balanced")
1203        provider._yandex_provider = mock_yandex
1204        provider._update_normalized_format()
1205
1206        assert provider._normalized_format.content_type == ContentType.PCM_S16LE
1207        assert provider._normalized_format.sample_rate == 44100
1208        assert provider._normalized_format.bit_depth == 16
1209
1210    async def test_invalid_sample_rate_override_falls_back_to_auto(self) -> None:
1211        """Stale/tampered output_sample_rate values fall back to auto-detected, not crash."""
1212        provider = _make_provider()
1213        provider._cfg_sample_rate = "bogus"
1214        provider._cfg_bit_depth = OUTPUT_AUTO
1215
1216        mock_yandex = MagicMock()
1217        mock_yandex.domain = "yandex_music"
1218        mock_yandex.type = ProviderType.MUSIC
1219        mock_yandex.config.get_value = MagicMock(return_value="superb")
1220        provider._yandex_provider = mock_yandex
1221        provider._update_normalized_format()
1222
1223        assert provider._normalized_format.sample_rate == 44100
1224        assert provider._normalized_format.bit_depth == 24
1225        assert provider._normalized_format.content_type == ContentType.PCM_S24LE
1226
1227    async def test_invalid_bit_depth_override_falls_back_to_auto(self) -> None:
1228        """Off-list output_bit_depth falls back to auto base, keeping content_type consistent."""
1229        provider = _make_provider()
1230        provider._cfg_sample_rate = OUTPUT_AUTO
1231        # 32-bit is not offered; previously this would silently become S16LE
1232        provider._cfg_bit_depth = "32"
1233
1234        mock_yandex = MagicMock()
1235        mock_yandex.domain = "yandex_music"
1236        mock_yandex.type = ProviderType.MUSIC
1237        mock_yandex.config.get_value = MagicMock(return_value="superb")
1238        provider._yandex_provider = mock_yandex
1239        provider._update_normalized_format()
1240
1241        assert provider._normalized_format.bit_depth == 24
1242        assert provider._normalized_format.content_type == ContentType.PCM_S24LE
1243
1244    async def test_audio_format_not_modified_by_stream(self) -> None:
1245        """AudioSource audio_format stays fixed (not updated from stream)."""
1246        provider = _make_provider()
1247        provider._in_use_by_queue = "player1"
1248
1249        mock_yandex = MagicMock()
1250        sd = MagicMock()
1251        sd.expiration = 600
1252        sd.duration = 200
1253        sd.audio_format = MagicMock()  # different format
1254        mock_yandex.get_stream_details = AsyncMock(return_value=sd)
1255
1256        async def _fake_audio_stream(_details: object) -> Any:
1257            yield b"data"
1258
1259        mock_yandex.get_audio_stream = _fake_audio_stream
1260        provider._yandex_provider = mock_yandex
1261
1262        mock_ynison = MagicMock()
1263        mock_ynison.update_playing_status = AsyncMock()
1264        mock_ynison.state.is_paused = False
1265        provider._ynison = mock_ynison
1266
1267        async def _fake_ffmpeg(**_kwargs: object) -> Any:
1268            yield b"pcm"
1269
1270        original_format = provider._normalized_format
1271
1272        with patch(
1273            "music_assistant.providers.yandex_ynison.provider.get_ffmpeg_stream",
1274            side_effect=_fake_ffmpeg,
1275        ):
1276            async for _ in provider._stream_track("track:123"):
1277                pass
1278
1279        # AudioSource carries the format via its ProviderMapping in the new model.
1280        # We deliberately store a *fresh copy* per mapping (not `is original_format`)
1281        # so that MA's in-place ffmpeg mutations on `_normalized_format` cannot
1282        # leak into the mapping — but value equality must hold.
1283        mapping = next(iter(provider._audio_source.provider_mappings))
1284        assert mapping.audio_format == original_format
1285        assert mapping.audio_format is not original_format
1286
1287    async def test_stream_track_api_error_returns_empty(self) -> None:
1288        """If get_stream_details fails, _stream_track yields nothing."""
1289        provider = _make_provider()
1290        mock_yandex = MagicMock()
1291        provider._yandex_provider = mock_yandex
1292
1293        collected: list[bytes] = []
1294        with patch.object(
1295            provider,
1296            "_get_stream_details_with_retry",
1297            new=AsyncMock(side_effect=Exception("API error")),
1298        ):
1299            async for chunk in provider._stream_track("track:bad"):
1300                collected.append(chunk)
1301
1302        assert collected == []
1303
1304    async def test_stream_track_provider_unloaded_mid_stream_aborts_cleanly(
1305        self,
1306    ) -> None:
1307        """
1308        Unloaded linked provider mid-stream aborts cleanly (no AttributeError).
1309
1310        Regression: the path between `await _get_stream_details_with_retry`
1311        and the ffmpeg stream builder used to dereference
1312        `self._yandex_provider` directly, racing with
1313        `_check_yandex_provider_match` which nulls the attribute on unload.
1314        """
1315        provider = _make_provider()
1316        mock_yandex = MagicMock()
1317        sd = MagicMock()
1318        sd.expiration = 600
1319        sd.audio_format = MagicMock()
1320        sd.to_dict.return_value = {"track_id": "t1"}
1321        sd.data = {"url": "https://cdn.example.com/audio.mp3"}
1322
1323        async def fetch_and_null(_track_id: str, _media_type: Any = None) -> Any:
1324            # Simulate the background unload task firing while we awaited.
1325            provider._yandex_provider = None
1326            return sd
1327
1328        mock_yandex.get_stream_details = AsyncMock(side_effect=fetch_and_null)
1329        provider._yandex_provider = mock_yandex
1330
1331        collected: list[bytes] = []
1332        async for chunk in provider._stream_track("t1"):
1333            collected.append(chunk)
1334
1335        assert collected == []
1336        assert provider._stream_stop_event.is_set()
1337
1338
1339def _make_ym_provider_stub(
1340    instance_id: str = "ym-inst",
1341    token: str | None = None,
1342    x_token: str | None = None,
1343) -> MagicMock:
1344    """Build a stub yandex_music provider with a config exposing token/x_token."""
1345    values: dict[str, Any] = {"token": token, "x_token": x_token}
1346    ym_config = MagicMock()
1347    ym_config.get_value.side_effect = values.get
1348    ym = MagicMock()
1349    ym.instance_id = instance_id
1350    ym.domain = "yandex_music"
1351    ym.type = ProviderType.MUSIC
1352    ym.config = ym_config
1353    return ym
1354
1355
1356class TestPlayerRateSnap:
1357    """
1358    _update_normalized_format snaps the declared rate to player capability.
1359
1360    Spec 0006 AC9-11: the auto rate is snapped down to the nearest sample rate
1361    the target player supports, so MA's AudioSource passthrough fast path is
1362    hit and no second resampling ffmpeg runs. Explicit overrides are never
1363    snapped; an unresolvable player leaves the rate as the hint/floor produced.
1364    """
1365
1366    @staticmethod
1367    def _hint(sample_rate: int, bit_depth: int = 24) -> AudioFormat:
1368        return AudioFormat(
1369            content_type=ContentType.PCM_S24LE if bit_depth == 24 else ContentType.PCM_S16LE,
1370            sample_rate=sample_rate,
1371            bit_depth=bit_depth,
1372            channels=2,
1373        )
1374
1375    @staticmethod
1376    def _link_player(provider: YandexYnisonProvider, rates: list[tuple[int, int]]) -> None:
1377        player = MagicMock()
1378        player.get_supported_sample_rates = MagicMock(return_value=rates)
1379        provider._active_player_id = "p1"
1380        provider.mass.players.get_player = MagicMock(return_value=player)  # type: ignore[attr-defined]
1381
1382    async def test_hi_res_rate_snapped_down_to_supported(self) -> None:
1383        """A 96 kHz hint on a 48 kHz-max player declares 48 kHz (fast-path hit)."""
1384        provider = _make_provider()
1385        self._link_player(provider, [(44100, 16), (48000, 24)])
1386        provider._update_normalized_format(hint=self._hint(96000))
1387        assert provider._normalized_format.sample_rate == 48000
1388        # bit depth is outside the snap — it follows the hint, not the player.
1389        assert provider._normalized_format.bit_depth == 24
1390
1391    async def test_rate_snapped_to_only_supported_value(self) -> None:
1392        """A 48 kHz hint on a 44.1 kHz-only player declares 44.1 kHz."""
1393        provider = _make_provider()
1394        self._link_player(provider, [(44100, 16)])
1395        provider._update_normalized_format(hint=self._hint(48000))
1396        assert provider._normalized_format.sample_rate == 44100
1397
1398    async def test_supported_rate_left_untouched(self) -> None:
1399        """A 48 kHz hint on a player that supports 48 kHz is not snapped (AC10)."""
1400        provider = _make_provider()
1401        self._link_player(provider, [(44100, 16), (48000, 24)])
1402        provider._update_normalized_format(hint=self._hint(48000))
1403        assert provider._normalized_format.sample_rate == 48000
1404
1405    async def test_no_resolvable_player_keeps_hint_rate(self) -> None:
1406        """With no target player the hint rate survives and nothing raises (AC10)."""
1407        provider = _make_provider()
1408        # default mock: get_player → None, all_players → [] → target player None
1409        provider._update_normalized_format(hint=self._hint(96000))
1410        assert provider._normalized_format.sample_rate == 96000
1411
1412    async def test_explicit_override_beats_snap(self) -> None:
1413        """Explicit output_sample_rate wins over the player snap (AC11)."""
1414        provider = _make_provider()
1415        provider._cfg_sample_rate = "96000"
1416        self._link_player(provider, [(44100, 16), (48000, 24)])
1417        provider._update_normalized_format(hint=self._hint(48000))
1418        assert provider._normalized_format.sample_rate == 96000
1419
1420
1421class TestResolveTokenOwnMode:
1422    """_resolve_token in own mode (manual token, no refresh)."""
1423
1424    async def test_returns_stored_token(self) -> None:
1425        """Returns the manually configured music token as-is."""
1426        provider = _make_provider()
1427        provider.config = _make_mock_config(
1428            {CONF_TOKEN: "manual-token", CONF_YM_INSTANCE: YM_INSTANCE_OWN}
1429        )
1430        provider._ym_instance_id = None
1431
1432        result = await provider._resolve_token()
1433
1434        assert result.get_secret() == "manual-token"
1435
1436    async def test_raises_when_no_token(self) -> None:
1437        """Raises LoginFailed when CONF_TOKEN and CONF_X_TOKEN are both empty."""
1438        provider = _make_provider()
1439        provider.config = _make_mock_config(
1440            {CONF_TOKEN: None, CONF_X_TOKEN: None, CONF_YM_INSTANCE: YM_INSTANCE_OWN}
1441        )
1442        provider._ym_instance_id = None
1443
1444        with pytest.raises(LoginFailed, match="No Yandex Music token"):
1445            await provider._resolve_token()
1446
1447    async def test_falls_back_to_x_token_refresh_when_token_missing(self) -> None:
1448        """Own mode with stored x_token but no music token refreshes in-memory."""
1449        provider = _make_provider()
1450        provider.config = _make_mock_config(
1451            {CONF_TOKEN: None, CONF_X_TOKEN: "own-x-token", CONF_YM_INSTANCE: YM_INSTANCE_OWN}
1452        )
1453        provider._ym_instance_id = None
1454
1455        with patch(
1456            "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
1457            new_callable=AsyncMock,
1458            return_value=SecretStr("refreshed"),
1459        ) as mock_refresh:
1460            result = await provider._resolve_token()
1461
1462        assert result.get_secret() == "refreshed"
1463        mock_refresh.assert_awaited_once()
1464        await_args = mock_refresh.await_args
1465        assert await_args is not None
1466        sent: SecretStr = await_args.args[0]
1467        assert sent.get_secret() == "own-x-token"
1468
1469
1470class TestResolveTokenBorrowMode:
1471    """_resolve_token in borrow mode (reads from linked yandex_music instance)."""
1472
1473    async def test_uses_ym_token_when_available(self) -> None:
1474        """Returns the music token from the linked YM instance config."""
1475        provider = _make_provider()
1476        provider._ym_instance_id = "ym-inst"
1477        provider._borrow_source = BorrowedCredentialSource(provider.mass, "ym-inst")
1478        ym = _make_ym_provider_stub(token="ym-music-token")
1479        _stub_attr(provider.mass, "get_provider", MagicMock(return_value=ym))
1480
1481        result = await provider._resolve_token()
1482
1483        assert result.get_secret() == "ym-music-token"
1484
1485    async def test_refreshes_in_memory_when_only_x_token(self) -> None:
1486        """Falls back to in-memory refresh via x_token; does not write config."""
1487        provider = _make_provider()
1488        provider._ym_instance_id = "ym-inst"
1489        provider._borrow_source = BorrowedCredentialSource(provider.mass, "ym-inst")
1490        ym = _make_ym_provider_stub(token=None, x_token="ym-x-token")
1491        _stub_attr(provider.mass, "get_provider", MagicMock(return_value=ym))
1492
1493        with patch(
1494            "ya_passport_auth.ma.borrow.refresh_music_token",
1495            new_callable=AsyncMock,
1496            return_value=SecretStr("fresh-token"),
1497        ) as mock_refresh:
1498            result = await provider._resolve_token()
1499
1500        assert result.get_secret() == "fresh-token"
1501        mock_refresh.assert_awaited_once()
1502
1503    async def test_raises_when_ym_has_no_credentials(self) -> None:
1504        """Raises LoginFailed when YM instance config has neither token nor x_token."""
1505        provider = _make_provider()
1506        provider._ym_instance_id = "ym-inst"
1507        provider._borrow_source = BorrowedCredentialSource(provider.mass, "ym-inst")
1508        ym = _make_ym_provider_stub(token=None, x_token=None)
1509        _stub_attr(provider.mass, "get_provider", MagicMock(return_value=ym))
1510
1511        with pytest.raises(LoginFailed, match="no credentials"):
1512            await provider._resolve_token()
1513
1514    async def test_raises_when_ym_instance_unavailable(self) -> None:
1515        """A missing YM instance is a startup-ordering condition — transient error."""
1516        provider = _make_provider()
1517        provider._ym_instance_id = "ym-inst"
1518        provider._borrow_source = BorrowedCredentialSource(provider.mass, "ym-inst")
1519        _stub_attr(provider.mass, "get_provider", MagicMock(return_value=None))
1520
1521        with pytest.raises(ResourceTemporarilyUnavailable, match="not loaded"):
1522            await provider._resolve_token()
1523
1524    async def test_raises_when_linked_provider_is_not_yandex_music(self) -> None:
1525        """Stale/edited instance id pointing at a non-YM provider yields a clear error."""
1526        provider = _make_provider()
1527        provider._ym_instance_id = "some-other-id"
1528        provider._borrow_source = BorrowedCredentialSource(provider.mass, "some-other-id")
1529        wrong = _make_ym_provider_stub()
1530        wrong.domain = "spotify"  # not yandex_music
1531        _stub_attr(provider.mass, "get_provider", MagicMock(return_value=wrong))
1532
1533        with pytest.raises(LoginFailed, match="not a Yandex Music"):
1534            await provider._resolve_token()
1535
1536
1537class TestRefreshYnisonToken:
1538    """_refresh_ynison_token on YnisonClient auth-failure callback."""
1539
1540    async def test_own_mode_no_x_token_raises_login_failed(self) -> None:
1541        """Own mode with neither token nor stored x_token — surface LoginFailed."""
1542        provider = _make_provider()
1543        provider._ym_instance_id = None
1544        # Stub the config: no token, no x_token.
1545        provider.config = MagicMock()
1546        provider.config.get_value = MagicMock(return_value=None)
1547
1548        with pytest.raises(LoginFailed, match="Re-authenticate"):
1549            await provider._refresh_ynison_token()
1550
1551    async def test_own_mode_with_stored_x_token_refreshes(self) -> None:
1552        """Own mode with CONF_X_TOKEN set refreshes in-memory via passport."""
1553        provider = _make_provider()
1554        provider._ym_instance_id = None
1555        provider.config = MagicMock()
1556        provider.config.get_value = MagicMock(
1557            side_effect=lambda key, default=None: "own-x-token" if key == CONF_X_TOKEN else default
1558        )
1559
1560        with patch(
1561            "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
1562            new_callable=AsyncMock,
1563            return_value=SecretStr("fresh"),
1564        ) as mock_refresh:
1565            result = await provider._refresh_ynison_token()
1566
1567        assert result.get_secret() == "fresh"
1568        mock_refresh.assert_awaited_once()
1569        # The refresh argument is a SecretStr wrapping the stored x_token.
1570        await_args = mock_refresh.await_args
1571        assert await_args is not None
1572        sent: SecretStr = await_args.args[0]
1573        assert sent.get_secret() == "own-x-token"
1574
1575    async def test_borrow_mode_refreshes_from_ym_x_token(self) -> None:
1576        """Reads x_token from linked YM and refreshes in-memory only."""
1577        provider = _make_provider()
1578        provider._ym_instance_id = "ym-inst"
1579        provider._borrow_source = BorrowedCredentialSource(provider.mass, "ym-inst")
1580        ym = _make_ym_provider_stub(token="stale", x_token="ym-x-token")
1581        _stub_attr(provider.mass, "get_provider", MagicMock(return_value=ym))
1582        # Ensure config writes are not invoked
1583        mock_update_config = MagicMock()
1584        _stub_attr(provider, "_update_config_value", mock_update_config)
1585
1586        with patch(
1587            "ya_passport_auth.ma.borrow.refresh_music_token",
1588            new_callable=AsyncMock,
1589            return_value=SecretStr("fresh-token"),
1590        ) as mock_refresh:
1591            result = await provider._refresh_ynison_token()
1592
1593        assert result.get_secret() == "fresh-token"
1594        mock_refresh.assert_awaited_once()
1595        mock_update_config.assert_not_called()
1596
1597    async def test_borrow_mode_raises_without_x_token(self) -> None:
1598        """Raises LoginFailed when YM has no x_token for refresh."""
1599        provider = _make_provider()
1600        provider._ym_instance_id = "ym-inst"
1601        provider._borrow_source = BorrowedCredentialSource(provider.mass, "ym-inst")
1602        ym = _make_ym_provider_stub(token="only-token", x_token=None)
1603        _stub_attr(provider.mass, "get_provider", MagicMock(return_value=ym))
1604
1605        with pytest.raises(LoginFailed, match="no x_token"):
1606            await provider._refresh_ynison_token()
1607
1608    async def test_borrow_mode_raises_when_ym_not_loaded(self) -> None:
1609        """A missing YM instance is transient on reactive refresh too."""
1610        provider = _make_provider()
1611        provider._ym_instance_id = "ym-inst"
1612        provider._borrow_source = BorrowedCredentialSource(provider.mass, "ym-inst")
1613        _stub_attr(provider.mass, "get_provider", MagicMock(return_value=None))
1614
1615        with pytest.raises(ResourceTemporarilyUnavailable, match="not loaded"):
1616            await provider._refresh_ynison_token()
1617
1618
1619class TestYandexProviderMatch:
1620    """_check_yandex_provider_match obeys _ym_instance_id in borrow mode."""
1621
1622    async def test_borrow_mode_ignores_other_ym_instances(self) -> None:
1623        """Does not link to a YM instance with a different instance_id."""
1624        provider = _make_provider()
1625        provider._ym_instance_id = "wanted"
1626        other = _make_ym_provider_stub(instance_id="other")
1627        _stub_attr(provider.mass, "get_providers", MagicMock(return_value=[other]))
1628
1629        await provider._check_yandex_provider_match()
1630
1631        assert provider._yandex_provider is None
1632
1633    async def test_borrow_mode_matches_on_instance_id(self) -> None:
1634        """Links to the specific YM instance requested by config."""
1635        provider = _make_provider()
1636        provider._ym_instance_id = "wanted"
1637        wanted = _make_ym_provider_stub(instance_id="wanted")
1638        other = _make_ym_provider_stub(instance_id="other")
1639        _stub_attr(provider.mass, "get_providers", MagicMock(return_value=[other, wanted]))
1640
1641        await provider._check_yandex_provider_match()
1642
1643        assert provider._yandex_provider is wanted
1644
1645    async def test_own_mode_accepts_any_ym(self) -> None:
1646        """In own mode, the first available yandex_music provider is used."""
1647        provider = _make_provider()
1648        provider._ym_instance_id = None
1649        ym = _make_ym_provider_stub(instance_id="any")
1650        _stub_attr(provider.mass, "get_providers", MagicMock(return_value=[ym]))
1651
1652        await provider._check_yandex_provider_match()
1653
1654        assert provider._yandex_provider is ym
1655
1656
1657# ------------------------------------------------------------------
1658# Instance name postfix
1659# ------------------------------------------------------------------
1660
1661
1662class TestInstanceNamePostfix:
1663    """Tests for instance_name_postfix property."""
1664
1665    def test_returns_custom_display_name(self) -> None:
1666        """Returns display_name when it differs from the default."""
1667        config = _make_mock_config({CONF_PUBLISH_NAME: "Living Room"})
1668        mass = _make_mock_mass()
1669        manifest = _make_mock_manifest()
1670        provider = YandexYnisonProvider(mass, manifest, config, {ProviderFeature.AUDIO_SOURCE})
1671        assert provider.instance_name_postfix == "Living Room"
1672
1673    def test_returns_none_for_default_name(self) -> None:
1674        """Returns None when display_name is the default (falls back to index)."""
1675        provider = _make_provider()
1676        assert provider.instance_name_postfix is None
1677
1678
1679# ------------------------------------------------------------------
1680# Yandex Music instance enumeration
1681# ------------------------------------------------------------------
1682
1683
1684class TestListYandexMusicInstances:
1685    """Tests for list_yandex_music_instances."""
1686
1687    def test_returns_empty_when_none_configured(self) -> None:
1688        """Empty list when no yandex_music instances exist."""
1689        mass = _make_mock_mass()
1690        mass.config.get = MagicMock(return_value={})
1691        assert list_yandex_music_instances(mass) == []
1692
1693    def test_lists_instances_with_display_name(self) -> None:
1694        """Returns (instance_id, display_name) pairs for yandex_music domains."""
1695        mass = _make_mock_mass()
1696        mass.config.get = MagicMock(
1697            return_value={
1698                "ym-a": {"domain": "yandex_music", "name": "Main Account"},
1699                "ym-b": {"domain": "yandex_music", "name": "Family"},
1700                "ynison-1": {"domain": "yandex_ynison", "name": "Ynison"},
1701            }
1702        )
1703        result = list_yandex_music_instances(mass)
1704        assert sorted(result) == [("ym-a", "Main Account"), ("ym-b", "Family")]
1705
1706    def test_falls_back_to_instance_id_when_name_missing(self) -> None:
1707        """Uses instance_id as display name when 'name' is absent."""
1708        mass = _make_mock_mass()
1709        mass.config.get = MagicMock(return_value={"ym-a": {"domain": "yandex_music"}})
1710        result = list_yandex_music_instances(mass)
1711        assert result == [("ym-a", "ym-a")]
1712
1713
1714class TestPCMFrameAlignment:
1715    """Tests for PCM frame alignment padding in get_audio_stream."""
1716
1717    async def test_frame_alignment_padding_s24le(self) -> None:
1718        """Verify padding math for s24le stereo (frame_size=6)."""
1719        provider = _make_provider()
1720        provider._normalized_format = make_pcm_format(PCM_LOSSLESS_PARAMS)
1721        fmt = provider._normalized_format
1722        frame_size = (fmt.bit_depth // 8) * fmt.channels
1723        assert frame_size == 6  # 3 bytes x 2 channels
1724
1725        # 4096 bytes yielded: 4096 % 6 = 4, need 2 bytes padding
1726        bytes_yielded = 4096
1727        remainder = bytes_yielded % frame_size
1728        assert remainder == 4
1729        pad = frame_size - remainder
1730        assert pad == 2
1731
1732    async def test_frame_alignment_padding_s16le(self) -> None:
1733        """Verify padding math for s16le stereo (frame_size=4)."""
1734        provider = _make_provider()
1735        provider._normalized_format = make_pcm_format(PCM_LOSSY_PARAMS)
1736        fmt = provider._normalized_format
1737        frame_size = (fmt.bit_depth // 8) * fmt.channels
1738        assert frame_size == 4  # 2 bytes x 2 channels
1739
1740        # 4096 is already aligned to 4
1741        assert 4096 % frame_size == 0
1742
1743        # 4097 needs 3 bytes padding
1744        assert 4097 % frame_size == 1
1745        assert frame_size - (4097 % frame_size) == 3
1746
1747    async def test_no_padding_when_aligned(self) -> None:
1748        """No padding needed when bytes_yielded is already frame-aligned."""
1749        fmt = make_pcm_format(PCM_LOSSLESS_PARAMS)
1750        frame_size = (fmt.bit_depth // 8) * fmt.channels
1751        # 6000 bytes = 1000 frames of s24le stereo
1752        assert 6000 % frame_size == 0
1753
1754
1755# ------------------------------------------------------------------
1756# Playback controls
1757# ------------------------------------------------------------------
1758
1759
1760def _make_ynison_state(
1761    *,
1762    progress_ms: int = 5000,
1763    duration_ms: int = 120000,
1764    paused: bool = False,
1765    current_playable_index: int = 0,
1766    playable_list: list[dict[str, Any]] | None = None,
1767    device_id: str = "test-device-uuid",
1768) -> YnisonState:
1769    """Build a YnisonState for control-flow tests."""
1770    if playable_list is None:
1771        playable_list = [{"playable_id": "track1"}]
1772    return YnisonState(
1773        active_device_id=device_id,
1774        player_state={
1775            "status": {
1776                "paused": paused,
1777                "progress_ms": progress_ms,
1778                "duration_ms": duration_ms,
1779            },
1780            "player_queue": {
1781                "current_playable_index": current_playable_index,
1782                "playable_list": playable_list,
1783            },
1784        },
1785    )
1786
1787
1788def _mock_ynison(
1789    state: YnisonState | None = None,
1790    connected: bool = True,
1791    device_id: str = "test-device-uuid",
1792) -> MagicMock:
1793    """Create a mock YnisonClient with sensible defaults."""
1794    mock = MagicMock()
1795    mock.connected = connected
1796    mock.state = state or _make_ynison_state()
1797    mock.device_id = device_id
1798    mock.update_playing_status = AsyncMock()
1799    mock.update_player_state = AsyncMock()
1800    return mock
1801
1802
1803class TestPlaybackControls:
1804    """Tests for _on_play, _on_pause, _on_next, _on_previous, _on_seek."""
1805
1806    async def test_on_play_sends_progress_unpaused(self) -> None:
1807        """_on_play sends update_playing_status with paused=False."""
1808        provider = _make_provider()
1809        provider._actual_duration_ms = 120000
1810        state = _make_ynison_state(progress_ms=5000, duration_ms=120000, paused=True)
1811        mock_yn = _mock_ynison(state)
1812        provider._ynison = mock_yn
1813
1814        await provider._on_play()
1815
1816        mock_yn.update_playing_status.assert_awaited_once_with(
1817            progress_ms=5000, duration_ms=120000, paused=False, strict=True
1818        )
1819
1820    async def test_on_play_no_ynison_raises(self) -> None:
1821        """_on_play raises when Ynison is not connected."""
1822        provider = _make_provider()
1823        provider._ynison = None
1824
1825        with pytest.raises(UnsupportedFeaturedException):
1826            await provider._on_play()
1827
1828    async def test_on_pause_sends_progress_paused(self) -> None:
1829        """_on_pause sends update_playing_status with paused=True."""
1830        provider = _make_provider()
1831        provider._actual_duration_ms = 120000
1832        state = _make_ynison_state(progress_ms=5000, duration_ms=120000, paused=False)
1833        mock_yn = _mock_ynison(state)
1834        provider._ynison = mock_yn
1835
1836        await provider._on_pause()
1837
1838        mock_yn.update_playing_status.assert_awaited_once_with(
1839            progress_ms=5000, duration_ms=120000, paused=True, strict=True
1840        )
1841
1842    async def test_on_pause_no_ynison_raises(self) -> None:
1843        """_on_pause raises when Ynison is not connected."""
1844        provider = _make_provider()
1845        provider._ynison = None
1846
1847        with pytest.raises(UnsupportedFeaturedException):
1848            await provider._on_pause()
1849
1850    async def test_on_next_calls_signal_completion(self) -> None:
1851        """_on_next triggers _signal_track_completion."""
1852        provider = _make_provider()
1853        state = _make_ynison_state(
1854            progress_ms=180000,
1855            duration_ms=200000,
1856            playable_list=[{"playable_id": "t1"}, {"playable_id": "t2"}],
1857        )
1858        mock_yn = _mock_ynison(state)
1859        provider._ynison = mock_yn
1860
1861        await provider._on_next()
1862
1863        # Should have reported completion and advanced
1864        mock_yn.update_playing_status.assert_awaited_once()
1865        mock_yn.update_player_state.assert_awaited_once()
1866
1867    async def test_on_next_no_ynison_raises(self) -> None:
1868        """_on_next raises when Ynison is not connected."""
1869        provider = _make_provider()
1870        provider._ynison = None
1871
1872        with pytest.raises(UnsupportedFeaturedException):
1873            await provider._on_next()
1874
1875    async def test_on_previous_decrements_index(self) -> None:
1876        """_on_previous decrements current_playable_index by 1."""
1877        provider = _make_provider()
1878        state = _make_ynison_state(
1879            current_playable_index=2,
1880            playable_list=[
1881                {"playable_id": "t1"},
1882                {"playable_id": "t2"},
1883                {"playable_id": "t3"},
1884            ],
1885        )
1886        mock_yn = _mock_ynison(state)
1887        provider._ynison = mock_yn
1888
1889        await provider._on_previous()
1890
1891        mock_yn.update_player_state.assert_awaited_once()
1892        sent = mock_yn.update_player_state.call_args.kwargs["player_state"]
1893        assert sent["player_queue"]["current_playable_index"] == 1
1894
1895    async def test_on_previous_at_zero_no_op(self) -> None:
1896        """_on_previous at index 0 does nothing."""
1897        provider = _make_provider()
1898        state = _make_ynison_state(current_playable_index=0)
1899        mock_yn = _mock_ynison(state)
1900        provider._ynison = mock_yn
1901
1902        await provider._on_previous()
1903
1904        mock_yn.update_player_state.assert_not_called()
1905
1906    async def test_on_previous_no_ynison_raises(self) -> None:
1907        """_on_previous raises when Ynison is not connected."""
1908        provider = _make_provider()
1909        provider._ynison = None
1910
1911        with pytest.raises(UnsupportedFeaturedException):
1912            await provider._on_previous()
1913
1914    async def test_on_seek_updates_position(self) -> None:
1915        """_on_seek sends progress and triggers local stream restart."""
1916        provider = _make_provider()
1917        provider._actual_duration_ms = 200000
1918        state = _make_ynison_state(progress_ms=5000, duration_ms=200000, paused=False)
1919        mock_yn = _mock_ynison(state)
1920        provider._ynison = mock_yn
1921
1922        await provider._on_seek(30)  # 30 seconds
1923
1924        assert provider._seek_position_ms == 30000
1925        assert provider._track_changed_event.is_set()
1926        mock_yn.update_playing_status.assert_awaited_once_with(
1927            progress_ms=30000, duration_ms=200000, paused=False, strict=True
1928        )
1929
1930    async def test_on_seek_no_ynison_raises(self) -> None:
1931        """_on_seek raises when Ynison is not connected."""
1932        provider = _make_provider()
1933        provider._ynison = None
1934
1935        with pytest.raises(UnsupportedFeaturedException):
1936            await provider._on_seek(10)
1937
1938
1939# ------------------------------------------------------------------
1940# _send_progress_to_ynison
1941# ------------------------------------------------------------------
1942
1943
1944class TestSendProgressToYnison:
1945    """Tests for _send_progress_to_ynison."""
1946
1947    async def test_clamps_to_duration(self) -> None:
1948        """Progress is clamped to duration_ms."""
1949        provider = _make_provider()
1950        provider._ynison = _mock_ynison()
1951
1952        await provider._send_progress_to_ynison(150000, 100000, False)
1953
1954        provider._ynison.update_playing_status.assert_awaited_once_with(
1955            progress_ms=100000, duration_ms=100000, paused=False, strict=False
1956        )
1957
1958    async def test_zero_duration_no_send(self) -> None:
1959        """Does not send when duration is 0."""
1960        provider = _make_provider()
1961        provider._ynison = _mock_ynison()
1962
1963        await provider._send_progress_to_ynison(5000, 0, False)
1964
1965        provider._ynison.update_playing_status.assert_not_called()
1966
1967    async def test_not_connected_no_send(self) -> None:
1968        """Does not send when Ynison is disconnected."""
1969        provider = _make_provider()
1970        provider._ynison = _mock_ynison(connected=False)
1971
1972        await provider._send_progress_to_ynison(5000, 10000, False)
1973
1974        provider._ynison.update_playing_status.assert_not_called()
1975
1976    async def test_no_ynison_no_send(self) -> None:
1977        """Does not crash when _ynison is None."""
1978        provider = _make_provider()
1979        provider._ynison = None
1980
1981        await provider._send_progress_to_ynison(5000, 10000, False)
1982        # No assertion — just verify no crash.
1983
1984
1985# ------------------------------------------------------------------
1986# _pause_playback
1987# ------------------------------------------------------------------
1988
1989
1990class TestPausePlayback:
1991    """Tests for `_pause_playback` — external pause releases the player."""
1992
1993    async def test_sets_stop_event_and_cmd_stops_queue(self) -> None:
1994        """External pause sets the stop event and calls cmd_stop on the queue id."""
1995        provider = _make_provider()
1996        provider._active_player_id = "spb_bridge1"
1997        provider._in_use_by_queue = "player1"
1998
1999        await provider._pause_playback()
2000
2001        assert provider._stream_stop_event.is_set()
2002        provider.mass.players.cmd_stop.assert_awaited_once_with("player1")
2003
2004    async def test_rewrites_active_player_id_after_successful_cmd_stop(self) -> None:
2005        """
2006        On a successful cmd_stop, `_active_player_id` demotes to the queue id.
2007
2008        Queues live on the bare ALSA UUID; bridge wrappers (`spb_*`) do
2009        not own one. Resume's `play_media(_active_player_id, ...)`
2010        must target the queue id — otherwise MA raises
2011        `PlayerUnavailableError`. The demotion happens AFTER cmd_stop
2012        so a failure path leaves the bridge id intact for next attempt.
2013        """
2014        provider = _make_provider()
2015        provider._active_player_id = "spb_bridge1"
2016        provider._in_use_by_queue = "player1"
2017
2018        # capture _active_player_id at the moment cmd_stop is invoked —
2019        # must still be the bridge id (rewrite is post-success).
2020        captured: dict[str, str | None] = {}
2021
2022        async def _capture_stop(_player_id: str) -> None:
2023            captured["active_player_id_at_call"] = provider._active_player_id
2024
2025        provider.mass.players.cmd_stop = AsyncMock(side_effect=_capture_stop)
2026
2027        await provider._pause_playback()
2028
2029        assert captured["active_player_id_at_call"] == "spb_bridge1"
2030        assert provider._active_player_id == "player1"
2031        # _externally_paused flag set so the resume edge in _activate_playback can
2032        # detect us even if _stream_stop_event has been cleared by some
2033        # other code path.
2034        assert provider._externally_paused is True
2035
2036    async def test_cmd_stop_failure_keeps_bridge_id_intact(self) -> None:
2037        """
2038        A cmd_stop failure must not demote `_active_player_id`.
2039
2040        If we demoted to the queue id but cmd_stop never reached MA,
2041        the next `_activate_playback` would try `play_media(bare_uuid)`
2042        without MA ever having released the bridge — wedges the bridge
2043        in an inconsistent state. Better to keep the bridge id pinned
2044        and let the next pause attempt redo the cycle.
2045        """
2046        provider = _make_provider()
2047        provider._active_player_id = "spb_bridge1"
2048        provider._in_use_by_queue = "player1"
2049        provider.mass.players.cmd_stop = AsyncMock(side_effect=RuntimeError("boom"))
2050
2051        await provider._pause_playback()
2052
2053        assert provider._active_player_id == "spb_bridge1"
2054        # Stream stop event still set — generator must exit even if MA
2055        # never confirmed; otherwise we'd serve real audio against a
2056        # player MA thinks is detached.
2057        assert provider._stream_stop_event.is_set()
2058        # _externally_paused stays False on the failure path — we're not in a
2059        # "successfully paused, expecting resume" state.
2060        assert provider._externally_paused is False
2061
2062    async def test_no_active_player_is_a_noop(self) -> None:
2063        """Pause with no active queue does not call cmd_stop or set the stop event."""
2064        provider = _make_provider()
2065        provider._active_player_id = None
2066        provider._in_use_by_queue = None
2067
2068        await provider._pause_playback()
2069
2070        assert not provider._stream_stop_event.is_set()
2071        provider.mass.players.cmd_stop.assert_not_called()
2072
2073
2074# ------------------------------------------------------------------
2075# Echo suppression via YnisonState.last_update_is_echo
2076# ------------------------------------------------------------------
2077
2078
2079class TestEchoSuppression:
2080    """Seek detection in _handle_ynison_state honours the state echo flag."""
2081
2082    def _player(self, provider: YandexYnisonProvider) -> MagicMock:
2083        player = MagicMock()
2084        player.player_id = "player1"
2085        player.display_name = "Player 1"
2086        player.state.playback_state = PlaybackState.PLAYING
2087        provider.mass.players.all_players.return_value = [player]  # type: ignore[attr-defined]
2088        provider.mass.players.get_player.return_value = player  # type: ignore[attr-defined]
2089        return player
2090
2091    async def _prime_same_track(self, provider: YandexYnisonProvider) -> None:
2092        """Set provider state so seek detection is the only active branch."""
2093        self._player(provider)
2094        provider._current_streaming_track_id = "track1"
2095        provider._active_player_id = "player1"
2096        provider._in_use_by_queue = "player1"
2097        provider._streaming_progress_ms = 1000
2098        provider._seek_grace_until = 0.0  # grace expired
2099
2100    async def test_echo_suppresses_seek_detection(self) -> None:
2101        """last_update_is_echo=True makes large drift ignored."""
2102        provider = _make_provider()
2103        await self._prime_same_track(provider)
2104
2105        state = _make_ynison_state(progress_ms=10000)  # drift 9000ms vs 1000
2106        state.last_update_is_echo = True
2107
2108        await provider._handle_ynison_state(state)
2109
2110        assert not provider._track_changed_event.is_set()
2111
2112    async def test_non_echo_triggers_seek(self) -> None:
2113        """last_update_is_echo=False with large drift triggers seek."""
2114        provider = _make_provider()
2115        await self._prime_same_track(provider)
2116
2117        state = _make_ynison_state(progress_ms=10000)
2118        state.last_update_is_echo = False
2119
2120        await provider._handle_ynison_state(state)
2121
2122        assert provider._track_changed_event.is_set()
2123        assert provider._seek_position_ms == 10000
2124
2125    async def test_default_echo_flag_false(self) -> None:
2126        """YnisonState default last_update_is_echo is False — seek still fires."""
2127        provider = _make_provider()
2128        await self._prime_same_track(provider)
2129
2130        state = _make_ynison_state(progress_ms=10000)
2131        # No explicit override — the dataclass default is False.
2132
2133        await provider._handle_ynison_state(state)
2134
2135        assert provider._track_changed_event.is_set()
2136
2137
2138# ------------------------------------------------------------------
2139# _sync_progress
2140# ------------------------------------------------------------------
2141
2142
2143class TestSyncProgress:
2144    """Tests for _sync_progress."""
2145
2146    async def test_updates_metadata_and_ynison(self) -> None:
2147        """Sync updates MA metadata and sends progress to Ynison."""
2148        provider = _make_provider()
2149        provider._actual_duration_ms = 200000
2150        provider._ynison = _mock_ynison()
2151
2152        # 5 seconds of 44100Hz/16bit/2ch audio
2153        byte_rate = 44100 * 2 * 2
2154        bytes_yielded = byte_rate * 5
2155
2156        await provider._sync_progress(0, bytes_yielded, "player1")
2157
2158        # live progress lives on _stream_metadata, pushed through streamdetails
2159        assert provider._stream_metadata.elapsed_time == 5
2160        provider.mass.players.trigger_player_update.assert_called_with("player1")  # type: ignore[attr-defined]
2161        provider._ynison.update_playing_status.assert_awaited_once()
2162
2163    async def test_with_seek_offset(self) -> None:
2164        """Seek offset is added to byte-based progress."""
2165        provider = _make_provider()
2166        provider._actual_duration_ms = 200000
2167        provider._ynison = _mock_ynison()
2168
2169        byte_rate = 44100 * 2 * 2
2170        bytes_yielded = byte_rate * 2  # 2 seconds of audio
2171        seek_ms = 30000
2172
2173        await provider._sync_progress(seek_ms, bytes_yielded, "player1")
2174
2175        # 30000ms + 2000ms = 32000ms → 32s
2176        assert provider._stream_metadata.elapsed_time == 32
2177        assert provider._streaming_progress_ms == 32000
2178
2179    async def test_no_player_id_skips_trigger(self) -> None:
2180        """When player_id is None, does not trigger player update."""
2181        provider = _make_provider()
2182        provider._actual_duration_ms = 200000
2183        provider._ynison = _mock_ynison()
2184
2185        await provider._sync_progress(0, 0, None)
2186
2187        provider.mass.players.trigger_player_update.assert_not_called()  # type: ignore[attr-defined]
2188
2189
2190# ------------------------------------------------------------------
2191# _bytes_to_ms
2192# ------------------------------------------------------------------
2193
2194
2195class TestBytesToMs:
2196    """Tests for _bytes_to_ms."""
2197
2198    def test_16bit(self) -> None:
2199        """16-bit stereo 44100Hz: 176400 bytes = 1000ms."""
2200        provider = _make_provider()
2201        # Default format is 44100/16/2
2202        assert provider._bytes_to_ms(176400) == 1000
2203
2204    def test_24bit(self) -> None:
2205        """24-bit stereo 48000Hz: 288000 bytes = 1000ms."""
2206        provider = _make_provider()
2207        # Explicit 48000/24 format (decoupled from the no-hint floor constant)
2208        # so this exercises the byte→ms math, not whatever rate the floor uses.
2209        provider._normalized_format = make_pcm_format(
2210            {
2211                "content_type": ContentType.PCM_S24LE,
2212                "sample_rate": 48000,
2213                "bit_depth": 24,
2214                "channels": 2,
2215            }
2216        )
2217        assert provider._bytes_to_ms(288000) == 1000
2218
2219    def test_zero(self) -> None:
2220        """Zero bytes = zero milliseconds."""
2221        provider = _make_provider()
2222        assert provider._bytes_to_ms(0) == 0
2223
2224
2225# ------------------------------------------------------------------
2226# _get_stream_details_with_retry
2227# ------------------------------------------------------------------
2228
2229
2230@pytest.mark.asyncio
2231class TestGetStreamDetailsWithRetry:
2232    """Tests for _get_stream_details_with_retry."""
2233
2234    async def test_success_first_attempt(self) -> None:
2235        """Returns stream details on first try and caches result."""
2236        provider = _make_provider()
2237        mock_yp = MagicMock()
2238        sd = MagicMock()
2239        sd.expiration = 600
2240        sd.to_dict.return_value = {"track_id": "t1"}
2241        sd.data = {"url": "https://cdn.example.com/audio.mp3", "decryption_key": "abc"}
2242        mock_yp.get_stream_details = AsyncMock(return_value=sd)
2243        provider._yandex_provider = mock_yp
2244
2245        result = await provider._get_stream_details_with_retry("t1")
2246        assert result is sd
2247        mock_yp.get_stream_details.assert_awaited_once()
2248        # Verify cache.set was called with data field preserved
2249        provider.mass.cache.set.assert_awaited_once()  # type: ignore[attr-defined]
2250        cached_value = provider.mass.cache.set.call_args[0][1]  # type: ignore[attr-defined]
2251        assert cached_value["data"] == sd.data
2252
2253    async def test_cache_hit_skips_api(self) -> None:
2254        """Returns cached stream details without API call."""
2255        provider = _make_provider()
2256        cached_sd = MagicMock()
2257        cached_sd.expiration = 600
2258        provider.mass.cache.get = AsyncMock(return_value=cached_sd)  # type: ignore[method-assign]
2259        mock_yp = MagicMock()
2260        mock_yp.get_stream_details = AsyncMock()
2261        provider._yandex_provider = mock_yp
2262
2263        result = await provider._get_stream_details_with_retry("t1")
2264        assert result is cached_sd
2265        mock_yp.get_stream_details.assert_not_awaited()
2266
2267    async def test_retries_on_failure(self) -> None:
2268        """Retries on transient error, succeeds on second attempt."""
2269        provider = _make_provider()
2270        mock_yp = MagicMock()
2271        sd = MagicMock()
2272        sd.expiration = 600
2273        sd.to_dict.return_value = {"track_id": "t1"}
2274        mock_yp.get_stream_details = AsyncMock(side_effect=[RuntimeError("transient"), sd])
2275        provider._yandex_provider = mock_yp
2276
2277        with patch(
2278            "music_assistant.providers.yandex_ynison.provider.asyncio.sleep", new_callable=AsyncMock
2279        ):
2280            result = await provider._get_stream_details_with_retry("t1")
2281        assert result is sd
2282        assert mock_yp.get_stream_details.await_count == 2
2283
2284    async def test_raises_after_max_retries(self) -> None:
2285        """Raises RuntimeError after all retries exhausted."""
2286        provider = _make_provider()
2287        mock_yp = MagicMock()
2288        mock_yp.get_stream_details = AsyncMock(side_effect=RuntimeError("always fails"))
2289        provider._yandex_provider = mock_yp
2290
2291        with (
2292            patch(
2293                "music_assistant.providers.yandex_ynison.provider.asyncio.sleep",
2294                new_callable=AsyncMock,
2295            ),
2296            pytest.raises(RuntimeError, match="failed after"),
2297        ):
2298            await provider._get_stream_details_with_retry("t1")
2299        assert mock_yp.get_stream_details.await_count == _API_MAX_RETRIES
2300
2301    async def test_cancellation_not_retried(self) -> None:
2302        """CancelledError propagates immediately, no retry."""
2303        provider = _make_provider()
2304        mock_yp = MagicMock()
2305        mock_yp.get_stream_details = AsyncMock(side_effect=asyncio.CancelledError())
2306        provider._yandex_provider = mock_yp
2307
2308        with pytest.raises(asyncio.CancelledError):
2309            await provider._get_stream_details_with_retry("t1")
2310        mock_yp.get_stream_details.assert_awaited_once()
2311
2312    async def test_unloaded_provider_raises_login_failed_not_attribute_error(
2313        self,
2314    ) -> None:
2315        """
2316        Linked yandex_music unloaded → LoginFailed, not AttributeError.
2317
2318        Regression: _yandex_provider can be set to None by the background
2319        _check_yandex_provider_match task between awaits in this function.
2320        Prior code dereferenced `self._yandex_provider.get_stream_details`
2321        directly, raising AttributeError and hard-stopping the audio
2322        generator.  We now capture a local ref at entry and surface a
2323        clean LoginFailed instead.
2324        """
2325        provider = _make_provider()
2326        provider._yandex_provider = None
2327
2328        with pytest.raises(LoginFailed, match="not loaded"):
2329            await provider._get_stream_details_with_retry("t1")
2330
2331
2332# ------------------------------------------------------------------
2333# _advance_queue_index
2334# ------------------------------------------------------------------
2335
2336
2337class TestAdvanceQueueIndex:
2338    """Tests for _advance_queue_index."""
2339
2340    async def test_sends_state(self) -> None:
2341        """Advances queue index and sends new state."""
2342        provider = _make_provider()
2343        state = _make_ynison_state(
2344            current_playable_index=0,
2345            playable_list=[{"playable_id": "t1"}, {"playable_id": "t2"}],
2346        )
2347        mock_yn = _mock_ynison(state, device_id="own-device-id")
2348        provider._ynison = mock_yn
2349
2350        await provider._advance_queue_index(3)
2351
2352        mock_yn.update_player_state.assert_awaited_once()
2353        sent = mock_yn.update_player_state.call_args.kwargs["player_state"]
2354        assert sent["player_queue"]["current_playable_index"] == 3
2355        assert sent["status"]["progress_ms"] == "0"
2356        assert sent["status"]["duration_ms"] == "0"
2357        assert sent["status"]["paused"] is False
2358        # Outgoing state authored by our device_id, timestamps as strings.
2359        queue_version = sent["player_queue"]["version"]
2360        status_version = sent["status"]["version"]
2361        assert queue_version["device_id"] == "own-device-id"
2362        assert status_version["device_id"] == "own-device-id"
2363        assert isinstance(queue_version["version"], str)
2364        assert queue_version["timestamp_ms"] == "0"
2365        assert isinstance(status_version["version"], str)
2366        assert status_version["timestamp_ms"] == "0"
2367
2368    async def test_with_expanded_list(self) -> None:
2369        """Expanded list replaces playable_list in sent state."""
2370        provider = _make_provider()
2371        state = _make_ynison_state(
2372            playable_list=[{"playable_id": "t1"}],
2373        )
2374        mock_yn = _mock_ynison(state, device_id="own-device-id")
2375        provider._ynison = mock_yn
2376
2377        expanded = [{"playable_id": "t1"}, {"playable_id": "t2"}]
2378        await provider._advance_queue_index(1, expanded_list=expanded)
2379
2380        sent = mock_yn.update_player_state.call_args.kwargs["player_state"]
2381        assert sent["player_queue"]["playable_list"] == expanded
2382        assert sent["player_queue"]["version"]["device_id"] == "own-device-id"
2383
2384    async def test_not_connected_waits_then_sends(self) -> None:
2385        """Waits for reconnection before sending state."""
2386        provider = _make_provider()
2387        state = _make_ynison_state()
2388        mock_yn = _mock_ynison(state, connected=False)
2389        provider._ynison = mock_yn
2390
2391        call_count = 0
2392
2393        def _get_connected(_self: object) -> bool:
2394            nonlocal call_count
2395            call_count += 1
2396            # Reconnect after 2 checks
2397            return call_count > 2
2398
2399        type(mock_yn).connected = property(_get_connected)
2400
2401        with patch(
2402            "music_assistant.providers.yandex_ynison.provider.asyncio.sleep",
2403            new_callable=AsyncMock,
2404        ) as sleep:
2405            await provider._advance_queue_index(1)
2406
2407        assert sleep.await_count == 2
2408        mock_yn.update_player_state.assert_awaited_once()
2409
2410    async def test_timeout_no_send(self) -> None:
2411        """Gives up after timeout when Ynison stays disconnected."""
2412        provider = _make_provider()
2413        state = _make_ynison_state()
2414        mock_yn = _mock_ynison(state, connected=False)
2415        provider._ynison = mock_yn
2416
2417        # Patch asyncio.sleep to skip real waiting
2418        with patch("asyncio.sleep", new_callable=AsyncMock):
2419            await provider._advance_queue_index(1)
2420
2421        mock_yn.update_player_state.assert_not_called()
2422
2423    async def test_no_ynison_returns(self) -> None:
2424        """Returns immediately when _ynison is None."""
2425        provider = _make_provider()
2426        provider._ynison = None
2427
2428        await provider._advance_queue_index(1)
2429        # No crash, no calls
2430
2431
2432# ------------------------------------------------------------------
2433# _activate_playback
2434# ------------------------------------------------------------------
2435
2436
2437class TestActivatePlayback:
2438    """Tests for _activate_playback."""
2439
2440    async def test_selects_source_on_new_player(self) -> None:
2441        """Selects source on target player when not yet active."""
2442        provider = _make_provider()
2443        provider._active_player_id = None
2444
2445        player = MagicMock()
2446        player.player_id = "player1"
2447        player.display_name = "Player 1"
2448        player.state.playback_state = PlaybackState.IDLE
2449        provider.mass.players.all_players.return_value = [player]  # type: ignore[attr-defined]
2450        provider.mass.players.get_player.return_value = player  # type: ignore[attr-defined]
2451
2452        state = _make_ynison_state(progress_ms=0, paused=False)
2453
2454        await provider._activate_playback(state)
2455
2456        assert provider._active_player_id == "player1"
2457        provider.mass.create_task.assert_called()  # type: ignore[unreachable]
2458
2459    async def test_unpause_after_external_pause_fires_play_media(self) -> None:
2460        """
2461        Resume after pause schedules play_media for the (queue-id) player.
2462
2463        Simulates the post-`_pause_playback` state: `_stream_stop_event`
2464        set, `_active_player_id` already demoted to the queue id (the
2465        pause path's post-cmd_stop rewrite), `_in_use_by_queue` cleared
2466        by `on_source_unselected`. `_activate_playback` should fire
2467        `play_media(queue_id, audio_source.uri)` and clear the stop
2468        event so the next session is free to run.
2469        """
2470        provider = _make_provider()
2471        # Post-pause state: AriaCast path demoted _active_player_id to
2472        # the queue id (bare UUID) after cmd_stop succeeded.
2473        provider._stream_stop_event.set()
2474        provider._active_player_id = "player1"
2475        provider._in_use_by_queue = None
2476
2477        player = MagicMock()
2478        player.player_id = "player1"
2479        provider.mass.players.all_players.return_value = [player]
2480        provider.mass.players.get_player.return_value = player
2481
2482        play_media_calls = _arm_play_media_recorder(provider)
2483
2484        state = _make_ynison_state(progress_ms=10_000, paused=False)
2485
2486        await provider._activate_playback(state)
2487        await asyncio.sleep(0)  # let the scheduled play_media coro run
2488
2489        assert len(play_media_calls) == 1
2490        target_id, uri = play_media_calls[0]
2491        # Target must be the (queue id) bare player, NOT the bridge.
2492        # Bridge id has no queue → PlayerUnavailableError.
2493        assert target_id == "player1"
2494        assert uri == str(provider._audio_source.uri)
2495        assert not provider._stream_stop_event.is_set()
2496
2497    async def test_resume_via_externally_paused_flag_alone(self) -> None:
2498        """Resume fires play_media even if `_stream_stop_event` was cleared."""
2499        provider = _make_provider()
2500        provider._stream_stop_event.clear()
2501        provider._externally_paused = True
2502        provider._active_player_id = "player1"
2503        provider._in_use_by_queue = None
2504
2505        player = MagicMock()
2506        player.player_id = "player1"
2507        provider.mass.players.all_players.return_value = [player]
2508        provider.mass.players.get_player.return_value = player
2509
2510        play_media_calls = _arm_play_media_recorder(provider)
2511
2512        await provider._activate_playback(_make_ynison_state(progress_ms=10_000, paused=False))
2513        await asyncio.sleep(0)
2514
2515        assert play_media_calls, "play_media must fire even without stop event"
2516        assert provider._externally_paused is False  # cleared by _activate_playback
2517
2518    async def test_detects_track_change(self) -> None:
2519        """Detects track change and updates streaming track id."""
2520        provider = _make_provider()
2521        provider._current_streaming_track_id = "track1"
2522
2523        player = MagicMock()
2524        player.player_id = "player1"
2525        provider.mass.players.all_players.return_value = [player]  # type: ignore[attr-defined]
2526        provider.mass.players.get_player.return_value = player  # type: ignore[attr-defined]
2527        provider._active_player_id = "player1"
2528
2529        state = _make_ynison_state(
2530            progress_ms=0,
2531            paused=False,
2532            playable_list=[{"playable_id": "track2"}],
2533        )
2534
2535        await provider._activate_playback(state)
2536
2537        assert provider._current_streaming_track_id == "track2"
2538        assert provider._track_changed_event.is_set()
2539
2540    async def test_resume_after_pause(self) -> None:
2541        """Resume after pause triggers reselect and seek."""
2542        provider = _make_provider()
2543        provider._active_player_id = "player1"
2544        provider._current_streaming_track_id = "track1"
2545        provider._stream_stop_event.set()  # simulate paused
2546
2547        player = MagicMock()
2548        player.player_id = "player1"
2549        provider.mass.players.get_player.return_value = player  # type: ignore[attr-defined]
2550
2551        state = _make_ynison_state(
2552            progress_ms=50000,
2553            paused=False,
2554            playable_list=[{"playable_id": "track1"}],
2555        )
2556
2557        await provider._activate_playback(state)
2558
2559        assert provider._seek_position_ms == 50000
2560        assert provider._track_changed_event.is_set()
2561
2562    async def test_no_target_player_returns(self) -> None:
2563        """Returns early when no target player is available."""
2564        provider = _make_provider()
2565        provider.mass.players.all_players.return_value = []  # type: ignore[attr-defined]
2566        provider.mass.players.get_player.return_value = None  # type: ignore[attr-defined]
2567
2568        state = _make_ynison_state()
2569
2570        await provider._activate_playback(state)
2571
2572        assert provider._active_player_id is None
2573
2574
2575class TestInvalidateStreamCache:
2576    """Tests for _invalidate_stream_cache method."""
2577
2578    async def test_deletes_cache_entry(self) -> None:
2579        """_invalidate_stream_cache calls mass.cache.delete."""
2580        provider = _make_provider()
2581        provider.mass.cache = MagicMock()
2582        provider.mass.cache.delete = AsyncMock()
2583
2584        await provider._invalidate_stream_cache("track:42")
2585
2586        provider.mass.cache.delete.assert_called_once_with(
2587            "ynison_sd_track:42",
2588            provider=provider.instance_id,
2589        )
2590
2591
2592# ------------------------------------------------------------------
2593# Protocol invariants ported from the pre-AudioSource branch
2594# ------------------------------------------------------------------
2595
2596
2597def _make_active_state(track_id: str | None = "track42", *, paused: bool = False) -> YnisonState:
2598    """Build a YnisonState that reports our device as active and optionally playing."""
2599    state = YnisonState()
2600    state.active_device_id = "dev1"  # ours
2601    queue: dict[str, Any] = {"player_queue": {}}
2602    if track_id is not None:
2603        queue["player_queue"] = {
2604            "playable_list": [{"playable_id": track_id}],
2605            "current_playable_index": 0,
2606        }
2607    queue["status"] = {"paused": paused, "progress_ms": 0, "duration_ms": 60_000}
2608    state.player_state = queue
2609    return state
2610
2611
2612class TestPostReconnectSettleWindow:
2613    """The 2 s settle window after a WS reconnect drops the first inbound state."""
2614
2615    async def test_settle_window_skips_activate(self) -> None:
2616        """While in the settle window, _handle_ynison_state must not act."""
2617        provider = _make_provider()
2618        provider._device_id = "dev1"
2619        provider._ynison = MagicMock()
2620        provider._ynison.in_post_reconnect_settle = True
2621        activate_calls: list[Any] = []
2622
2623        async def _spy_activate(state: YnisonState) -> None:
2624            activate_calls.append(state)
2625
2626        _stub_attr(provider, "_activate_playback", _spy_activate)
2627
2628        await provider._handle_ynison_state(_make_active_state())
2629        assert activate_calls == []
2630
2631    async def test_no_settle_window_runs_activate(self) -> None:
2632        """Outside the settle window, normal handling resumes."""
2633        provider = _make_provider()
2634        provider._device_id = "dev1"
2635        provider._ynison = MagicMock()
2636        provider._ynison.in_post_reconnect_settle = False
2637        activate_calls: list[Any] = []
2638
2639        async def _spy_activate(state: YnisonState) -> None:
2640            activate_calls.append(state)
2641
2642        _stub_attr(provider, "_activate_playback", _spy_activate)
2643        # _maybe_prefetch is sync — patch with a no-op so we can spy on activate alone
2644        _stub_attr(provider, "_maybe_prefetch", MagicMock())
2645
2646        await provider._handle_ynison_state(_make_active_state())
2647        assert len(activate_calls) == 1
2648
2649
2650class TestIdempotencyTTL:
2651    """`_idempotent(action, key)` collapses duplicates within TTL."""
2652
2653    def test_second_call_within_ttl_returns_false(self) -> None:
2654        """Second call within `_COMMAND_IDEMPOTENCY_TTL` is suppressed."""
2655        provider = _make_provider()
2656        with patch(
2657            "music_assistant.providers.yandex_ynison.provider.time.monotonic",
2658            side_effect=[100.0, 100.5],
2659        ):
2660            assert provider._idempotent("on_pause", None) is True
2661            assert provider._idempotent("on_pause", None) is False
2662
2663    def test_call_past_ttl_returns_true(self) -> None:
2664        """A call past the TTL window passes again."""
2665        provider = _make_provider()
2666        gap = _COMMAND_IDEMPOTENCY_TTL + 0.1
2667        with patch(
2668            "music_assistant.providers.yandex_ynison.provider.time.monotonic",
2669            side_effect=[100.0, 100.0 + gap, 100.0 + gap],
2670        ):
2671            assert provider._idempotent("on_pause", None) is True
2672            assert provider._idempotent("on_pause", None) is True
2673
2674    def test_separate_keys_do_not_collide(self) -> None:
2675        """Different (action, key) tuples have independent debounce windows."""
2676        provider = _make_provider()
2677        with patch(
2678            "music_assistant.providers.yandex_ynison.provider.time.monotonic",
2679            side_effect=[100.0, 100.1],
2680        ):
2681            assert provider._idempotent("on_play", None) is True
2682            assert provider._idempotent("on_pause", None) is True
2683
2684    async def test_on_pause_double_call_sends_one_update(self) -> None:
2685        """Two _on_pause() within 1 s → exactly one update_playing_status."""
2686        provider = _make_provider()
2687        provider._ynison = MagicMock()
2688        provider._ynison.connected = True
2689        provider._ynison.state.progress_ms = 1000
2690        provider._ynison.state.duration_ms = 60000
2691        provider._ynison.update_playing_status = AsyncMock()
2692
2693        with patch(
2694            "music_assistant.providers.yandex_ynison.provider.time.monotonic",
2695            side_effect=[100.0, 100.5],
2696        ):
2697            await provider._on_pause()
2698            await provider._on_pause()
2699
2700        assert provider._ynison.update_playing_status.await_count == 1
2701
2702
2703class TestProgressClamp:
2704    """`_send_progress_to_ynison` clamps progress_ms to duration_ms."""
2705
2706    async def test_progress_clamped_to_duration(self) -> None:
2707        """progress_ms above duration_ms is clamped before WS send."""
2708        provider = _make_provider()
2709        provider._ynison = MagicMock()
2710        provider._ynison.connected = True
2711        provider._ynison.update_playing_status = AsyncMock()
2712
2713        await provider._send_progress_to_ynison(progress_ms=5000, duration_ms=4000, paused=False)
2714        called_kwargs = provider._ynison.update_playing_status.await_args.kwargs
2715        assert called_kwargs["progress_ms"] == 4000
2716        assert called_kwargs["duration_ms"] == 4000
2717
2718    async def test_progress_below_duration_passes_through(self) -> None:
2719        """Healthy progress values are forwarded unchanged."""
2720        provider = _make_provider()
2721        provider._ynison = MagicMock()
2722        provider._ynison.connected = True
2723        provider._ynison.update_playing_status = AsyncMock()
2724
2725        await provider._send_progress_to_ynison(progress_ms=3000, duration_ms=4000, paused=False)
2726        called_kwargs = provider._ynison.update_playing_status.await_args.kwargs
2727        assert called_kwargs["progress_ms"] == 3000
2728
2729
2730class TestDriftClassifier:
2731    """`_classify_drift` returns ignore / queue_rebuild / seek per heuristic."""
2732
2733    def test_drift_under_threshold_ignored(self) -> None:
2734        """Drift below the 3 s threshold is classified as `ignore`."""
2735        assert YandexYnisonProvider._classify_drift(60_500, 60_000) == "ignore"
2736
2737    def test_queue_rebuild_pattern(self) -> None:
2738        """Ynison near zero while we're deep into the track → queue-rebuild echo."""
2739        assert YandexYnisonProvider._classify_drift(0, 120_000) == "queue_rebuild"
2740        assert YandexYnisonProvider._classify_drift(500, 30_000) == "queue_rebuild"
2741
2742    def test_genuine_seek(self) -> None:
2743        """Large non-zero drift is treated as a genuine user seek."""
2744        assert YandexYnisonProvider._classify_drift(120_000, 60_000) == "seek"
2745
2746    def test_custom_threshold(self) -> None:
2747        """Custom threshold raises the bar for `ignore`."""
2748        assert YandexYnisonProvider._classify_drift(5_000, 0, threshold_ms=10_000) == "ignore"
2749
2750
2751class TestOnSourceUnselectedStaleRejection:
2752    """`on_source_unselected` rejects stale stream_session_id callbacks."""
2753
2754    async def test_stale_session_id_keeps_claim(self) -> None:
2755        """A teardown callback from a superseded session must not release the claim."""
2756        provider = _make_provider()
2757        provider._in_use_by_queue = "queue1"
2758        provider._active_session_id = "live"
2759
2760        await provider.on_source_unselected(AUDIO_SOURCE_ID, "queue1", "stale")
2761
2762        assert provider._in_use_by_queue == "queue1"
2763        assert provider._active_session_id == "live"
2764
2765    async def test_matching_session_id_releases_claim(self) -> None:
2766        """The live session id matches → lock and session id clear."""
2767        provider = _make_provider()
2768        provider._in_use_by_queue = "queue1"
2769        provider._active_session_id = "live"
2770
2771        await provider.on_source_unselected(AUDIO_SOURCE_ID, "queue1", "live")
2772
2773        assert provider._in_use_by_queue is None
2774        assert provider._active_session_id is None
2775
2776
2777class TestUpdateSourceCapabilitiesStamping:
2778    """`_update_source_capabilities` rebuilds and stamps the live queue item."""
2779
2780    def _setup_provider_with_queue(self) -> tuple[YandexYnisonProvider, MagicMock, MagicMock]:
2781        provider = _make_provider()
2782        # Linked yandex_music provider available → capabilities ON
2783        provider._yandex_provider = MagicMock()
2784        provider._in_use_by_queue = "queue1"
2785
2786        old_source = provider._audio_source
2787        queue = MagicMock()
2788        current_item = MagicMock()
2789        current_item.media_item = old_source
2790        queue.current_item = current_item
2791        provider.mass.player_queues.get = MagicMock(return_value=queue)
2792        provider.mass.player_queues.signal_update = MagicMock()
2793        provider.mass.players.trigger_player_update = MagicMock()
2794        return provider, queue, current_item
2795
2796    def test_stamps_new_audio_source_with_caps_on(self) -> None:
2797        """A capability flip stamps a fresh AudioSource onto the live queue item."""
2798        provider, _queue, current_item = self._setup_provider_with_queue()
2799        # Mark the old source's media_type explicitly so identity logic accepts it
2800        old = provider._audio_source
2801        current_item.media_item = AudioSource(
2802            item_id=old.item_id,
2803            provider=old.provider,
2804            name=old.name,
2805            provider_mappings=old.provider_mappings,
2806            can_play_pause=False,
2807            can_seek=False,
2808            can_next_previous=False,
2809            exclusive=True,
2810            allow_external_trigger=True,
2811        )
2812
2813        provider._update_source_capabilities()
2814
2815        new_source = current_item.media_item
2816        assert isinstance(new_source, AudioSource)
2817        assert new_source.can_play_pause is True
2818        assert new_source.can_seek is True
2819        assert new_source.can_next_previous is True
2820        provider.mass.player_queues.signal_update.assert_called_once_with(
2821            "queue1", items_changed=True
2822        )
2823
2824    def test_skips_when_in_use_by_queue_is_none(self) -> None:
2825        """When no queue currently consumes our stream, just rebuild and return."""
2826        provider, _queue, _current_item = self._setup_provider_with_queue()
2827        provider._in_use_by_queue = None
2828
2829        provider._update_source_capabilities()
2830
2831        # No stamp on the queue item.
2832        provider.mass.player_queues.signal_update.assert_not_called()
2833
2834    def test_skips_when_queue_lookup_returns_none(self) -> None:
2835        """Defensive: queue gone between claim and capability tick → no stamp."""
2836        provider, _queue, _current_item = self._setup_provider_with_queue()
2837        provider.mass.player_queues.get = MagicMock(return_value=None)
2838
2839        provider._update_source_capabilities()
2840
2841        provider.mass.player_queues.signal_update.assert_not_called()
2842
2843
2844class TestPrefetchOrdering:
2845    """Format pre-fetch must complete BEFORE `play_media` is queued."""
2846
2847    async def test_prefetch_called_before_play_media(self) -> None:
2848        """`_prefetch_format_for_track` is awaited before `play_media` is queued."""
2849        provider = _make_provider()
2850        provider._device_id = "dev1"
2851        # Force target_player resolution
2852        provider._default_player_id = "player1"
2853
2854        order: list[str] = []
2855
2856        async def _fake_prefetch(track_id: str) -> None:
2857            order.append(f"prefetch:{track_id}")
2858
2859        async def _fake_play_media(player_id: str, uri: str) -> None:
2860            order.append(f"play_media:{player_id}:{uri}")
2861
2862        _stub_attr(provider, "_prefetch_format_for_track", _fake_prefetch)
2863        provider.mass.player_queues.play_media = _fake_play_media
2864        # Replace create_task with sync await so ordering is observable
2865        provider.mass.create_task = lambda coro, *_a, **_kw: asyncio.get_event_loop().create_task(
2866            coro
2867        )
2868        # Player resolution returns our target
2869        provider.mass.players.all_players = MagicMock(return_value=[])
2870        _stub_attr(
2871            provider,
2872            "_get_target_player_id",
2873            MagicMock(return_value="player1"),
2874        )
2875
2876        await provider._activate_playback(_make_active_state())
2877        # Let the create_task scheduled coro finish.
2878        await asyncio.sleep(0)
2879
2880        assert order, "expected at least a prefetch call"
2881        assert order[0].startswith("prefetch:track42")
2882        assert any(c.startswith("play_media:player1") for c in order[1:])
2883
2884
2885class TestPrefetchFlowsThroughToStreamDetails:
2886    """
2887    `get_stream_details` returns the *prefetched* AudioFormat.
2888
2889    Pins the contract that MA's upstream passthrough path (#3969,
2890    `_select_audio_source_pcm_format`) honors: MA reads
2891    ``streamdetails.audio_format`` and only invokes ffmpeg when it
2892    cannot match the player's supported rates. A regression that
2893    decouples ``_prefetch_format_for_track`` from
2894    ``self._normalized_params`` (or that returns a stale snapshot in
2895    ``get_stream_details``) would silently downgrade hi-res passthrough
2896    to a forced ffmpeg resample with no functional indicator beyond
2897    log entropy.
2898    """
2899
2900    async def test_prefetch_updates_streamdetails_audio_format(self) -> None:
2901        """Prefetched source rate/bit-depth must reach `get_stream_details`."""
2902        from music_assistant_models.enums import MediaType  # noqa: PLC0415
2903
2904        provider = _make_provider()
2905        # Default before prefetch: lossy PCM (16-bit / 44.1 kHz auto base).
2906        default_rate = provider._normalized_params["sample_rate"]
2907        assert default_rate != 96_000  # sanity: ensure we'll see a change
2908
2909        mock_yandex = MagicMock()
2910
2911        async def _fake_get_stream_details(_track_id: str, _media_type: MediaType) -> Any:
2912            sd = MagicMock()
2913            sd.expiration = 60
2914            sd.duration = 200
2915            sd.data = None
2916            sd.audio_format = AudioFormat(
2917                content_type=ContentType.FLAC,
2918                sample_rate=96_000,
2919                bit_depth=24,
2920                channels=2,
2921            )
2922            sd.to_dict = MagicMock(return_value={})
2923            return sd
2924
2925        mock_yandex.get_stream_details = AsyncMock(side_effect=_fake_get_stream_details)
2926        provider._yandex_provider = mock_yandex
2927        # Set explicit AUTO so prefetch hint is allowed to promote both axes.
2928        provider._cfg_sample_rate = OUTPUT_AUTO
2929        provider._cfg_bit_depth = OUTPUT_AUTO
2930
2931        await provider._prefetch_format_for_track("track42")
2932
2933        # `_normalized_params` lifted to source rate/bit-depth.
2934        assert provider._normalized_params["sample_rate"] == 96_000
2935        assert provider._normalized_params["bit_depth"] == 24
2936        # And get_stream_details now reflects that — MA's
2937        # `_select_audio_source_pcm_format` consumes this.
2938        sd = await provider.get_stream_details("main", MediaType.AUDIO_SOURCE)
2939        assert sd.media_type == MediaType.AUDIO_SOURCE
2940        assert sd.audio_format.sample_rate == 96_000
2941        assert sd.audio_format.bit_depth == 24
2942        assert sd.audio_format.channels == 2
2943
2944    async def test_streamdetails_audio_format_is_fresh_copy_per_call(self) -> None:
2945        """
2946        Each `get_stream_details` returns a fresh AudioFormat instance.
2947
2948        `AudioFormat` is mutable (MA's outer ffmpeg sets `codec_type` in
2949        place). A shared instance across `get_stream_details` calls
2950        would let one consumer's mutation poison the next one's
2951        snapshot — which #3969's passthrough specifically depends on
2952        for the format-match comparison.
2953        """
2954        from music_assistant_models.enums import MediaType  # noqa: PLC0415
2955
2956        provider = _make_provider()
2957        sd1 = await provider.get_stream_details("main", MediaType.AUDIO_SOURCE)
2958        sd2 = await provider.get_stream_details("main", MediaType.AUDIO_SOURCE)
2959
2960        assert sd1.media_type == MediaType.AUDIO_SOURCE
2961        assert sd1.audio_format == sd2.audio_format  # value-equal
2962        assert sd1.audio_format is not sd2.audio_format  # not the same instance
2963
2964
2965class TestAudioStreamPausedReturn:
2966    """`get_audio_stream` exits immediately when Ynison reports paused."""
2967
2968    async def test_paused_state_exits_generator_without_yielding(self) -> None:
2969        """A paused-at-entry session yields zero chunks and the generator ends."""
2970        provider = _make_provider()
2971        provider._yandex_provider = MagicMock()
2972        provider._in_use_by_queue = "player1"
2973        provider._active_session_id = "session-1"
2974
2975        ynison = MagicMock()
2976        ynison.state.is_paused = True
2977        ynison.state.current_track_id = "track42"
2978        provider._ynison = ynison
2979
2980        streamdetails = MagicMock()
2981        gen = provider.get_audio_stream(streamdetails, seek_position=0)
2982
2983        chunks: list[bytes] = []
2984        with suppress(StopAsyncIteration):
2985            async for chunk in gen:
2986                chunks.append(chunk)
2987
2988        assert chunks == []
2989
2990
2991class TestNaturalEndDifferentiation:
2992    """
2993    `_signal_track_completion` fires only on clean iterator exhaustion.
2994
2995    These tests exercise the post-inner-loop branch via
2996    ``_wait_for_track_change`` as the outer-loop gate. The previous
2997    iteration set ``_stream_stop_event`` between yields, which made the
2998    stop-event guard above ``natural_end`` short-circuit the very logic
2999    we wanted to verify — the tests passed for the wrong reason. The
3000    rewritten infra lets ``natural_end`` actually evaluate and uses the
3001    ``_wait_for_track_change`` stub to terminate the outer loop.
3002    """
3003
3004    def _build(self) -> YandexYnisonProvider:
3005        provider = _make_provider()
3006        provider._yandex_provider = MagicMock()
3007        provider._in_use_by_queue = "player1"
3008        provider._active_session_id = "session-1"
3009        ynison = MagicMock()
3010        ynison.connected = True
3011        ynison.state.is_paused = False
3012        ynison.state.current_track_id = "track42"
3013        ynison.state.player_state = {"status": {"paused": False}}
3014        ynison.update_playing_status = AsyncMock()
3015        provider._ynison = ynison
3016        return provider
3017
3018    @staticmethod
3019    def _spy_signal(provider: YandexYnisonProvider) -> list[int]:
3020        calls: list[int] = []
3021
3022        async def _spy() -> None:
3023            calls.append(1)
3024
3025        _stub_attr(provider, "_signal_track_completion", _spy)
3026        return calls
3027
3028    @staticmethod
3029    def _gate_outer_loop_after_signal(provider: YandexYnisonProvider) -> None:
3030        """
3031        `_wait_for_track_change` returns False → outer loop exits.
3032
3033        ``natural_end`` calls `_wait_for_track_change`; we use its
3034        return as the gate so the test terminates AFTER natural_end
3035        has had a chance to evaluate. For non-natural-end paths
3036        (track-change / session-change), `_wait_for_track_change` is
3037        never called — we gate those via `_stream_stop_event.set()`
3038        inside the chunk-yield stub, AFTER one full outer iteration.
3039        """
3040
3041        async def _wait_false(_old: str, timeout: float = 30.0) -> bool:  # noqa: ARG001
3042            provider._stream_stop_event.set()
3043            return False
3044
3045        _stub_attr(provider, "_wait_for_track_change", _wait_false)
3046
3047    @staticmethod
3048    async def _drive_to_exhaustion(provider: YandexYnisonProvider) -> None:
3049        streamdetails = MagicMock()
3050        gen = provider.get_audio_stream(streamdetails, seek_position=0)
3051        try:
3052            with suppress(StopAsyncIteration):
3053                async for _ in gen:
3054                    pass
3055        finally:
3056            with suppress(StopAsyncIteration, asyncio.CancelledError):
3057                await gen.aclose()
3058
3059    async def test_clean_exhaustion_signals_completion(self) -> None:
3060        """Inner loop exhausts naturally → `_signal_track_completion` fires once."""
3061        provider = self._build()
3062        calls = self._spy_signal(provider)
3063        self._gate_outer_loop_after_signal(provider)
3064
3065        async def _natural_end(
3066            _track_id: str, *, seek_ms: int = 0, session_params: dict[str, Any] | None = None
3067        ) -> Any:
3068            del seek_ms, session_params  # signature-compat with _stream_track
3069            yield b"\x00\x00\x00\x00"
3070            # generator exhausts cleanly — no break flag set
3071
3072        _stub_attr(provider, "_stream_track", _natural_end)
3073
3074        await self._drive_to_exhaustion(provider)
3075
3076        assert calls == [1]
3077
3078    async def test_track_change_during_chunk_loop_suppresses_signal(self) -> None:
3079        """
3080        `_track_changed_event` set mid-stream → natural_end False → no signal.
3081
3082        Uses a two-invocation stub: first call arms `_track_changed_event`
3083        (the natural_end check we want to verify), second call sets
3084        `_stream_stop_event` to terminate the test. If we set the stop
3085        event in the first call, the stop-event guard above natural_end
3086        would short-circuit before the differentiation logic runs.
3087        """
3088        provider = self._build()
3089        calls = self._spy_signal(provider)
3090
3091        invocation_count = 0
3092
3093        async def _two_pass_stream(
3094            _track_id: str, *, seek_ms: int = 0, session_params: dict[str, Any] | None = None
3095        ) -> Any:
3096            del seek_ms, session_params  # signature-compat with _stream_track
3097            nonlocal invocation_count
3098            invocation_count += 1
3099            yield b"\x00\x00\x00\x00"
3100            if invocation_count == 1:
3101                # First pass: arm the break flag. natural_end will see it
3102                # and (correctly) suppress the signal. Outer loop re-iterates.
3103                provider._track_changed_event.set()
3104            else:
3105                # Second pass: stop the test.
3106                provider._stream_stop_event.set()
3107
3108        _stub_attr(provider, "_stream_track", _two_pass_stream)
3109
3110        await self._drive_to_exhaustion(provider)
3111
3112        assert calls == []
3113        assert invocation_count == 2  # natural_end must have evaluated on pass 1
3114
3115    async def test_session_change_during_chunk_loop_suppresses_signal(self) -> None:
3116        """
3117        Session-id rotation mid-stream → `broke_for_session_change` → no signal.
3118
3119        The session-mismatch breaks both the inner chunk loop's break
3120        guard AND the outer-loop's session check, so the generator
3121        exits after one iteration without further help — `natural_end`
3122        runs once with `broke_for_session_change=True` and must not
3123        signal.
3124        """
3125        provider = self._build()
3126        calls = self._spy_signal(provider)
3127
3128        async def _yield_then_rotate_session(
3129            _track_id: str, *, seek_ms: int = 0, session_params: dict[str, Any] | None = None
3130        ) -> Any:
3131            del seek_ms, session_params  # signature-compat with _stream_track
3132            yield b"\x00\x00\x00\x00"
3133            provider._active_session_id = "different-session"
3134
3135        _stub_attr(provider, "_stream_track", _yield_then_rotate_session)
3136
3137        await self._drive_to_exhaustion(provider)
3138
3139        assert calls == []
3140
3141    # NOTE: `broke_for_pause` inside natural_end is defensive: in
3142    # practice every external pause routes through `_pause_playback`,
3143    # which sets `_stream_stop_event` BEFORE the chunk loop can reach
3144    # the natural_end check (the stop-event guard at the top of
3145    # `get_audio_stream`'s outer loop short-circuits first). There is
3146    # no production code path that lands at natural_end with
3147    # `is_paused=True` and `_stream_stop_event=False`, so the clause
3148    # cannot be exercised by a black-box test. It survives as
3149    # belt-and-braces; intentionally untested.
3150
3151
3152class TestBypassThrottlerScope:
3153    """`BYPASS_THROTTLER` is set inside `_stream_track`, NOT inside prefetch."""
3154
3155    async def test_bypass_active_during_in_flight_stream_fetch(self) -> None:
3156        """The in-flight stream-details fetch runs with BYPASS_THROTTLER=True."""
3157        provider = _make_provider()
3158        observed: list[bool] = []
3159        mock_yandex = MagicMock()
3160
3161        async def _fake_get_stream_details(_track_id: str, _media_type: Any) -> Any:
3162            observed.append(BYPASS_THROTTLER.get())
3163            sd = MagicMock()
3164            sd.expiration = 0
3165            sd.duration = 1
3166            sd.audio_format = MagicMock()
3167            return sd
3168
3169        mock_yandex.get_stream_details = AsyncMock(side_effect=_fake_get_stream_details)
3170
3171        async def _fake_audio(_details: object) -> Any:
3172            yield b"x"
3173
3174        mock_yandex.get_audio_stream = _fake_audio
3175        provider._yandex_provider = mock_yandex
3176
3177        mock_ynison = MagicMock()
3178        mock_ynison.update_playing_status = AsyncMock()
3179        mock_ynison.state.is_paused = False
3180        provider._ynison = mock_ynison
3181
3182        async def _fake_ffmpeg(**_kwargs: object) -> Any:
3183            yield b"pcm"
3184
3185        with patch(
3186            "music_assistant.providers.yandex_ynison.provider.get_ffmpeg_stream",
3187            side_effect=_fake_ffmpeg,
3188        ):
3189            gen = provider._stream_track("track1")
3190            try:
3191                async for _ in gen:
3192                    break
3193            finally:
3194                await gen.aclose()
3195
3196        assert observed == [True], (
3197            f"BYPASS_THROTTLER should be True inside _stream_track, got {observed}"
3198        )
3199        assert BYPASS_THROTTLER.get() is False
3200
3201    async def test_bypass_not_active_during_prefetch(self) -> None:
3202        """The prefetch path is intentionally NOT bypassed — opportunistic only."""
3203        provider = _make_provider()
3204        observed: list[bool] = []
3205        mock_yandex = MagicMock()
3206
3207        async def _fake_get_stream_details(_track_id: str, _media_type: Any) -> Any:
3208            observed.append(BYPASS_THROTTLER.get())
3209            sd = MagicMock()
3210            sd.expiration = 0
3211            sd.audio_format = MagicMock()
3212            sd.audio_format.sample_rate = 44100
3213            sd.audio_format.bit_depth = 16
3214            return sd
3215
3216        mock_yandex.get_stream_details = AsyncMock(side_effect=_fake_get_stream_details)
3217        provider._yandex_provider = mock_yandex
3218
3219        await provider._prefetch_format_for_track("track1")
3220
3221        assert observed == [False], (
3222            f"BYPASS_THROTTLER must NOT be set during prefetch, got {observed}"
3223        )
3224
3225    async def test_bypass_resets_on_exception(self) -> None:
3226        """A raise inside the bypassed call must still reset the context-var."""
3227        provider = _make_provider()
3228        mock_yandex = MagicMock()
3229        mock_yandex.get_stream_details = AsyncMock(side_effect=RuntimeError("boom"))
3230        provider._yandex_provider = mock_yandex
3231
3232        mock_ynison = MagicMock()
3233        mock_ynison.state.is_paused = False
3234        provider._ynison = mock_ynison
3235
3236        # _stream_track swallows the exception, sets the stop event, returns.
3237        # Patch asyncio.sleep so the inner retry-with-backoff (2s + 4s) does
3238        # not block the test in real time.
3239        with patch(
3240            "music_assistant.providers.yandex_ynison.provider.asyncio.sleep", new=AsyncMock()
3241        ):
3242            async for _ in provider._stream_track("track1"):
3243                pass
3244
3245        assert BYPASS_THROTTLER.get() is False
3246
3247
3248# ------------------------------------------------------------------
3249# Music-token cache (spec 0004)
3250# ------------------------------------------------------------------
3251
3252
3253class TestMusicTokenCache:
3254    """Tests for the in-memory cache around `refresh_music_token`."""
3255
3256    @staticmethod
3257    def _own_provider_with_x_token(x_token: str = "xtok-1") -> YandexYnisonProvider:  # noqa: S107 — test fixture value
3258        """Construct an own-mode provider whose x_token drives refresh."""
3259        provider = _make_provider()
3260        provider._ym_instance_id = None
3261        provider.config = _make_mock_config(
3262            {CONF_TOKEN: None, CONF_X_TOKEN: x_token, CONF_YM_INSTANCE: YM_INSTANCE_OWN}
3263        )
3264        return provider
3265
3266    async def test_resolve_token_caches_x_token_refresh(self) -> None:
3267        """Second `_resolve_token` call within TTL is a cache hit."""
3268        provider = self._own_provider_with_x_token("xtok-1")
3269
3270        with patch(
3271            "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
3272            new_callable=AsyncMock,
3273            return_value=SecretStr("music-tok"),
3274        ) as mock_refresh:
3275            t1 = await provider._resolve_token()
3276            t2 = await provider._resolve_token()
3277
3278        assert t1.get_secret() == "music-tok"
3279        assert t2.get_secret() == "music-tok"
3280        assert mock_refresh.await_count == 1
3281
3282    async def test_resolve_token_refreshes_after_ttl_expires(self) -> None:
3283        """Time advancing past the TTL forces a fresh refresh."""
3284        provider = self._own_provider_with_x_token("xtok-1")
3285
3286        clock = {"now": 1000.0}
3287        provider._now = lambda: clock["now"]
3288
3289        with patch(
3290            "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
3291            new_callable=AsyncMock,
3292            return_value=SecretStr("music-tok"),
3293        ) as mock_refresh:
3294            await provider._resolve_token()  # cold miss
3295            clock["now"] += 60 * 60  # +60 min, past 50-min TTL
3296            await provider._resolve_token()  # must refresh
3297
3298        assert mock_refresh.await_count == 2
3299
3300    async def test_refresh_ynison_token_invalidates_cache(self) -> None:
3301        """A 401-driven refresh must bypass + drop the cached entry."""
3302        provider = self._own_provider_with_x_token("xtok-1")
3303
3304        with patch(
3305            "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
3306            new_callable=AsyncMock,
3307            return_value=SecretStr("music-tok"),
3308        ) as mock_refresh:
3309            # Warm the cache.
3310            await provider._resolve_token()
3311            assert mock_refresh.await_count == 1
3312
3313            # A 401 reconnect triggers the refresh path. The previously
3314            # cached token is provably stale and must be bypassed.
3315            mock_refresh.return_value = SecretStr("music-tok-2")
3316            result = await provider._refresh_ynison_token()
3317
3318        assert result.get_secret() == "music-tok-2"
3319        assert mock_refresh.await_count == 2
3320
3321        # The follow-up resolve uses the new cached value, not a third refresh.
3322        with patch(
3323            "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
3324            new_callable=AsyncMock,
3325        ) as mock_refresh_after:
3326            after = await provider._resolve_token()
3327            assert after.get_secret() == "music-tok-2"
3328            mock_refresh_after.assert_not_called()
3329
3330    async def test_concurrent_resolve_token_calls_refresh_once(self) -> None:
3331        """Two concurrent `_resolve_token` calls coalesce into one refresh."""
3332        provider = self._own_provider_with_x_token("xtok-1")
3333
3334        refresh_started = asyncio.Event()
3335        refresh_release = asyncio.Event()
3336
3337        async def slow_refresh(_x_token: SecretStr) -> SecretStr:
3338            refresh_started.set()
3339            await refresh_release.wait()
3340            return SecretStr("music-tok")
3341
3342        with patch(
3343            "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
3344            side_effect=slow_refresh,
3345        ) as mock_refresh:
3346            task_a = asyncio.create_task(provider._resolve_token())
3347            task_b = asyncio.create_task(provider._resolve_token())
3348            await refresh_started.wait()
3349            refresh_release.set()
3350            r_a, r_b = await asyncio.gather(task_a, task_b)
3351
3352        assert r_a.get_secret() == "music-tok"
3353        assert r_b.get_secret() == "music-tok"
3354        assert mock_refresh.await_count == 1
3355
3356    async def test_cache_lru_evicts_oldest_after_four_x_tokens(self) -> None:
3357        """When a 5th distinct x_token arrives, the oldest entry is evicted."""
3358        provider = self._own_provider_with_x_token("xtok-1")
3359
3360        async def fake_refresh(x_token: SecretStr) -> SecretStr:
3361            return SecretStr(f"music-for-{x_token.get_secret()}")
3362
3363        with patch(
3364            "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
3365            side_effect=fake_refresh,
3366        ):
3367            for i in range(1, 6):
3368                provider.config = _make_mock_config(
3369                    {
3370                        CONF_TOKEN: None,
3371                        CONF_X_TOKEN: f"xtok-{i}",
3372                        CONF_YM_INSTANCE: YM_INSTANCE_OWN,
3373                    }
3374                )
3375                await provider._resolve_token()
3376
3377        import hashlib  # noqa: PLC0415 — test-local
3378
3379        # 4-entry LRU after 5 distinct keys → oldest evicted, newest retained.
3380        assert len(provider._token_cache) == 4
3381        gone = hashlib.sha256(b"xtok-1").hexdigest()
3382        still_here = hashlib.sha256(b"xtok-5").hexdigest()
3383        assert gone not in provider._token_cache
3384        assert still_here in provider._token_cache
3385
3386    async def test_cache_does_not_log_secrets_or_hashes(
3387        self, caplog: pytest.LogCaptureFixture
3388    ) -> None:
3389        """No log record may contain the x_token, the music token, or its hash."""
3390        provider = self._own_provider_with_x_token("xtok-secret")
3391
3392        with (
3393            caplog.at_level("DEBUG"),
3394            patch(
3395                "music_assistant.providers.yandex_ynison.provider.refresh_music_token",
3396                new_callable=AsyncMock,
3397                return_value=SecretStr("music-secret"),
3398            ),
3399        ):
3400            await provider._resolve_token()
3401            await provider._refresh_ynison_token()
3402
3403        import hashlib  # noqa: PLC0415 — test-local
3404
3405        forbidden = (
3406            "xtok-secret",
3407            "music-secret",
3408            hashlib.sha256(b"xtok-secret").hexdigest(),
3409        )
3410        for record in caplog.records:
3411            blob = record.getMessage()
3412            for needle in forbidden:
3413                assert needle not in blob, f"Credential leaked: {needle!r} in {blob!r}"
3414
3415
3416# ------------------------------------------------------------------
3417# Strict-mode delivery signalling (spec 0003)
3418# ------------------------------------------------------------------
3419
3420
3421class TestStrictModeDeliverySignal:
3422    """Tests for `_send`/`update_*` strict-mode propagation in `provider.py`."""
3423
3424    async def test_on_play_raises_player_command_failed_when_send_fails(self) -> None:
3425        """`_on_play` translates `YnisonSendError` into `PlayerCommandFailed`."""
3426        provider = _make_provider()
3427        provider._actual_duration_ms = 120000
3428        state = _make_ynison_state(progress_ms=5000, duration_ms=120000, paused=True)
3429        mock_yn = _mock_ynison(state)
3430        mock_yn.update_playing_status = AsyncMock(side_effect=YnisonSendError("ws down"))
3431        provider._ynison = mock_yn
3432
3433        with pytest.raises(PlayerCommandFailed):
3434            await provider._on_play()
3435
3436    async def test_on_pause_raises_player_command_failed_when_send_fails(self) -> None:
3437        """`_on_pause` translates `YnisonSendError` into `PlayerCommandFailed`."""
3438        provider = _make_provider()
3439        provider._actual_duration_ms = 120000
3440        state = _make_ynison_state(progress_ms=5000, duration_ms=120000, paused=False)
3441        mock_yn = _mock_ynison(state)
3442        mock_yn.update_playing_status = AsyncMock(side_effect=YnisonSendError("ws down"))
3443        provider._ynison = mock_yn
3444
3445        with pytest.raises(PlayerCommandFailed):
3446            await provider._on_pause()
3447
3448    async def test_on_seek_raises_player_command_failed_when_send_fails(self) -> None:
3449        """`_on_seek` raises `PlayerCommandFailed` and leaves local seek state untouched."""
3450        provider = _make_provider()
3451        provider._actual_duration_ms = 200000
3452        provider._seek_position_ms = 0
3453        provider._track_changed_event.clear()
3454        state = _make_ynison_state(progress_ms=5000, duration_ms=200000, paused=False)
3455        mock_yn = _mock_ynison(state)
3456        mock_yn.update_playing_status = AsyncMock(side_effect=YnisonSendError("ws down"))
3457        provider._ynison = mock_yn
3458
3459        with pytest.raises(PlayerCommandFailed):
3460            await provider._on_seek(30)  # 30 seconds
3461
3462        # Local seek state must NOT be advanced past a send that never landed.
3463        assert provider._seek_position_ms == 0
3464        assert not provider._track_changed_event.is_set()
3465
3466    async def test_signal_track_completion_logs_on_send_failure(
3467        self, caplog: pytest.LogCaptureFixture
3468    ) -> None:
3469        """`_signal_track_completion` swallows `YnisonSendError` and logs a warning."""
3470        provider = _make_provider()
3471        state = YnisonState(
3472            active_device_id=provider._device_id,
3473            player_state={
3474                "status": {"paused": False, "progress_ms": 180000, "duration_ms": 200000},
3475                "player_queue": {
3476                    "current_playable_index": 0,
3477                    "playable_list": [{"playable_id": "t1"}, {"playable_id": "t2"}],
3478                    "entity_type": "PLAYLIST",
3479                },
3480            },
3481        )
3482        mock_ynison = MagicMock()
3483        mock_ynison.state = state
3484        mock_ynison.connected = True
3485        mock_ynison.device_id = provider._device_id
3486        mock_ynison.update_playing_status = AsyncMock(side_effect=YnisonSendError("ws down"))
3487        mock_ynison.update_player_state = AsyncMock()
3488        provider._ynison = mock_ynison
3489
3490        # Must not raise — end-of-track has no command to fail back to.
3491        with caplog.at_level("WARNING"):
3492            await provider._signal_track_completion()
3493        assert any("Track-completion signal dropped" in r.message for r in caplog.records)
3494
3495    async def test_advance_queue_index_returns_on_send_failure(
3496        self, caplog: pytest.LogCaptureFixture
3497    ) -> None:
3498        """`_advance_queue_index` swallows `YnisonSendError` and logs a warning."""
3499        provider = _make_provider()
3500        state = _make_ynison_state(
3501            current_playable_index=0,
3502            playable_list=[{"playable_id": "t1"}, {"playable_id": "t2"}],
3503        )
3504        mock_yn = _mock_ynison(state)
3505        mock_yn.update_player_state = AsyncMock(side_effect=YnisonSendError("ws down"))
3506        provider._ynison = mock_yn
3507
3508        with caplog.at_level("WARNING"):
3509            await provider._advance_queue_index(1)
3510        assert any("Queue-advance dropped" in r.message for r in caplog.records)
3511
3512    async def test_sync_progress_uses_non_strict_send(self) -> None:
3513        """
3514        `_sync_progress` heartbeat must call into the non-strict send path.
3515
3516        Regression guard: heartbeats stay fire-and-forget so a single bad
3517        send tick does not crash the streaming generator. We verify the
3518        forwarded `strict=False` kwarg rather than the bubble-up
3519        behaviour (which is the contract of the underlying `_send`
3520        already covered in `tests/test_ynison_client.py`).
3521        """
3522        provider = _make_provider()
3523        provider._actual_duration_ms = 200000
3524        provider._stream_metadata.duration = 200
3525        state = _make_ynison_state(progress_ms=5000, duration_ms=200000, paused=False)
3526        mock_yn = _mock_ynison(state)
3527        provider._ynison = mock_yn
3528
3529        await provider._sync_progress(seek_ms=0, bytes_yielded=0, player_id=None)
3530
3531        mock_yn.update_playing_status.assert_awaited_once()
3532        _args, kwargs = mock_yn.update_playing_status.call_args
3533        assert kwargs.get("strict", False) is False
3534
3535    async def test_send_progress_strict_raises_when_not_connected(self) -> None:
3536        """`_send_progress_to_ynison(strict=True)` raises when Ynison is disconnected."""
3537        provider = _make_provider()
3538        provider._ynison = _mock_ynison(connected=False)
3539
3540        with pytest.raises(YnisonSendError):
3541            await provider._send_progress_to_ynison(
3542                progress_ms=1000, duration_ms=2000, paused=False, strict=True
3543            )
3544
3545    async def test_send_progress_non_strict_silent_when_not_connected(self) -> None:
3546        """`_send_progress_to_ynison` default behaviour stays silent when disconnected."""
3547        provider = _make_provider()
3548        provider._ynison = _mock_ynison(connected=False)
3549
3550        # Must not raise
3551        await provider._send_progress_to_ynison(progress_ms=1000, duration_ms=2000, paused=False)
3552
3553
3554# ------------------------------------------------------------------
3555# Connected-Ynison guard helper (spec 0005)
3556# ------------------------------------------------------------------
3557
3558
3559class TestRequireConnectedYnison:
3560    """Tests for the extracted `_require_connected_ynison` helper."""
3561
3562    def test_raises_unsupported_when_client_missing(self) -> None:
3563        """`_ynison is None` raises `UnsupportedFeaturedException`."""
3564        provider = _make_provider()
3565        provider._ynison = None
3566
3567        with pytest.raises(UnsupportedFeaturedException, match="not initialized"):
3568            provider._require_connected_ynison()
3569
3570    def test_raises_command_failed_when_disconnected(self) -> None:
3571        """`_ynison.connected is False` raises `PlayerCommandFailed`."""
3572        provider = _make_provider()
3573        provider._ynison = _mock_ynison(connected=False)
3574
3575        with pytest.raises(PlayerCommandFailed, match="disconnected"):
3576            provider._require_connected_ynison()
3577
3578    def test_returns_client_when_ready(self) -> None:
3579        """Happy path returns the live client unchanged."""
3580        provider = _make_provider()
3581        mock_yn = _mock_ynison(connected=True)
3582        provider._ynison = mock_yn
3583
3584        result = provider._require_connected_ynison()
3585        assert result is mock_yn
3586