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