music-assistant-server

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