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