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