music-assistant-server

1"""
2Unit tests for the Spotify Soloist playback backend.
3
4The backend runs one continuous soloist session, feeds it one track ahead and
5splits the captured PCM into per-item streams. These tests lock down the pure
6logic around that: lead-silence trimming, the item channels and where the
7session cuts between them, event handling, the crossfade handed to the engine,
8feeding the follower, completeness validation, paired-session adoption and
9setup. No real process or PulseAudio is involved.
10"""
11
12from __future__ import annotations
13
14import asyncio
15import os
16import time
17from collections.abc import AsyncGenerator, Callable, Iterator
18from contextlib import contextmanager, suppress
19from pathlib import Path
20from typing import Any, cast
21from unittest.mock import AsyncMock, MagicMock, patch
22
23import pytest
24from music_assistant_models.enums import ContentType, MediaType
25from music_assistant_models.errors import AudioError, LoginFailed
26from music_assistant_models.media_items import AudioFormat
27from music_assistant_models.streamdetails import StreamDetails
28
29from music_assistant.helpers.pulse_capture import CAPTURE_SAMPLE_RATE
30from music_assistant.models.music_provider import ProviderStreamLimitError
31from music_assistant.providers.spotify.backends import soloist as soloist_backend
32from music_assistant.providers.spotify.backends.soloist import (
33    _BYTES_PER_SECOND,
34    _FRAME_BYTES,
35    _IDLE_TIMEOUT_S,
36    _MAX_APP_PAUSE_RESUMES,
37    _MAX_LEAD_TRIM_S,
38    _READ_CHUNK_SIZE,
39    SoloistAppControl,
40    SoloistAppControlError,
41    SoloistBackend,
42    _CaptureShaper,
43    _ItemAudio,
44    _SoloistSession,
45    _trim_lead_silence,
46)
47from music_assistant.providers.spotify.constants import (
48    CONF_SOLOIST_API_KEY,
49    CONF_SOLOIST_CONSENT,
50    CONF_SOLOIST_SESSION_DIR,
51    SOLOIST_DATA_DIR_NAME,
52    SOLOIST_DEVICE_NAME,
53)
54from music_assistant.providers.spotify.helpers import soloist_session_present
55from music_assistant.providers.spotify.provider import SpotifyProvider
56from music_assistant.providers.spotify_connect.provider import DEFAULT_PUBLISH_NAME
57from music_assistant.providers.spotify_connect.soloist.runtime import (
58    WS_ADDR_FILE,
59    WS_PORT_FILE,
60    SoloistAuthState,
61    SoloistDeviceChanged,
62    SoloistEntity,
63    SoloistError,
64    SoloistEvent,
65    SoloistOptionsChanged,
66    SoloistPlaybackOptions,
67    SoloistPlaybackState,
68    SoloistPosition,
69    SoloistTrackChanged,
70    SoloistVolumeChanged,
71)
72
73TRACK_A = "spotify:track:aaa"
74TRACK_B = "spotify:track:bbb"
75
76
77def test_trim_drops_an_all_zero_chunk_within_the_bound() -> None:
78    """A pure-silence chunk inside the trim budget is dropped entirely."""
79    chunk = b"\x00" * 1024
80    trimmed, skipped = _trim_lead_silence(chunk, 0)
81    assert trimmed == b""
82    assert skipped == 1024
83
84
85def test_trim_keeps_frame_alignment_when_audio_starts_mid_chunk() -> None:
86    """Audio starting mid-chunk is cut on a sample-frame boundary."""
87    # audio starts one byte into the third frame: the trim must keep that frame whole
88    chunk = b"\x00" * (_FRAME_BYTES * 2 + 1) + b"\x01" * 64
89    trimmed, skipped = _trim_lead_silence(chunk, 0)
90    assert skipped == _FRAME_BYTES * 2
91    assert len(trimmed) % _FRAME_BYTES == 1  # the partial frame's remainder is preserved
92    assert trimmed.endswith(b"\x01" * 64)
93
94
95def test_trim_passes_silence_through_once_the_bound_is_exceeded() -> None:
96    """Beyond the trim budget, silence is genuine content and is delivered."""
97    chunk = b"\x00" * 1024
98    trimmed, skipped = _trim_lead_silence(chunk, int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND))
99    assert trimmed == chunk
100    assert skipped == 0
101
102
103def test_seek_is_confirmed_only_within_tolerance(tmp_path: Path) -> None:
104    """A position report confirms a seek only once it reaches the tolerance window."""
105    item = _make_item(tmp_path, TRACK_A)
106    item.seek_target_ms = 60_000
107    item.observe_position(50_000)
108    assert not item.seek_confirmed.is_set()
109    item.observe_position(58_500)
110    assert item.seek_confirmed.is_set()
111
112
113def test_small_seek_target_is_not_confirmed_by_a_pre_seek_zero_report(tmp_path: Path) -> None:
114    """A position-0 report before the seek lands cannot confirm a small target."""
115    item = _make_item(tmp_path, TRACK_A)
116    item.seek_target_ms = 1_500
117    item.observe_position(0)
118    assert not item.seek_confirmed.is_set()
119    item.observe_position(1_500)
120    assert item.seek_confirmed.is_set()
121
122
123def test_position_never_regresses_and_stops_at_the_cut(tmp_path: Path) -> None:
124    """The furthest position is kept, and reports after the cut belong to the next item."""
125    item = _make_item(tmp_path, TRACK_A)
126    item.observe_position(120_000)
127    # the engine's stop/idle snapshot at the end of an item reports position 0
128    item.observe_position(0)
129    assert item.last_position_ms == 120_000
130    item.close()
131    item.observe_position(5_000)
132    assert item.last_position_ms == 120_000
133
134
135async def test_item_stream_ends_where_the_session_moves_on(tmp_path: Path) -> None:
136    """An item's audio ends at the track change, and the next item's begins there."""
137    session = _make_session(tmp_path)
138    item_a = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
139    session._current = item_a
140    item_a.started.set()
141    item_a.claim()
142    item_a.write(b"a" * 16)
143    await session._observe_current(TRACK_B, 200_000)
144    item_a.write(b"late" * 4)  # written after the cut: goes nowhere
145    chunks = [chunk async for chunk in item_a.read()]
146    assert b"".join(chunks) == b"a" * 16
147    # the next item exists, carries the duration and now receives the audio
148    item_b = session._items[TRACK_B]
149    assert session.current is item_b
150    assert item_b.duration_ms == 200_000
151
152
153async def test_the_engines_restored_state_does_not_cut_a_pending_item(
154    tmp_path: Path,
155) -> None:
156    """A daemon reports the item it restored before playing ours; that is not a boundary."""
157    session = _make_session(tmp_path)
158    requested = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
159    session._current = requested
160    requested.claim()
161    # the engine announces the state it came up with, which is someone else's item
162    await session._observe_current("spotify:track:restored", 152_000)
163    # closing our item here would end its stream before it delivered anything
164    assert requested._closed is False
165    assert requested.started.is_set() is False
166    # ... and the restored item is never offered as an item's audio
167    assert session.item_for("spotify:track:restored") is None
168    # then ours starts for real, and picks up from there
169    await session._observe_current(TRACK_A, 200_000)
170    assert session.current is requested
171    assert requested.started.is_set() is True
172    requested.write(b"\x01" * 32)
173    requested.close()
174    assert b"".join([chunk async for chunk in requested.read()]) == b"\x01" * 32
175
176
177async def test_leaving_the_engines_restored_item_is_not_a_takeover(tmp_path: Path) -> None:
178    """The restored item is part-way through a track, and we are about to leave it."""
179    session = _make_session(tmp_path)
180    # as _play leaves it: the channel exists, its stream is not reading it yet
181    requested = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
182    session._current = requested
183    await session._observe_current("spotify:track:restored", 152_000)
184    restored = session.current
185    assert restored is not None
186    restored.observe_position(20_000)
187
188    # our own play() lands and the engine leaves the restored item for ours
189    await session._observe_current(TRACK_A, 200_000)
190    assert session.usable is True
191    assert session.current is requested
192
193
194async def test_audio_read_before_the_stream_opens_is_kept(tmp_path: Path) -> None:
195    """Audio captured before an item's stream opens is buffered, not dropped."""
196    session = _make_session(tmp_path)
197    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
198    session._current = item
199    item.write(b"head" * 8)
200    item.claim()
201    item.close()
202    chunks = [chunk async for chunk in item.read()]
203    assert b"".join(chunks) == b"head" * 8
204
205
206async def test_a_channel_is_only_ever_served_once(tmp_path: Path) -> None:
207    """A consumed channel cannot be replayed, so the item needs a fresh session."""
208    session = _make_session(tmp_path)
209    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
210    item.started.set()
211    assert session.item_for(TRACK_A) is item
212    item.claim()
213    item.close()
214    item.release()
215    # this is what a queue holding the same track twice, or repeat wrapping back
216    # to the top, asks for: it must not be handed a drained channel
217    assert session.item_for(TRACK_A) is None
218
219
220async def test_an_abandoned_channel_cannot_be_continued(tmp_path: Path) -> None:
221    """A stream abandoned mid-item cannot resume where it left off either."""
222    session = _make_session(tmp_path)
223    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
224    item.started.set()
225    item.claim()
226    item.release()
227    assert session.item_for(TRACK_A) is None
228
229
230async def test_a_stuck_item_fails_instead_of_streaming_forever(tmp_path: Path) -> None:
231    """An item that runs far past its duration without a track change fails."""
232    session = _make_session(tmp_path)
233    item = _ItemAudio(TRACK_A, session)
234    item.duration_ms = 1_000
235    item.claim()
236    limit = item._overrun_limit()
237    assert limit is not None
238    item.write(b"\x01" * (limit + _FRAME_BYTES))
239    with pytest.raises(AudioError, match="never moved on"):
240        async for _ in item.read():
241            pass
242
243
244async def test_the_first_logged_out_snapshot_is_not_a_lost_pairing(tmp_path: Path) -> None:
245    """A daemon reports logged_in=False until it has restored its session."""
246    session = _make_session(tmp_path)
247    session._logged_in = None
248    session._was_active = False
249    await session._handle_event(_auth_event(logged_in=False, is_active=False))
250    # failing here would break every playback on a perfectly good pairing
251    assert session.usable is True
252    await session._handle_event(_auth_event(logged_in=True, is_active=False))
253    assert session.usable is True
254
255
256async def test_losing_an_established_login_fails_the_session(tmp_path: Path) -> None:
257    """A login that goes away mid-session is real, and ends the session."""
258    session = _make_session(tmp_path)
259    await session._handle_event(_auth_event(logged_in=True))
260    await session._handle_event(_auth_event(logged_in=False))
261    assert session.usable is False
262    assert session._error == "the session was logged out"
263
264
265async def test_buffering_gates_the_sink_once_demand_started(tmp_path: Path) -> None:
266    """Once PCM demand started, playing runs the sink and buffering suspends it again."""
267    session = _make_session(tmp_path)
268    session._demand_started = True
269    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
270    session._pending.append(TRACK_B)
271    sink = _sink_of(session)
272    # the sink is created suspended, so there is nothing to suspend yet
273    await session._handle_event(_playback_event("buffering"))
274    sink.suspend.assert_not_awaited()
275    await session._handle_event(_playback_event("playing"))
276    sink.resume.assert_awaited_once()
277    assert session._current is not None
278    assert session._current.playing_seen is True
279    # the engine stalling on a rebuffer keeps that silence out of the PCM
280    await session._handle_event(_playback_event("buffering"))
281    sink.suspend.assert_awaited_once()
282
283
284async def test_sink_is_not_gated_before_demand_started(tmp_path: Path) -> None:
285    """Buffering/playing before PCM demand leave the (still suspended) sink alone."""
286    session = _make_session(tmp_path)
287    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
288    sink = _sink_of(session)
289    await session._handle_event(_playback_event("buffering"))
290    await session._handle_event(_playback_event("playing"))
291    sink.suspend.assert_not_awaited()
292    sink.resume.assert_not_awaited()
293    # the status is recorded either way, so session start can decide when to resume
294    assert session._current.status == "playing"
295
296
297@pytest.mark.parametrize("end_status", ["stopped", "idle", "paused"])
298async def test_the_last_item_is_drained_rather_than_cut(tmp_path: Path, end_status: str) -> None:
299    """However the engine reports the end of a run, the last item drains and closes."""
300    session = _make_session(tmp_path)
301    session._demand_started = True
302    session._sink_running = True
303    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
304    item.duration_ms = 1_000
305    item.last_position_ms = 1_000
306    one_second = 1_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
307    sink = _sink_of(session)
308    await session._handle_event(_playback_event(end_status, position_ms=1_000))
309    # the sink stays open for now, so audio still in the FIFO can arrive...
310    sink.suspend.assert_not_awaited()
311    assert item.draining is True
312    assert item._closed is False
313    # ... but only that item's own audio is taken, never the padding silence the
314    # sink keeps rendering afterwards
315    item.write(b"\x01" * one_second)
316    item.write(b"\x00" * 4096)
317    assert item.buffered == one_second
318    await _wait_for(lambda: item._closed)
319    sink.suspend.assert_awaited_once()
320
321
322async def test_an_app_pause_midway_through_the_last_item_is_not_the_end(
323    tmp_path: Path,
324) -> None:
325    """Pausing in the Spotify app halfway through the last track must not truncate it."""
326    session = _make_session(tmp_path)
327    session._demand_started = True
328    session._sink_running = True
329    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
330    item.duration_ms = 200_000
331    item.last_position_ms = 90_000
332    await session._handle_event(_playback_event("paused", position_ms=90_000))
333    assert item.draining is False
334    assert item._closed is False
335    # treated as interference instead: the sink is gated and playback resumed
336    _sink_of(session).suspend.assert_awaited_once()
337    _client_of(session).resume.assert_awaited_once()
338
339
340async def test_a_resumed_item_cancels_its_tail_drain(tmp_path: Path) -> None:
341    """An armed drain is undone when the engine turns out to have been rebuffering."""
342    session = _make_session(tmp_path)
343    session._demand_started = True
344    session._sink_running = True
345    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
346    item.duration_ms = 200_000
347    item.last_position_ms = 199_000
348    await session._handle_event(_playback_event("stopped", position_ms=199_000))
349    armed = item.draining
350    await session._handle_event(_playback_event("playing", position_ms=199_500))
351    assert armed is True
352    assert item.draining is False
353    assert item._closed is False
354    assert item.drain_task is None
355
356
357async def test_the_cushion_is_capped_by_suspending_the_sink(tmp_path: Path) -> None:
358    """Undelivered audio is handed back as backpressure rather than piling up."""
359    session = _make_session(tmp_path)
360    session._demand_started = True
361    session._sink_running = True
362    session._engine_playing = True
363    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
364    item.claim()
365    sink = _sink_of(session)
366    await session._apply_sink_state()
367    sink.suspend.assert_not_awaited()
368    # the engine has run this far ahead of what the player has taken
369    item.write(b"\x01" * int((soloist_backend._MAX_RETAINED_S + 1) * _BYTES_PER_SECOND))
370    await session._apply_sink_state()
371    sink.suspend.assert_awaited_once()
372    assert session._backpressured is True
373    # and it comes back once the player has drained enough of it
374    item._buffered = int(soloist_backend._RESUME_RETAINED_S * _BYTES_PER_SECOND) - 1
375    await session._apply_sink_state()
376    sink.resume.assert_awaited_once()
377    assert session._backpressured is False
378
379
380async def test_a_pause_with_more_queued_suspends_the_sink(tmp_path: Path) -> None:
381    """A pause while another item is queued behind is ordinary interference, not the end."""
382    session = _make_session(tmp_path)
383    session._demand_started = True
384    session._sink_running = True
385    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
386    session._pending.append(TRACK_B)
387    await session._handle_event(_playback_event("paused"))
388    _sink_of(session).suspend.assert_awaited_once()
389
390
391async def test_nothing_is_sent_before_the_websocket_is_up(
392    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
393) -> None:
394    """Commands travel over the events socket: a published endpoint is not enough."""
395    monkeypatch.setattr(soloist_backend, "_STARTUP_TIMEOUT_S", 0.05)
396    session = _make_session(tmp_path)
397    client = _client_of(session)
398    # the endpoint file exists, but the events task has not connected yet
399    client.connected = False
400    endpoint_published = asyncio.Event()
401    endpoint_published.set()
402    with pytest.raises(AudioError, match="did not connect and log in"):
403        await session._play(TRACK_A, 0, endpoint_published)
404    client.activate.assert_not_awaited()
405    client.play.assert_not_awaited()
406
407
408async def test_nothing_is_sent_before_the_engine_has_logged_in(
409    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
410) -> None:
411    """The engine drops commands sent before it has restored its session."""
412    monkeypatch.setattr(soloist_backend, "_STARTUP_TIMEOUT_S", 0.05)
413    session = _make_session(tmp_path)
414    client = _client_of(session)
415    client.connected = True
416    # connected, but the engine has not announced its login yet
417    session._logged_in = None
418    endpoint_published = asyncio.Event()
419    endpoint_published.set()
420    with pytest.raises(AudioError, match="did not connect and log in"):
421        await session._play(TRACK_A, 0, endpoint_published)
422    client.activate.assert_not_awaited()
423    client.play.assert_not_awaited()
424
425
426async def test_startup_activates_before_it_plays(
427    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
428) -> None:
429    """A fresh daemon has to become the active device before it is told to play."""
430    session = _make_session(tmp_path)
431    client = _client_of(session)
432    client.connected = True
433    monkeypatch.setattr(session, "_await_item_ready", AsyncMock())
434    endpoint_published = asyncio.Event()
435    endpoint_published.set()
436    item = await session._play(TRACK_A, 0, endpoint_published)
437    assert item.uri == TRACK_A
438    client.activate.assert_awaited_once_with(await_result=True)
439    client.play.assert_awaited_once_with(TRACK_A)
440
441
442async def test_a_takeover_between_activate_and_play_stops_the_start(tmp_path: Path) -> None:
443    """Playing here would claim the device straight back off wherever the user moved to."""
444    session = _make_session(tmp_path)
445    session._was_active = False
446    client = _client_of(session)
447
448    async def _take_over(*_args: Any, **_kwargs: Any) -> None:
449        session._observe_active_device(is_active=False)
450
451    client.set_repeat_track.side_effect = _take_over
452    ready = asyncio.Event()
453    ready.set()
454    with pytest.raises(SoloistAppControlError):
455        await session._play(TRACK_A, 0, ready)
456    client.play.assert_not_awaited()
457
458
459async def test_a_refused_start_command_reports_soloist(tmp_path: Path) -> None:
460    """A dropped start command surfaces as a Soloist error, not a raw client one."""
461    session = _make_session(tmp_path)
462    client = _client_of(session)
463    client.connected = True
464    client.activate.side_effect = SoloistError("websocket is not connected")
465    endpoint_published = asyncio.Event()
466    endpoint_published.set()
467    with pytest.raises(AudioError, match="Spotify Soloist would not start"):
468        await session._play(TRACK_A, 0, endpoint_published)
469
470
471async def test_the_engine_is_told_not_to_shuffle_or_repeat(
472    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
473) -> None:
474    """MA owns the order, and a repeating engine would never reach the item fed behind."""
475    session = _make_session(tmp_path)
476    client = _client_of(session)
477    client.connected = True
478    monkeypatch.setattr(session, "_await_item_ready", AsyncMock())
479    endpoint_published = asyncio.Event()
480    endpoint_published.set()
481    await session._play(TRACK_A, 0, endpoint_published)
482    client.set_shuffle.assert_awaited_once_with(False)
483    client.set_repeat_context.assert_awaited_once_with(False)
484    client.set_repeat_track.assert_awaited_once_with(False)
485
486
487async def test_repeat_turned_on_from_the_app_is_pinned_back_off(tmp_path: Path) -> None:
488    """Repeat enabled in the Spotify app is undone before it can loop the item."""
489    session = _make_session(tmp_path)
490    await session._handle_event(
491        SoloistEvent(
492            type="options_changed",
493            data=SoloistOptionsChanged(
494                options=SoloistPlaybackOptions(shuffle=True, repeat="track")
495            ),
496            raw={},
497        )
498    )
499    client = _client_of(session)
500    client.set_shuffle.assert_awaited_once_with(False)
501    client.set_repeat_track.assert_awaited_once_with(False)
502    client.set_repeat_context.assert_awaited_once_with(False)
503    # options that are already off are left alone
504    client.set_shuffle.reset_mock()
505    await session._handle_event(
506        SoloistEvent(
507            type="options_changed",
508            data=SoloistOptionsChanged(options=SoloistPlaybackOptions()),
509            raw={},
510        )
511    )
512    client.set_shuffle.assert_not_awaited()
513
514
515async def test_a_busy_data_directory_is_reported_as_such(tmp_path: Path) -> None:
516    """A daemon left over from an earlier run is named, not reported as a generic failure."""
517    session = _make_session(tmp_path)
518    # the daemon's own parting complaint, which is all it gives (it exits with 1)
519    session._data_dir_busy = True
520    with pytest.raises(AudioError, match="Another Spotify Soloist session is still running"):
521        session._raise_startup_error("exited before playback started", TRACK_A)
522
523
524async def test_the_busy_marker_is_picked_up_from_the_daemon_output(tmp_path: Path) -> None:
525    """The marker is read off the daemon's stdout, with the API key still redacted."""
526    session = _make_session(tmp_path)
527    proc = MagicMock()
528    lines = [
529        'Error: another session is running for data directory "/data/x/soloist-data".',
530        "Stop the running session before starting soloist again.",
531    ]
532
533    async def _iter_stdout() -> AsyncGenerator[str]:
534        for line in lines:
535            yield line
536
537    proc.iter_stdout = _iter_stdout
538    await session._log_output(proc)
539    assert session._data_dir_busy is True
540
541
542async def test_a_pairing_that_never_logs_in_routes_through_setup(
543    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
544) -> None:
545    """A session that cannot log in sends the user back to setup, not a per-track error."""
546    session = _make_session(tmp_path)
547    unload_with_error = MagicMock()
548    monkeypatch.setattr(session.backend.provider, "unload_with_error", unload_with_error)
549    await session._handle_event(_auth_event(logged_in=False))
550    with pytest.raises(LoginFailed) as err:
551        session._raise_startup_error("timed out waiting for playback to start", TRACK_A)
552    assert err.value.translation_key == "soloist_pairing_required"
553    # ... and the provider is taken out of service, so the user is asked to redo setup
554    unload_with_error.assert_called_once()
555
556
557async def test_a_login_that_never_happened_is_not_confused_with_another_failure(
558    tmp_path: Path,
559) -> None:
560    """An unrelated failure keeps its own message even before any login was reported."""
561    session = _make_session(tmp_path)
562    session._fail("the capture sink was lost mid-stream")
563    with pytest.raises(AudioError, match="capture sink was lost"):
564        session._raise_startup_error("exited before playback started", TRACK_A)
565
566
567def test_a_seeked_item_only_expects_what_is_left_of_it(tmp_path: Path) -> None:
568    """A seeked item delivers the remainder, so its targets are based on that."""
569    session = _make_session(tmp_path)
570    item = _ItemAudio(TRACK_A, session)
571    item.duration_ms = 200_000
572    full = 200_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
573    assert item._duration_bytes() == full
574    item.seek_target_ms = 150_000
575    remainder = 50_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
576    assert item._duration_bytes() == remainder
577    # so the tail drain has a target it can actually reach
578    item.start_tail_drain()
579    item.write(b"\x01" * remainder)
580    assert item.tail_complete is True
581    # and the padding silence after it is refused
582    item.write(b"\x00" * 4096)
583    assert item.buffered == remainder
584
585
586def test_the_lead_trim_never_exceeds_its_budget() -> None:
587    """Silence beyond the budget is content, including where audio starts mid-chunk."""
588    budget = int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND)
589    # already at the budget, with a chunk whose silence runs well past it
590    chunk = b"\x00" * 4096 + b"\x01" * 64
591    trimmed, skipped = _trim_lead_silence(chunk, budget - _FRAME_BYTES)
592    assert skipped == _FRAME_BYTES
593    assert len(trimmed) == len(chunk) - _FRAME_BYTES
594
595
596async def test_a_dying_log_reader_fails_the_session(tmp_path: Path) -> None:
597    """Nothing else drains the daemon's stdout, so a dead reader must not go unnoticed."""
598    session = _make_session(tmp_path)
599
600    async def _boom() -> None:
601        raise RuntimeError("reader blew up")
602
603    session._log_task = asyncio.create_task(_boom())
604    session._log_task.add_done_callback(session._task_done)
605    await asyncio.sleep(0)
606    await _wait_for(lambda: not session.usable)
607    assert session._error is not None
608    assert "reader blew up" in session._error
609
610
611async def test_feeding_never_replaces_a_channel_already_in_use(tmp_path: Path) -> None:
612    """If the engine reaches the fed item first, its live channel must survive."""
613    session = _make_session(tmp_path, queue_id="player1")
614    streamdetails = MagicMock()
615    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
616    queues = _queues_of(session)
617    queues.get.return_value = MagicMock(current_index=0)
618    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
619    queues.get_next_item.return_value = _queue_item(TRACK_B)
620
621    async def _engine_gets_there_first(_uri: str, **_kwargs: Any) -> None:
622        # the events task advances to the fed item while the command is in flight
623        await session._observe_current(TRACK_B, 200_000)
624
625    _client_of(session).add_to_queue.side_effect = _engine_gets_there_first
626    await session.feed_after(streamdetails, TRACK_A)
627    live = session.current
628    assert live is not None
629    assert live.uri == TRACK_B
630    # the channel the reader is writing to is the one a stream will be handed
631    assert session._items[TRACK_B] is live
632    assert session.item_for(TRACK_B) is live
633    # and it is not queued as pending, because it already started
634    assert session.has_pending is False
635
636
637async def test_seeking_the_playing_item_restarts_the_session(
638    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
639) -> None:
640    """
641    A seek re-opens the item that is playing, which is a restart of the session.
642
643    A realtime source has not captured anything past the play position, so any
644    forward seek lands outside the buffer and comes back here.
645    """
646    backend = _make_backend(tmp_path)
647    backend._server = MagicMock()
648    backend._binary = Path("/nonexistent/soloist")
649    session = _SoloistSession(backend, "player1")
650    backend._session = session
651    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
652    item.started.set()
653    # its own stream is still attached when the seek re-opens it
654    item.claim()
655    stopped = AsyncMock()
656    monkeypatch.setattr(session, "stop", stopped)
657    _install_fake_binary_manager(monkeypatch)
658    monkeypatch.setattr(
659        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
660    )
661    with pytest.raises(AudioError, match="spawn"):
662        await backend._acquire(TRACK_A, 90, "player1")
663    stopped.assert_awaited_once()
664
665
666@pytest.mark.parametrize(
667    ("requested", "other_queue"),
668    [
669        # another player, whatever it asks for - including the very track this
670        # session is in the middle of delivering
671        pytest.param(TRACK_B, "player2", id="other_player"),
672        pytest.param(TRACK_A, "player2", id="other_player_same_track"),
673        # an early fetch across a boundary this session does not drive, such as a
674        # podcast episode or audiobook chapter
675        pytest.param(TRACK_B, "player1", id="unstitched_boundary"),
676    ],
677)
678async def test_a_session_in_use_is_never_cut_short(
679    tmp_path: Path, requested: str, other_queue: str
680) -> None:
681    """
682    An item the session cannot serve must not stop one it is still delivering.
683
684    Reported as capacity, so a speculative prepare gives up softly.
685    """
686    backend = _make_backend(tmp_path)
687    backend._server = MagicMock()
688    backend._binary = Path("/nonexistent/soloist")
689    session = _SoloistSession(backend, "player1")
690    backend._session = session
691    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
692    item.started.set()
693    item.claim()
694    # the session really is playing TRACK_A, so a same-track request from another
695    # player cannot be mistaken for a seek
696    session._current = item
697    with pytest.raises(ProviderStreamLimitError) as err:
698        await backend._acquire(requested, 0, other_queue)
699    # a stream-limit error so the item is not marked unplayable, but the message
700    # is about the session, not the provider's source-stream budget
701    assert err.value.limit == 1
702    assert err.value.translation_key == "soloist_session_busy"
703    # the session that was playing is untouched
704    assert backend._session is session
705    assert session.usable is True
706
707
708async def test_a_session_nobody_reads_is_replaced_for_another_item(
709    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
710) -> None:
711    """Once the other item has been released, the same request gets the session."""
712    backend = _make_backend(tmp_path)
713    backend._server = MagicMock()
714    backend._binary = Path("/nonexistent/soloist")
715    session = _SoloistSession(backend, "player1")
716    backend._session = session
717    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
718    item.started.set()
719    item.claim()
720    item.close()
721    item.release()
722    _install_fake_binary_manager(monkeypatch)
723    monkeypatch.setattr(
724        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
725    )
726    monkeypatch.setattr(session, "stop", AsyncMock())
727    with pytest.raises(AudioError, match="spawn"):
728        await backend._acquire(TRACK_B, 0, "player1")
729
730
731async def test_a_replacement_waits_for_the_old_daemon_to_be_gone(
732    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
733) -> None:
734    """The engine refuses to start while another daemon still holds its data dir."""
735    backend = _make_backend(tmp_path)
736    backend._server = MagicMock()
737    backend._binary = Path("/nonexistent/soloist")
738    session = _SoloistSession(backend, "player1")
739    backend._session = session
740    order: list[str] = []
741
742    async def _slow_stop() -> None:
743        order.append("stop-start")
744        await asyncio.sleep(0.05)
745        order.append("stop-done")
746
747    monkeypatch.setattr(session, "stop", _slow_stop)
748    _install_fake_binary_manager(monkeypatch)
749
750    async def _spawn(_self: Any, _uri: str, _seek: int) -> None:
751        order.append("spawn")
752        raise AudioError("spawn")
753
754    monkeypatch.setattr(soloist_backend._SoloistSession, "start", _spawn)
755    # the session failed, so its teardown is under way when the next item arrives
756    discard = asyncio.create_task(backend.discard_session(session))
757    await asyncio.sleep(0)
758    with pytest.raises(AudioError, match="spawn"):
759        await backend._acquire(TRACK_B, 0, "player1")
760    await discard
761    assert order == ["stop-start", "stop-done", "spawn"]
762
763
764async def test_an_idle_session_is_taken_over(
765    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
766) -> None:
767    """A session nobody is reading is replaced instead of blocking another player."""
768    backend = _make_backend(tmp_path)
769    backend._server = MagicMock()
770    backend._binary = Path("/nonexistent/soloist")
771    session = _SoloistSession(backend, "player1")
772    session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
773    backend._session = session
774    stopped = AsyncMock()
775    monkeypatch.setattr(session, "stop", stopped)
776    _install_fake_binary_manager(monkeypatch)
777    # the replacement spawn is out of scope here; only the takeover decision is
778    monkeypatch.setattr(
779        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
780    )
781    with pytest.raises(AudioError, match="spawn"):
782        await backend._acquire(TRACK_B, 0, "player2")
783    stopped.assert_awaited_once()
784
785
786def test_a_dead_session_task_fails_the_session(tmp_path: Path) -> None:
787    """A session task that dies of an unexpected error takes the session with it."""
788    session = _make_session(tmp_path)
789    task: Any = MagicMock()
790    task.cancelled.return_value = False
791    task.exception.return_value = RuntimeError("reader blew up")
792    session._task_done(task)
793    assert session.usable is False
794    assert session._error is not None
795    assert "reader blew up" in session._error
796
797
798def test_a_cancelled_session_task_is_not_a_failure(tmp_path: Path) -> None:
799    """Teardown cancels the session's tasks; that must not be reported as an error."""
800    session = _make_session(tmp_path)
801    task: Any = MagicMock()
802    task.cancelled.return_value = True
803    session._task_done(task)
804    assert session.usable is True
805
806
807async def test_failed_sink_control_fails_the_session(tmp_path: Path) -> None:
808    """A failed suspend/resume fails the session instead of leaking stall silence."""
809    session = _make_session(tmp_path)
810    session._demand_started = True
811    session._sink_running = True
812    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
813    session._pending.append(TRACK_B)
814    _sink_of(session).suspend.side_effect = RuntimeError("pactl failed")
815    await session._handle_event(_playback_event("buffering"))
816    assert session._error is not None
817    assert "capture sink control failed" in session._error
818
819
820async def test_app_pause_is_fought_with_a_resume(tmp_path: Path) -> None:
821    """A pause from the Spotify app is undone: this session has no user-facing pause."""
822    session = _make_session(tmp_path)
823    session._demand_started = True
824    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
825    session._pending.append(TRACK_B)
826    await session._handle_event(_playback_event("paused"))
827    _client_of(session).resume.assert_awaited_once()
828
829
830async def test_an_app_pause_is_only_undone_so_many_times(tmp_path: Path) -> None:
831    """Someone who keeps pausing means it: the session gives up instead of fighting on."""
832    session = _make_session(tmp_path)
833    session._demand_started = True
834    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
835    session._pending.append(TRACK_B)
836    for _ in range(_MAX_APP_PAUSE_RESUMES):
837        await session._handle_event(_playback_event("playing"))
838        await session._handle_event(_playback_event("paused"))
839    assert _client_of(session).resume.await_count == _MAX_APP_PAUSE_RESUMES
840    assert session._error is None
841
842    await session._handle_event(_playback_event("playing"))
843    await session._handle_event(_playback_event("paused"))
844    assert _client_of(session).resume.await_count == _MAX_APP_PAUSE_RESUMES
845    assert session.usable is False
846    assert session._app_control is SoloistAppControl.PAUSED
847
848
849async def test_one_pause_reported_twice_counts_once(tmp_path: Path) -> None:
850    """A repeated snapshot of the same pause is not a new pause."""
851    session = _make_session(tmp_path)
852    session._demand_started = True
853    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
854    session._pending.append(TRACK_B)
855    for _ in range(_MAX_APP_PAUSE_RESUMES + 2):
856        await session._handle_event(_playback_event("paused"))
857    assert session.usable is True
858
859
860async def test_the_pause_budget_resets_on_the_next_item(tmp_path: Path) -> None:
861    """Each item gets its own budget; pausing one track does not spend the next one's."""
862    session = _make_session(tmp_path)
863    session._demand_started = True
864    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
865    session._pending.append(TRACK_B)
866    for _ in range(_MAX_APP_PAUSE_RESUMES):
867        await session._handle_event(_playback_event("playing"))
868        await session._handle_event(_playback_event("paused"))
869    await session._observe_current(TRACK_B, 200_000)
870    assert session._app_pauses == 0
871
872
873async def test_a_pause_is_not_undone_once_the_device_is_gone(tmp_path: Path) -> None:
874    """A bare resume on a device Spotify no longer routes to would play to nobody."""
875    session = _make_session(tmp_path)
876    session._demand_started = True
877    session._was_active = False
878    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
879    session._pending.append(TRACK_B)
880    await session._handle_event(_playback_event("paused"))
881    _client_of(session).resume.assert_not_awaited()
882
883
884async def test_losing_the_active_device_ends_the_session(tmp_path: Path) -> None:
885    """Playback moved to another device from the Spotify app: this session is over."""
886    session = _make_session(tmp_path)
887    await session._handle_event(_device_event(is_active=False))
888    assert session.usable is False
889    assert session._app_control is SoloistAppControl.TOOK_OVER
890
891
892async def test_a_takeover_reported_on_the_auth_state_ends_the_session(tmp_path: Path) -> None:
893    """The active-device state also rides on auth_state, and counts the same there."""
894    session = _make_session(tmp_path)
895    await session._handle_event(_auth_event(logged_in=True, is_active=False))
896    assert session.usable is False
897
898
899async def test_an_inactive_device_before_activation_is_not_a_takeover(tmp_path: Path) -> None:
900    """A fresh daemon is inactive until the session claims it; that is not a takeover."""
901    session = _make_session(tmp_path)
902    session._was_active = False
903    await session._handle_event(_device_event(is_active=False))
904    await session._handle_event(_auth_event(logged_in=True, is_active=False))
905    assert session.usable is True
906
907    # nor does a respawned daemon reporting the session Spotify still has for
908    # the account: only the status _play claimed is followed
909    await session._handle_event(_device_event(is_active=True))
910    await session._handle_event(_device_event(is_active=False))
911    assert session.usable is True
912
913
914async def test_a_reconnect_snapshot_keeps_an_active_session_alive(tmp_path: Path) -> None:
915    """The events connection re-snapshots after a drop; that is not a device change."""
916    session = _make_session(tmp_path)
917    await session._handle_event(_auth_event(logged_in=True, is_active=True))
918    await session._handle_event(_device_event(is_active=True))
919    assert session.usable is True
920
921
922async def test_the_playback_snapshots_active_flag_is_ignored(tmp_path: Path) -> None:
923    """It is optional and rides on deltas, so only the dedicated reports are followed."""
924    session = _make_session(tmp_path)
925    session._demand_started = True
926    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
927    await session._handle_event(
928        SoloistEvent(
929            type="playback_changed",
930            data=SoloistPlaybackState(status="playing", is_active=False),
931            raw={},
932        )
933    )
934    assert session.usable is True
935
936
937async def test_backpressure_does_not_spend_the_pause_budget(tmp_path: Path) -> None:
938    """A sink suspended to cap the cushion is our doing, not the user pausing."""
939    session = _make_session(tmp_path)
940    session._demand_started = True
941    session._engine_playing = True
942    session._backpressured = True
943    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
944    session._pending.append(TRACK_B)
945    await session._handle_event(_playback_event("paused"))
946    _client_of(session).resume.assert_not_awaited()
947    assert session._app_pauses == 0
948
949
950async def test_a_lost_login_is_not_reported_as_a_takeover(tmp_path: Path) -> None:
951    """Losing the login wins over the inactive device it brings with it."""
952    session = _make_session(tmp_path)
953    await session._handle_event(_auth_event(logged_in=False, is_active=False))
954    assert session.usable is False
955    assert session._app_control is None
956
957
958async def test_a_track_started_from_the_app_ends_the_session(tmp_path: Path) -> None:
959    """The engine pulled off an item part-way through is the app playing something else."""
960    session = _make_session(tmp_path)
961    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
962    item.duration_ms = 200_000
963    await session._observe_current(TRACK_A, 200_000)
964    item.observe_position(20_000)
965
966    await session._observe_current("spotify:track:theirs", 180_000)
967    assert session.usable is False
968    assert session._app_control is SoloistAppControl.TOOK_OVER
969    assert session.current is item
970
971
972async def test_a_track_played_earlier_started_from_the_app_ends_the_session(
973    tmp_path: Path,
974) -> None:
975    """A known uri is no exemption: only the item fed behind this one is where we sent it."""
976    session = _make_session(tmp_path)
977    played = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
978    played.spent = True
979    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
980    item.duration_ms = 200_000
981    await session._observe_current(TRACK_A, 200_000)
982    item.observe_position(20_000)
983
984    await session._observe_current(TRACK_B, 180_000)
985    assert session.usable is False
986    assert session._app_control is SoloistAppControl.TOOK_OVER
987
988
989async def test_skipping_from_the_app_to_the_fed_item_is_followed(tmp_path: Path) -> None:
990    """The queue moves to that same track, so following the engine keeps the two in step."""
991    session = _make_session(tmp_path)
992    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
993    item.duration_ms = 200_000
994    await session._observe_current(TRACK_A, 200_000)
995    item.observe_position(20_000)
996    fed = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
997    session._pending.append(TRACK_B)
998
999    await session._observe_current(TRACK_B, 180_000)
1000    assert session.usable is True
1001    assert session.current is fed
1002    assert session.item_for(TRACK_B) is fed
1003
1004
1005async def test_a_takeover_snapshot_stops_pinning_volume_and_options(tmp_path: Path) -> None:
1006    """Once the app has the session, the rest of its snapshot must not reach the daemon."""
1007    session = _make_session(tmp_path)
1008    session._demand_started = True
1009    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1010    item.duration_ms = 200_000
1011    await session._observe_current(TRACK_A, 200_000)
1012    item.observe_position(20_000)
1013
1014    await session._handle_event(
1015        SoloistEvent(
1016            type="playback_changed",
1017            data=SoloistPlaybackState(
1018                status="playing",
1019                item=SoloistEntity(uri="spotify:track:theirs", entity_type="track"),
1020                volume=40,
1021                options=SoloistPlaybackOptions(shuffle=True, repeat="context"),
1022            ),
1023            raw={},
1024        )
1025    )
1026    assert session.usable is False
1027    _client_of(session).set_volume.assert_not_awaited()
1028    _client_of(session).set_shuffle.assert_not_awaited()
1029
1030
1031async def test_the_engine_moving_on_at_a_track_end_is_not_a_takeover(tmp_path: Path) -> None:
1032    """An unasked-for item the engine reaches at a boundary is its own autoplay."""
1033    session = _make_session(tmp_path)
1034    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1035    item.duration_ms = 200_000
1036    await session._observe_current(TRACK_A, 200_000)
1037    item.observe_position(200_000)
1038
1039    await session._observe_current("spotify:track:autoplay", 180_000)
1040    assert session.usable is True
1041    assert session.item_for("spotify:track:autoplay") is None
1042
1043
1044async def test_a_long_crossfade_boundary_is_not_a_takeover(tmp_path: Path) -> None:
1045    """With crossfade the engine moves on a crossfade short of the duration."""
1046    session = _make_session(tmp_path)
1047    session.crossfade_ms = 15_000
1048    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1049    await session._observe_current(TRACK_A, 200_000)
1050    # the last position reported before the engine crossfades into the next track
1051    item.observe_position(200_000 - session.crossfade_ms)
1052
1053    await session._observe_current("spotify:track:autoplay", 180_000)
1054    assert session.usable is True
1055
1056
1057async def test_an_ended_item_says_what_the_app_did(tmp_path: Path) -> None:
1058    """The item's stream fails with the takeover, not a generic session error."""
1059    session = _make_session(tmp_path)
1060    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1061    await session._handle_event(_device_event(is_active=False))
1062    with pytest.raises(SoloistAppControlError) as err:
1063        await session.validate_item(item)
1064    assert err.value.translation_key == SoloistAppControl.TOOK_OVER.value
1065    assert isinstance(err.value, ProviderStreamLimitError)
1066
1067
1068async def test_a_session_being_torn_down_does_not_hold_off_the_next_one(tmp_path: Path) -> None:
1069    """Teardown pauses the daemon; that must not read as the user pausing."""
1070    session = _make_session(tmp_path)
1071    session._demand_started = True
1072    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
1073    session._pending.append(TRACK_B)
1074    session._stopped = True
1075    for _ in range(_MAX_APP_PAUSE_RESUMES + 1):
1076        await session._handle_event(_playback_event("playing"))
1077        await session._handle_event(_playback_event("paused"))
1078    await session._handle_event(_device_event(is_active=False))
1079    session.backend._raise_if_app_controlled()
1080    _client_of(session).resume.assert_not_awaited()
1081
1082
1083async def test_no_session_is_started_while_the_app_holds_the_last_one(tmp_path: Path) -> None:
1084    """A replacement would claim the Connect device straight back off the user."""
1085    backend = _make_backend(tmp_path)
1086    backend._note_app_control(SoloistAppControl.TOOK_OVER)
1087    with pytest.raises(SoloistAppControlError):
1088        await backend._acquire(TRACK_A, 0, "player1")
1089
1090
1091async def test_the_hold_on_a_new_session_expires(tmp_path: Path) -> None:
1092    """Coming back to Music Assistant later plays again without any fuss."""
1093    backend = _make_backend(tmp_path)
1094    backend._note_app_control(SoloistAppControl.TOOK_OVER)
1095    backend._app_control_until = time.monotonic() - 1
1096    backend._raise_if_app_controlled()
1097    assert backend._held_by_app() is None
1098
1099
1100async def test_an_audiobook_gives_up_on_capacity_instead_of_burning_chapters(
1101    tmp_path: Path,
1102) -> None:
1103    """Skipping ahead would cost the audiobook its availability and the caller its retry."""
1104    provider = _make_provider(tmp_path)
1105    calls: list[str] = []
1106
1107    async def _refuse(uri: str, *_args: Any, **_kwargs: Any) -> AsyncGenerator[bytes]:
1108        calls.append(uri)
1109        for _ in ():  # never yields; only makes this an async generator
1110            yield b""
1111        raise SoloistAppControlError(provider, SoloistAppControl.TOOK_OVER)
1112
1113    provider.backend = MagicMock(stream_spotify_uri=_refuse)
1114    streamdetails = MagicMock(
1115        media_type=MediaType.AUDIOBOOK,
1116        data={"chapters": [TRACK_A, TRACK_B, "spotify:track:ccc"], "chapters_data": []},
1117    )
1118
1119    with pytest.raises(SoloistAppControlError):
1120        async for _ in provider.get_audio_stream(streamdetails):
1121            pass
1122    # the first chapter's refusal ends it: no chapter is skipped over
1123    assert calls == [TRACK_A]
1124
1125
1126def test_the_playback_device_is_named_apart_from_the_connect_one() -> None:
1127    """Two identically named devices in the Spotify app is what causes the takeovers."""
1128    assert SOLOIST_DEVICE_NAME != DEFAULT_PUBLISH_NAME
1129
1130
1131async def test_app_volume_change_is_pinned_back_to_unity(tmp_path: Path) -> None:
1132    """An off-unity volume set from the Spotify app is pinned back to 100."""
1133    session = _make_session(tmp_path)
1134    await session._handle_event(
1135        SoloistEvent(type="volume_changed", data=SoloistVolumeChanged(volume=40), raw={})
1136    )
1137    _client_of(session).set_volume.assert_awaited_once_with(100)
1138    _client_of(session).set_volume.reset_mock()
1139    await session._handle_event(
1140        SoloistEvent(type="volume_changed", data=SoloistVolumeChanged(volume=100), raw={})
1141    )
1142    _client_of(session).set_volume.assert_not_awaited()
1143
1144
1145async def test_track_change_signals_the_queue_when_it_matches_the_next_item(
1146    tmp_path: Path,
1147) -> None:
1148    """Reaching a fed item tells the queue to start filling that item's buffer."""
1149    session = _make_session(tmp_path, queue_id="player1")
1150    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
1151    session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1152    queues = _queues_of(session)
1153    queues.get.return_value = MagicMock(next_item=_queue_item(TRACK_B), current_index=0)
1154    await session._handle_event(
1155        SoloistEvent(
1156            type="track_changed",
1157            data=SoloistTrackChanged(item=SoloistEntity(uri=TRACK_B, entity_type="track")),
1158            raw={},
1159        )
1160    )
1161    queues.prepare_next_audio_buffer.assert_called_once_with("player1")
1162
1163
1164async def test_track_change_to_another_item_signals_nothing(tmp_path: Path) -> None:
1165    """An item the queue is not asking for next must not trigger a prebuffer."""
1166    session = _make_session(tmp_path, queue_id="player1")
1167    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
1168    queues = _queues_of(session)
1169    queues.get.return_value = MagicMock(next_item=_queue_item(TRACK_B), current_index=0)
1170    await session._handle_event(
1171        SoloistEvent(
1172            type="track_changed",
1173            data=SoloistTrackChanged(
1174                item=SoloistEntity(uri="spotify:track:surprise", entity_type="track")
1175            ),
1176            raw={},
1177        )
1178    )
1179    queues.prepare_next_audio_buffer.assert_not_called()
1180
1181
1182async def test_the_follower_of_the_streamed_item_is_fed(tmp_path: Path) -> None:
1183    """The item after the one being streamed is handed to the engine."""
1184    session = _make_session(tmp_path, queue_id="player1")
1185    streamdetails = MagicMock()
1186    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1187    follower = _queue_item(TRACK_B)
1188    queues = _queues_of(session)
1189    queues.get.return_value = MagicMock(current_index=3)
1190    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 3 else None
1191    queues.get_next_item.return_value = follower
1192    await session.feed_after(streamdetails, TRACK_A)
1193    _client_of(session).add_to_queue.assert_awaited_once_with(TRACK_B)
1194    assert TRACK_B in session._items
1195    assert session.has_pending is True
1196
1197
1198async def test_an_item_the_queue_resolved_elsewhere_is_not_fed(tmp_path: Path) -> None:
1199    """A track the queue will stream from another provider must not be queued here."""
1200    session = _make_session(tmp_path, queue_id="player1")
1201    streamdetails = MagicMock()
1202    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1203    # same track, but the queue already picked a different provider for it
1204    follower = _queue_item(TRACK_B, streamdetails=MagicMock(provider="tidal--x"))
1205    queues = _queues_of(session)
1206    queues.get.return_value = MagicMock(current_index=0)
1207    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1208    queues.get_next_item.return_value = follower
1209    await session.feed_after(streamdetails, TRACK_A)
1210    _client_of(session).add_to_queue.assert_not_awaited()
1211
1212
1213async def test_skipping_to_the_fed_item_keeps_the_session(tmp_path: Path) -> None:
1214    """A next-track lands on the item already fed, so the engine jumps instead of respawning."""
1215    backend = _make_backend(tmp_path)
1216    backend._server = MagicMock()
1217    backend._binary = Path("/nonexistent/soloist")
1218    session = _SoloistSession(backend, "player1")
1219    session._client = AsyncMock()
1220    session._logged_in = True
1221    backend._session = session
1222    playing = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
1223    playing.started.set()
1224    # fed one ahead and not reached yet, which is where a next-track goes
1225    fed = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1226    session._pending.append(TRACK_B)
1227
1228    async def _engine_gets_there(**_kwargs: Any) -> None:
1229        await session._observe_current(TRACK_B, 200_000)
1230
1231    _client_of(session).skip_next.side_effect = _engine_gets_there
1232    got_session, got_item = await backend._acquire(TRACK_B, 0, "player1")
1233    # the same session, no respawn, and the item that was already queued
1234    assert got_session is session
1235    assert got_item is fed
1236    assert backend._session is session
1237    _client_of(session).skip_next.assert_awaited_once()
1238
1239
1240async def test_a_skip_drops_what_arrives_while_the_command_is_in_flight(
1241    tmp_path: Path,
1242) -> None:
1243    """
1244    Audio captured between the skip command and the engine's answer is dropped.
1245
1246    Only covers the marker's own window; what the pipeline still holds when the
1247    answer arrives is measured at the cut instead.
1248    """
1249    session = _make_session(tmp_path)
1250    leaving = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
1251    leaving.started.set()
1252    target = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1253    session._pending.append(TRACK_B)
1254    captured: list[bytes] = []
1255
1256    async def _engine_gets_there(**_kwargs: Any) -> None:
1257        # the pipeline still holds the old track while the command is in flight
1258        session._write_if_wanted(b"\x01" * 32)
1259        await session._observe_current(TRACK_B, 200_000)
1260        # from here on the audio really is the new item's
1261        session._write_if_wanted(b"\x02" * 32)
1262
1263    _client_of(session).skip_next.side_effect = _engine_gets_there
1264    await session.skip_to(target)
1265    captured.extend(target._chunks)
1266    assert b"".join(captured) == b"\x02" * 32
1267    assert session._discard_until is None
1268
1269
1270async def test_a_skip_the_engine_never_reaches_fails(
1271    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1272) -> None:
1273    """A skip that does not land is an error, not a wait for the track to end."""
1274    monkeypatch.setattr(soloist_backend, "_STARTUP_TIMEOUT_S", 0.05)
1275    session = _make_session(tmp_path)
1276    fed = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1277    session._pending.append(TRACK_B)
1278    with pytest.raises(AudioError, match="did not reach"):
1279        await session.skip_to(fed)
1280
1281
1282async def test_a_fed_item_the_engine_has_not_reached_is_not_served(tmp_path: Path) -> None:
1283    """Skipping to an already-fed item must not hand over a channel that fills later."""
1284    session = _make_session(tmp_path)
1285    fed = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1286    # fed, but the engine is still on the previous track
1287    assert session.item_for(TRACK_B) is None
1288    # once the engine gets there it is servable
1289    await session._observe_current(TRACK_B, 200_000)
1290    assert session.item_for(TRACK_B) is fed
1291
1292
1293def test_the_shaper_only_emits_whole_frames() -> None:
1294    """A read that ends mid-frame must never split a frame across two items."""
1295    shaper = soloist_backend._CaptureShaper()
1296    # the session's first bytes are infrastructure silence, and are dropped
1297    assert shaper.shape(b"\x00" * 4096) == b""
1298    # a mis-aligned read emits whole frames and carries the remainder
1299    first = shaper.shape(b"\x01" * (_FRAME_BYTES + 3))
1300    assert len(first) == _FRAME_BYTES
1301    # which is then completed by the next read, losing nothing
1302    second = shaper.shape(b"\x02" * (_FRAME_BYTES - 3))
1303    assert len(second) == _FRAME_BYTES
1304    assert second[:3] == b"\x01" * 3
1305    # an aligned read passes straight through
1306    assert shaper.shape(b"\x03" * _FRAME_BYTES) == b"\x03" * _FRAME_BYTES
1307
1308
1309def test_the_shaper_trims_lead_silence_only_once() -> None:
1310    """Silence after the audio has started is content, not pre-roll."""
1311    shaper = soloist_backend._CaptureShaper()
1312    assert shaper.shape(b"\x01" * _FRAME_BYTES) == b"\x01" * _FRAME_BYTES
1313    silence = b"\x00" * _FRAME_BYTES
1314    assert shaper.shape(silence) == silence
1315
1316
1317async def test_only_whole_sample_frames_are_handed_over(
1318    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1319) -> None:
1320    """A read that ends mid-frame must not split a frame across two items."""
1321    session = _make_session(tmp_path)
1322    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
1323    item.started.set()
1324    item.claim()
1325    session._demand_started = True
1326    session._sink_running = True
1327    # two reads that are each mis-aligned but whole together
1328    reads = [b"\x01" * (_FRAME_BYTES + 3), b"\x02" * (_FRAME_BYTES - 3), b""]
1329    reader = MagicMock()
1330
1331    async def _read(_size: int) -> bytes:
1332        return reads.pop(0) if reads else b""
1333
1334    reader.read = _read
1335    session._reader = reader
1336    monkeypatch.setattr(soloist_backend, "_PACE_RATE", 1000.0)
1337    await session._read_capture()
1338    # every write was frame-aligned, and no byte was lost
1339    assert item.buffered % _FRAME_BYTES == 0
1340    assert item.buffered == _FRAME_BYTES * 2
1341
1342
1343async def test_an_already_known_item_is_not_fed_twice(tmp_path: Path) -> None:
1344    """An item the session already plays or was fed is not queued again."""
1345    session = _make_session(tmp_path, queue_id="player1")
1346    session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1347    streamdetails = MagicMock()
1348    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1349    queues = _queues_of(session)
1350    queues.get.return_value = MagicMock(current_index=0)
1351    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1352    queues.get_next_item.return_value = _queue_item(TRACK_B)
1353    await session.feed_after(streamdetails, TRACK_A)
1354    _client_of(session).add_to_queue.assert_not_awaited()
1355
1356
1357async def test_only_tracks_are_fed_ahead(tmp_path: Path) -> None:
1358    """A podcast episode or audiobook chapter is played on its own, never stitched."""
1359    session = _make_session(tmp_path, queue_id="player1")
1360    await session.feed_after(MagicMock(), "spotify:episode:xyz")
1361    _client_of(session).add_to_queue.assert_not_awaited()
1362
1363
1364async def test_a_non_spotify_follower_is_not_fed(tmp_path: Path) -> None:
1365    """The run simply ends where the queue leaves this provider."""
1366    session = _make_session(tmp_path, queue_id="player1")
1367    streamdetails = MagicMock()
1368    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1369    follower = MagicMock(
1370        media_item=MagicMock(media_type=MediaType.TRACK, provider="tidal--x"), streamdetails=None
1371    )
1372    follower.media_item.provider_mappings = []
1373    queues = _queues_of(session)
1374    queues.get.return_value = MagicMock(current_index=0)
1375    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1376    queues.get_next_item.return_value = follower
1377    await session.feed_after(streamdetails, TRACK_A)
1378    _client_of(session).add_to_queue.assert_not_awaited()
1379
1380
1381async def test_a_library_item_is_fed_through_its_spotify_mapping(tmp_path: Path) -> None:
1382    """A library track is fed with the item id this provider instance knows it by."""
1383    session = _make_session(tmp_path, queue_id="player1")
1384    streamdetails = MagicMock()
1385    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1386    follower = MagicMock(
1387        media_item=MagicMock(media_type=MediaType.TRACK, provider="library", item_id="42"),
1388        streamdetails=None,
1389    )
1390    follower.media_item.provider_mappings = [
1391        MagicMock(provider_instance="other--y", item_id="wrong"),
1392        MagicMock(provider_instance="spotify--test", item_id="bbb"),
1393    ]
1394    queues = _queues_of(session)
1395    queues.get.return_value = MagicMock(current_index=0)
1396    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1397    queues.get_next_item.return_value = follower
1398    await session.feed_after(streamdetails, TRACK_A)
1399    _client_of(session).add_to_queue.assert_awaited_once_with(TRACK_B)
1400
1401
1402@pytest.mark.parametrize(
1403    ("provider_option", "player_setting", "expected"),
1404    [
1405        (True, "enabled", True),
1406        # the player's own switch decides first: off means nobody normalizes,
1407        # not that the job passes to Spotify
1408        (True, "disabled", False),
1409        (False, "enabled", False),
1410        (False, "disabled", False),
1411    ],
1412)
1413def test_who_normalizes_needs_both_switches(
1414    tmp_path: Path,
1415    monkeypatch: pytest.MonkeyPatch,
1416    provider_option: bool,
1417    player_setting: str,
1418    expected: bool,
1419) -> None:
1420    """The engine normalizes only when the provider option and the player agree."""
1421    session = _make_session(tmp_path, queue_id="player1")
1422    monkeypatch.setattr(
1423        type(session.backend.provider),
1424        "spotify_normalization_configured",
1425        property(lambda _self: provider_option),
1426    )
1427    cast("MagicMock", session.mass.config).get_effective_player_queue_config_value = MagicMock(
1428        return_value=player_setting
1429    )
1430    assert session._engine_normalization_enabled() is expected
1431
1432
1433def test_a_running_session_answers_for_what_the_engine_is_doing(tmp_path: Path) -> None:
1434    """
1435    The engine reads its settings at startup, so a later toggle must not split them.
1436
1437    Otherwise the streams core would start normalizing on top of audio the engine
1438    is still normalizing, or stop while it no longer is.
1439    """
1440    backend = _make_backend(tmp_path)
1441    provider = backend.provider
1442    streamdetails = _streamdetails_for(queue_id="player1")
1443    # nothing playing yet: the configuration is all there is to go on
1444    before_any_session = backend.session_normalizes(streamdetails)
1445    session = _SoloistSession(backend, "player1")
1446    session.engine_normalizes = True
1447    backend._session = session
1448    while_playing = backend.session_normalizes(streamdetails)
1449    # ... and a session that has been torn down no longer speaks for the engine
1450    session._stopped = True
1451    after_teardown = backend.session_normalizes(streamdetails)
1452    assert before_any_session is None
1453    assert while_playing is True
1454    assert after_teardown is None
1455    assert (
1456        provider.delivers_normalized_audio(streamdetails)
1457        is provider.spotify_normalization_configured
1458    )
1459
1460
1461def test_crossfade_comes_from_the_queue_preference(tmp_path: Path) -> None:
1462    """The queue's crossfade setting is handed to the engine, in milliseconds."""
1463    session = _make_session(tmp_path, queue_id="player1")
1464    _queues_of(session).get.return_value = MagicMock(queue_id="player1", crossfade_enabled=True)
1465    cast("MagicMock", session.mass.config).get_raw_core_config_value = MagicMock(return_value=6)
1466    assert session._queue_crossfade_ms() == 6000
1467
1468
1469def test_crossfade_off_is_zero(tmp_path: Path) -> None:
1470    """A queue with crossfade disabled gets an explicit zero (which clears the pref)."""
1471    session = _make_session(tmp_path, queue_id="player1")
1472    _queues_of(session).get.return_value = MagicMock(crossfade_enabled=False)
1473    assert session._queue_crossfade_ms() == 0
1474
1475
1476def test_no_queue_means_no_crossfade(tmp_path: Path) -> None:
1477    """Without a queue to read the preference from, the engine gets no crossfade."""
1478    session = _make_session(tmp_path, queue_id=None)
1479    assert session._queue_crossfade_ms() == 0
1480
1481
1482async def test_a_fed_boundary_the_engine_played_across_is_reported(tmp_path: Path) -> None:
1483    """The engine playing on into a fed item is what puts the overlap in its audio."""
1484    session = _make_session(tmp_path, queue_id="player1")
1485    session.crossfade_ms = 8000
1486    streamdetails = _streamdetails_for(uri=TRACK_B)
1487    session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1488    session._pending.append(TRACK_B)
1489    queues = _queues_of(session)
1490    queues.get.return_value = MagicMock(current_index=0)
1491    queues.get_item.return_value = None
1492
1493    # fed but not reached, so MA arriving here now means a jump - not a fade
1494    assert session.fades_a_boundary_of(streamdetails) is False
1495
1496    await session._observe_current(TRACK_B, 200_000)
1497
1498    assert session._items[TRACK_B].faded_in is True
1499    assert session.fades_a_boundary_of(streamdetails) is True
1500
1501
1502async def test_a_jumped_to_item_opens_on_a_hard_cut(tmp_path: Path) -> None:
1503    """A skip discards the overlap the engine rendered, so no fade reaches the item."""
1504    session = _make_session(tmp_path, queue_id="player1")
1505    session.crossfade_ms = 8000
1506    session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1507    session._pending.append(TRACK_B)
1508    session._discard_until = TRACK_B
1509    queues = _queues_of(session)
1510    queues.get.return_value = MagicMock(current_index=0)
1511    queues.get_item.return_value = None
1512
1513    await session._observe_current(TRACK_B, 200_000)
1514
1515    assert session._items[TRACK_B].faded_in is False
1516    assert session.fades_a_boundary_of(_streamdetails_for(uri=TRACK_B)) is False
1517
1518
1519async def test_a_feed_the_engine_acted_on_before_failing_still_fades(tmp_path: Path) -> None:
1520    """
1521    The engine can reach a fed item and the command still fail on the way back.
1522
1523    The channel is kept for it in that case, because the reader is already writing
1524    there - so the boundary was played across and denying the fade would describe a
1525    cut the engine never made.
1526    """
1527    session = _make_session(tmp_path, queue_id="player1")
1528    session.crossfade_ms = 8000
1529    streamdetails = _streamdetails_for(uri=TRACK_A)
1530    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1531    streamed = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1532    streamed.started.set()
1533    session._current = streamed
1534    queues = _queues_of(session)
1535    queues.get.return_value = MagicMock(current_index=0)
1536    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1537    queues.get_next_item.return_value = _queue_item(TRACK_B)
1538
1539    async def _reached_then_failed(_uri: str, **_kwargs: Any) -> None:
1540        await session._observe_current(TRACK_B, 200_000)
1541        raise TimeoutError
1542
1543    _client_of(session).add_to_queue.side_effect = _reached_then_failed
1544
1545    await session.feed_after(streamdetails, TRACK_A)
1546
1547    # the channel survived because the engine is playing it, so the fade is real
1548    assert session._items[TRACK_B].faded_in is True
1549    assert streamed.fades_out is True
1550
1551
1552async def test_an_item_only_fed_is_reached_by_a_jump(tmp_path: Path) -> None:
1553    """
1554    An item MA opens before the engine gets to it is jumped to, so no fade reaches it.
1555
1556    Its own follower would otherwise stand in and claim one, which is why being handed
1557    over but not reached has to answer first.
1558    """
1559    session = _make_session(tmp_path, queue_id="player1")
1560    session.crossfade_ms = 8000
1561    streamdetails = _streamdetails_for(uri=TRACK_B)
1562    playing = _queue_item(TRACK_B, streamdetails=streamdetails)
1563    session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1564    session._pending.append(TRACK_B)
1565    queues = _queues_of(session)
1566    queues.get.return_value = MagicMock(current_index=1)
1567    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 1 else None
1568    # a follower it could be fed, so only the pending state can produce the hard cut
1569    queues.get_next_item.return_value = _queue_item(TRACK_A)
1570
1571    assert session.fades_a_boundary_of(streamdetails) is False
1572
1573
1574async def test_an_item_the_engine_reached_unfed_carries_no_fade(tmp_path: Path) -> None:
1575    """
1576    Only an item handed to the engine can be played on into.
1577
1578    The engine also reports items nobody asked for - the session it restores at
1579    startup, its own autoplay - and crediting those would claim an overlap that
1580    was never rendered.
1581    """
1582    session = _make_session(tmp_path, queue_id="player1")
1583    session.crossfade_ms = 8000
1584
1585    await session._observe_current("spotify:track:restored", 200_000)
1586
1587    assert session._items["spotify:track:restored"].faded_in is False
1588
1589
1590async def test_an_earlier_play_of_a_track_does_not_answer_for_a_later_one(
1591    tmp_path: Path,
1592) -> None:
1593    """
1594    Channels are keyed by track, so a repeated one must not read the first play's fades.
1595
1596    In an A-B-A queue the first A faded out into B, but the last A plays on a fresh
1597    session with nothing after it, so both of its boundaries are hard cuts.
1598    """
1599    session = _make_session(tmp_path, queue_id="player1")
1600    session.crossfade_ms = 8000
1601    first_a = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1602    first_a.started.set()
1603    first_a.spent = True
1604    first_a.fades_out = True
1605    playing_b = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1606    playing_b.started.set()
1607    playing_b.faded_in = True
1608    session._current = playing_b
1609    last_a = _streamdetails_for(uri=TRACK_A)
1610    queues = _queues_of(session)
1611    queues.get.return_value = MagicMock(current_index=2)
1612    queues.get_item.side_effect = lambda _queue_id, index: (
1613        _queue_item(TRACK_A, streamdetails=last_a) if index == 2 else None
1614    )
1615    queues.get_next_item.return_value = None
1616
1617    assert session.fades_a_boundary_of(last_a) is False
1618
1619
1620async def test_a_fresh_sessions_first_item_rests_on_its_own_follower(tmp_path: Path) -> None:
1621    """
1622    Nothing was faded into the item a session starts on, so only its end can be faded.
1623
1624    A single-track play from the middle of a list is exactly this: the item before it
1625    never played, so crediting that boundary would report an overlap nobody rendered.
1626    """
1627    session = _make_session(tmp_path, queue_id="player1")
1628    session.crossfade_ms = 8000
1629    streamdetails = _streamdetails_for(uri=TRACK_A)
1630    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1631    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1632    item.started.set()
1633    queues = _queues_of(session)
1634    queues.get.return_value = MagicMock(current_index=0)
1635    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1636
1637    queues.get_next_item.return_value = None
1638    assert session.fades_a_boundary_of(streamdetails) is False
1639
1640    queues.get_next_item.return_value = _queue_item(TRACK_B)
1641    assert session.fades_a_boundary_of(streamdetails) is True
1642
1643
1644@pytest.mark.parametrize(
1645    ("crossfade_ms", "feed_fails", "expected"),
1646    [
1647        (8000, False, True),
1648        # a feed that did not land costs the crossfade at exactly that boundary
1649        (8000, True, False),
1650        # and an engine started without crossfade cuts hard whatever it was fed
1651        (0, False, False),
1652    ],
1653)
1654async def test_feeding_the_follower_settles_the_boundary_at_the_items_end(
1655    tmp_path: Path, crossfade_ms: int, feed_fails: bool, expected: bool
1656) -> None:
1657    """Once the feed has been attempted, its outcome is what the boundary rests on."""
1658    session = _make_session(tmp_path, queue_id="player1")
1659    session.crossfade_ms = crossfade_ms
1660    streamdetails = _streamdetails_for(uri=TRACK_A)
1661    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1662    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1663    item.started.set()
1664    session._current = item
1665    queues = _queues_of(session)
1666    queues.get.return_value = MagicMock(current_index=0)
1667    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1668    queues.get_next_item.return_value = _queue_item(TRACK_B)
1669    if feed_fails:
1670        _client_of(session).add_to_queue.side_effect = TimeoutError
1671
1672    await session.feed_after(streamdetails, TRACK_A)
1673
1674    assert item.fades_out is expected
1675    assert session.fades_a_boundary_of(streamdetails) is expected
1676
1677
1678@pytest.mark.parametrize(
1679    ("follower_state", "expected"),
1680    [
1681        # handed over and waiting: the engine reaches it by playing on
1682        ("pending", True),
1683        # the engine is already there, ahead of the stream MA is still reading
1684        ("current", True),
1685        # served once already, so the next play of it starts a fresh session - an
1686        # A-B-A queue reaches this after jumping to B
1687        ("drained", False),
1688    ],
1689)
1690async def test_a_follower_the_session_already_holds_decides_the_boundary(
1691    tmp_path: Path, follower_state: str, expected: bool
1692) -> None:
1693    """
1694    With nothing left to feed, the channel already there is what the boundary rests on.
1695
1696    A drained channel cannot be replayed, so crediting it would report an overlap the
1697    engine never renders - and the answer has to be the same before and after the feed
1698    that finds nothing to send.
1699    """
1700    session = _make_session(tmp_path, queue_id="player1")
1701    session.crossfade_ms = 8000
1702    streamdetails = _streamdetails_for(uri=TRACK_B)
1703    playing = _queue_item(TRACK_B, streamdetails=streamdetails)
1704    streamed = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1705    streamed.started.set()
1706    session._current = streamed
1707    follower = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1708    if follower_state == "pending":
1709        session._pending.append(TRACK_A)
1710    else:
1711        follower.started.set()
1712        if follower_state == "current":
1713            session._current = follower
1714        else:
1715            follower.spent = True
1716    queues = _queues_of(session)
1717    queues.get.return_value = MagicMock(current_index=1)
1718    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 1 else None
1719    queues.get_next_item.return_value = _queue_item(TRACK_A)
1720
1721    assert session.fades_a_boundary_of(streamdetails) is expected
1722
1723    await session.feed_after(streamdetails, TRACK_B)
1724
1725    _client_of(session).add_to_queue.assert_not_awaited()
1726    assert streamed.fades_out is expected
1727    assert session.fades_a_boundary_of(streamdetails) is expected
1728
1729
1730async def test_only_a_track_can_carry_a_source_fade(tmp_path: Path) -> None:
1731    """A podcast episode is never stitched, so both of its boundaries are hard cuts."""
1732    session = _make_session(tmp_path, queue_id="player1")
1733    session.crossfade_ms = 8000
1734    episode = _streamdetails_for(uri="spotify:episode:xyz", media_type=MediaType.PODCAST_EPISODE)
1735
1736    assert session.fades_a_boundary_of(episode) is False
1737
1738
1739async def test_short_delivery_is_rejected_as_incomplete(tmp_path: Path) -> None:
1740    """PCM that stops well short of the item's duration is rejected."""
1741    session = _make_session(tmp_path)
1742    item = _ItemAudio(TRACK_A, session)
1743    item.playing_seen = True
1744    item.duration_ms = 200_000
1745    item.last_position_ms = 100_000
1746    with pytest.raises(AudioError, match="incomplete"):
1747        await session.validate_item(item)
1748
1749
1750async def test_a_crossfade_shortfall_is_tolerated(tmp_path: Path) -> None:
1751    """With crossfade the engine reports the item short by design; that is not a failure."""
1752    session = _make_session(tmp_path)
1753    session.crossfade_ms = 12_000
1754    item = _ItemAudio(TRACK_A, session)
1755    item.playing_seen = True
1756    item.duration_ms = 200_000
1757    # 12s of crossfade plus the ordinary tolerance
1758    item.last_position_ms = 200_000 - 21_000
1759    await session.validate_item(item)
1760
1761
1762async def test_missing_position_is_rejected_as_incomplete(tmp_path: Path) -> None:
1763    """Without any position report there is no evidence the item played out."""
1764    session = _make_session(tmp_path)
1765    item = _ItemAudio(TRACK_A, session)
1766    item.playing_seen = True
1767    item.duration_ms = 200_000
1768    with pytest.raises(AudioError, match="incomplete"):
1769        await session.validate_item(item)
1770
1771
1772async def test_short_item_cannot_pass_at_position_zero(tmp_path: Path) -> None:
1773    """The tolerance never spans a whole item, so a short item cannot pass unplayed."""
1774    session = _make_session(tmp_path)
1775    item = _ItemAudio(TRACK_A, session)
1776    item.playing_seen = True
1777    item.duration_ms = 8_000
1778    item.last_position_ms = 0
1779    with pytest.raises(AudioError, match="incomplete"):
1780        await session.validate_item(item)
1781
1782
1783async def test_an_item_that_never_played_is_rejected(tmp_path: Path) -> None:
1784    """An item the engine never reported playing is a failure, whatever was delivered."""
1785    session = _make_session(tmp_path)
1786    item = _ItemAudio(TRACK_A, session)
1787    item.duration_ms = 200_000
1788    item.last_position_ms = 200_000
1789    with pytest.raises(AudioError, match="never started playing"):
1790        await session.validate_item(item)
1791
1792
1793async def test_a_duration_less_item_is_not_judged(tmp_path: Path) -> None:
1794    """Without a duration there is nothing to judge completeness against."""
1795    session = _make_session(tmp_path)
1796    item = _ItemAudio(TRACK_A, session)
1797    item.playing_seen = True
1798    await session.validate_item(item)
1799
1800
1801def test_an_unread_session_expires(tmp_path: Path) -> None:
1802    """A session no item stream reads from is ended so its daemon does not linger."""
1803    session = _make_session(tmp_path)
1804    session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1805    session._expire_idle()
1806    assert session._idle_since is not None
1807    assert session.usable is True
1808    session._idle_since = time.monotonic() - _IDLE_TIMEOUT_S - 1
1809    session._expire_idle()
1810    assert session.usable is False
1811
1812
1813def test_a_session_being_read_never_expires(tmp_path: Path) -> None:
1814    """An item stream reading the session keeps it alive indefinitely."""
1815    session = _make_session(tmp_path)
1816    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1817    item.claim()
1818    session._idle_since = time.monotonic() - _IDLE_TIMEOUT_S * 10
1819    session._expire_idle()
1820    assert session.usable is True
1821
1822
1823def test_pre_roll_silence_is_dropped_a_whole_frame_at_a_time() -> None:
1824    """
1825    Trimming pre-roll must leave the audio on the session's frame grid.
1826
1827    A FIFO read is not always a whole number of frames, and dropping a partial
1828    one would shift every sample that follows for the rest of the session.
1829    """
1830    shaper = _CaptureShaper()
1831    # pre-roll that ends mid-frame: the real audio starts at byte 1024
1832    assert shaper.shape(b"\x00" * 1021) == b""
1833    audio = bytes(range(1, 9)) * 4
1834    shaped = shaper.shape(b"\x00" * 3 + audio)
1835    assert shaped == audio
1836    assert shaper._lead_skipped % _FRAME_BYTES == 0
1837
1838
1839async def test_a_refused_skip_does_not_leave_the_audio_discarded(tmp_path: Path) -> None:
1840    """
1841    A skip that never landed must not keep the session dropping its audio.
1842
1843    The marker silences everything the session captures, so a command that
1844    failed has to clear it on the way out.
1845    """
1846    session = _make_session(tmp_path)
1847    client = cast("MagicMock", session._client)
1848    client.skip_next = AsyncMock(side_effect=TimeoutError)
1849    item = _ItemAudio(TRACK_B, session)
1850
1851    with pytest.raises(AudioError, match="would not skip"):
1852        await session.skip_to(item)
1853
1854    assert session._discard_until is None
1855
1856
1857async def test_a_daemon_that_will_not_die_is_reported_and_released(tmp_path: Path) -> None:
1858    """A close that could not terminate the daemon still finishes the teardown."""
1859    session = _make_session(tmp_path)
1860    proc = cast("MagicMock", session._proc)
1861    proc.close = AsyncMock()
1862    # AsyncProcess.close() gives up after a handful of kill attempts
1863    proc.returncode = None
1864    with patch.object(session.logger, "warning") as warning:
1865        await session.stop()
1866    assert warning.called
1867    assert session._teardown_done is True
1868    assert session._proc is None
1869
1870
1871async def test_a_cancelled_teardown_still_closes_the_daemon(tmp_path: Path) -> None:
1872    """
1873    A cancelled teardown must leave the retry something to close.
1874
1875    Dropping the references first is how a daemon survives to hold the data
1876    directory, which every later session is then refused for.
1877    """
1878    session = _make_session(tmp_path)
1879    proc = cast("MagicMock", session._proc)
1880    sink = cast("AsyncMock", session._sink)
1881
1882    async def _never_returns() -> None:
1883        await asyncio.Event().wait()
1884
1885    proc.close = _never_returns
1886    task = asyncio.create_task(session.stop())
1887    await asyncio.sleep(0.01)
1888    task.cancel()
1889    with suppress(asyncio.CancelledError):
1890        await task
1891    # the teardown did not finish, so nothing was dropped and it can be redone
1892    unfinished = session._teardown_done
1893    kept_proc = session._proc
1894    kept_sink = session._sink
1895    proc.close = AsyncMock()
1896    proc.returncode = 0
1897    await session.stop()
1898    assert unfinished is False
1899    assert kept_proc is proc
1900    assert kept_sink is sink
1901    assert session._teardown_done is True
1902    assert session._proc is None
1903    assert session._sink is None
1904    proc.close.assert_awaited()
1905    sink.unload.assert_awaited()
1906
1907
1908def test_a_failed_session_is_torn_down(tmp_path: Path) -> None:
1909    """A session that fails is discarded, so its daemon does not keep playing to nobody."""
1910    session = _make_session(tmp_path)
1911    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1912    item.claim()
1913    session._fail("audio stalled")
1914    assert session.usable is False
1915    # every waiting item is released and the teardown is scheduled
1916    assert item._closed is True
1917    # a startup wait must not sit out its timeout on a session that already failed
1918    assert item.started.is_set() is True
1919    discard = cast("MagicMock", session.mass.create_task)
1920    discard.assert_called_once_with(session.backend.discard_session, session)
1921    # a second failure does not queue a second teardown
1922    session._fail("and again")
1923    assert session._error == "audio stalled"
1924    assert discard.call_count == 1
1925
1926
1927async def test_an_item_the_engine_skipped_past_fails_instead_of_hanging(
1928    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1929) -> None:
1930    """A claimed channel the engine never reaches gives up rather than blocking forever."""
1931    monkeypatch.setattr(soloist_backend, "_READ_SLICE_S", 0.01)
1932    monkeypatch.setattr(soloist_backend, "_STALL_TIMEOUT_S", 0.05)
1933    session = _make_session(tmp_path)
1934    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1935    item.claim()
1936    # the engine is playing something else, so nothing is ever written here
1937    session._items["spotify:track:other"] = session._current = _ItemAudio(
1938        "spotify:track:other", session
1939    )
1940    with pytest.raises(AudioError, match="no audio"):
1941        async for _ in item.read():
1942            pass
1943
1944
1945async def test_adopt_paired_session_copies_into_the_canonical_dir(
1946    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1947) -> None:
1948    """A session paired by the setup flow is adopted into the per-instance data dir."""
1949    storage = tmp_path / "storage"
1950    pending = storage / "spotify" / "pairing" / "flow1"
1951    pending.mkdir(parents=True)
1952    (pending / "session.bin").write_bytes(b"session")
1953    prov = _make_provider(tmp_path, {CONF_SOLOIST_SESSION_DIR: "spotify/pairing/flow1"})
1954    update_setup_data = MagicMock()
1955    monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
1956    backend = SoloistBackend(prov)
1957    await backend._adopt_paired_session()
1958    canonical = storage / "spotify" / "spotify--test" / SOLOIST_DATA_DIR_NAME
1959    assert (canonical / "session.bin").read_bytes() == b"session"
1960    # a copy, not a move: the flow-private source must survive a failed
1961    # provider load so the setup flow can retry (the flow removes it at its end)
1962    assert (pending / "session.bin").exists()
1963    update_setup_data.assert_called_once_with(CONF_SOLOIST_SESSION_DIR, None)
1964
1965
1966def test_the_engine_is_told_not_to_normalize(tmp_path: Path) -> None:
1967    """MA normalizes this audio itself, so the engine's own normalization is switched off."""
1968    backend = _make_backend(tmp_path)
1969    prefs = backend._data_dir / "settings" / "Users" / "alice-user" / "prefs"
1970    prefs.parent.mkdir(parents=True)
1971    prefs.write_text("some.engine.key=1\n", encoding="utf-8")
1972    backend._prepare_data_dir(8000, normalize=False)
1973    content = prefs.read_text(encoding="utf-8").splitlines()
1974    assert "some.engine.key=1" in content
1975    assert "audio.normalize_v2=false" in content
1976    assert "audio.crossfade_v2=true" in content
1977    assert "audio.crossfade.time_v2=8000" in content
1978    # the ceiling is stated rather than left to the engine's own default
1979    assert "audio.play_bitrate_enumeration=5" in content
1980    assert "audio.play_bitrate_non_metered_enumeration=5" in content
1981    assert "audio.play_bitrate_non_metered_migrated=true" in content
1982
1983
1984def test_disabling_crossfade_writes_the_boolean(tmp_path: Path) -> None:
1985    """Crossfade off is written explicitly, so a stale 'on' cannot survive."""
1986    backend = _make_backend(tmp_path)
1987    prefs = backend._data_dir / "settings" / "prefs"
1988    prefs.parent.mkdir(parents=True)
1989    prefs.write_text("audio.crossfade_v2=true\naudio.crossfade.time_v2=8000\n", encoding="utf-8")
1990    backend._prepare_data_dir(0, normalize=False)
1991    content = prefs.read_text(encoding="utf-8").splitlines()
1992    assert "audio.crossfade_v2=false" in content
1993    assert not any(line.startswith("audio.crossfade.time_v2") for line in content)
1994
1995
1996async def test_setup_requires_an_api_key(tmp_path: Path) -> None:
1997    """Without a stored API key the user must be sent back through the setup flow."""
1998    backend = _make_backend(tmp_path)
1999    with pytest.raises(LoginFailed) as err:
2000        await backend.setup()
2001    assert err.value.translation_key == "soloist_pairing_required"
2002
2003
2004async def test_setup_requires_a_paired_session(
2005    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
2006) -> None:
2007    """An API key without a paired session also routes back to the setup flow."""
2008    backend = _make_backend(tmp_path, {CONF_SOLOIST_API_KEY: "k" * 20, CONF_SOLOIST_CONSENT: True})
2009    _install_fake_binary_manager(monkeypatch)
2010    with pytest.raises(LoginFailed) as err:
2011        await backend.setup()
2012    assert err.value.translation_key == "soloist_pairing_required"
2013
2014
2015async def test_streaming_without_setup_is_refused(tmp_path: Path) -> None:
2016    """A backend whose setup never ran refuses to stream instead of half-starting."""
2017    backend = _make_backend(tmp_path)
2018    with pytest.raises(AudioError, match="not started"):
2019        async for _ in backend.stream_spotify_uri(TRACK_A):
2020            pass
2021
2022
2023def test_session_present_detection(tmp_path: Path) -> None:
2024    """Only a dir holding something besides the WS endpoint files counts as paired."""
2025    data_dir = tmp_path / "soloist-data"
2026    assert soloist_session_present(data_dir) is False
2027    data_dir.mkdir()
2028    (data_dir / WS_ADDR_FILE).write_text("127.0.0.1", encoding="utf-8")
2029    (data_dir / WS_PORT_FILE).write_text("1234", encoding="utf-8")
2030    assert soloist_session_present(data_dir) is False
2031    (data_dir / "session.bin").write_bytes(b"x")
2032    assert soloist_session_present(data_dir) is True
2033
2034
2035async def test_a_skip_drops_the_audio_still_in_flight(tmp_path: Path) -> None:
2036    """The item jumped to opens with its own audio, not the tail of the one left behind."""
2037    session = _make_session(tmp_path)
2038    left_behind = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
2039    session._current = left_behind
2040    left_behind.started.set()
2041    left_behind.claim()
2042    jumped_to = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
2043    session._pending.append(TRACK_B)
2044    session._discard_until = TRACK_B
2045    with _capture_holding(session, fifo_bytes=2 * _FRAME_BYTES, reader_bytes=2 * _FRAME_BYTES):
2046        await session._observe_current(TRACK_B, 200_000)
2047    assert session._stale_budget == 4 * _FRAME_BYTES
2048    session._write_if_wanted(b"s" * (4 * _FRAME_BYTES))
2049    session._write_if_wanted(b"n" * (2 * _FRAME_BYTES))
2050    jumped_to.claim()
2051    jumped_to.close()
2052    assert b"".join([chunk async for chunk in jumped_to.read()]) == b"n" * (2 * _FRAME_BYTES)
2053
2054
2055async def test_a_skip_drops_the_stale_audio_across_reads(tmp_path: Path) -> None:
2056    """A budget larger than one read keeps dropping, and resumes on a frame boundary."""
2057    session = _make_session(tmp_path)
2058    session._stale_budget = 3 * _FRAME_BYTES
2059    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
2060    item.claim()
2061    session._write_if_wanted(b"s" * (2 * _FRAME_BYTES))
2062    session._write_if_wanted(b"s" * _FRAME_BYTES + b"n" * _FRAME_BYTES)
2063    item.close()
2064    assert b"".join([chunk async for chunk in item.read()]) == b"n" * _FRAME_BYTES
2065
2066
2067async def test_the_marker_spends_an_earlier_jumps_budget(tmp_path: Path) -> None:
2068    """What the marker drops still counts against a budget left from an earlier jump."""
2069    session = _make_session(tmp_path)
2070    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
2071    session._stale_budget = 4 * _FRAME_BYTES
2072    session._discard_until = TRACK_B
2073    session._write_if_wanted(b"s" * (3 * _FRAME_BYTES))
2074    assert session._stale_budget == _FRAME_BYTES
2075    # a refused command leaves only what is genuinely still in flight to drop
2076    session._discard_until = None
2077    session._write_if_wanted(b"s" * _FRAME_BYTES + b"n" * _FRAME_BYTES)
2078    item = session._current
2079    item.claim()
2080    item.close()
2081    assert b"".join([chunk async for chunk in item.read()]) == b"n" * _FRAME_BYTES
2082
2083
2084async def test_a_natural_cut_keeps_the_audio_in_flight(tmp_path: Path) -> None:
2085    """Nothing is dropped without a jump: what is in flight is the continuation."""
2086    session = _make_session(tmp_path)
2087    playing = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
2088    session._current = playing
2089    playing.started.set()
2090    playing.claim()
2091    with _capture_holding(session, fifo_bytes=4 * _FRAME_BYTES, reader_bytes=4 * _FRAME_BYTES):
2092        await session._observe_current(TRACK_B, 200_000)
2093    assert session._stale_budget == 0
2094
2095
2096def test_stale_bytes_spans_both_buffers_in_whole_frames(tmp_path: Path) -> None:
2097    """The in-flight measure covers the FIFO and the reader, and never splits a frame."""
2098    session = _make_session(tmp_path)
2099    with _capture_holding(session, fifo_bytes=3 * _FRAME_BYTES + 3, reader_bytes=2 * _FRAME_BYTES):
2100        assert session._stale_bytes() == 5 * _FRAME_BYTES
2101
2102
2103def test_stale_bytes_falls_back_when_the_reader_cannot_be_sized(tmp_path: Path) -> None:
2104    """Losing the reader's internal view drops extra rather than leaving audio behind."""
2105    session = _make_session(tmp_path)
2106    with _capture_holding(session, fifo_bytes=0, reader_bytes=None):
2107        assert session._stale_bytes() == 6 * _READ_CHUNK_SIZE
2108
2109
2110async def test_a_channel_abandoned_at_the_cut_stops_holding_the_cushion(
2111    tmp_path: Path,
2112) -> None:
2113    """A skip closes the channel first and only then unwinds its stream."""
2114    session = _make_session(tmp_path)
2115    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
2116    item.started.set()
2117    item.claim()
2118    item.write(b"x" * 4096)
2119    # the cut lands while the abandoned stream is still unwinding
2120    await session._observe_current(TRACK_B, 200_000)
2121    assert session._retained_bytes() == 4096
2122    item.release()
2123    assert session._retained_bytes() == 0
2124
2125
2126def test_an_abandoned_channel_stops_holding_the_cushion(tmp_path: Path) -> None:
2127    """A channel skipped away from frees its buffer instead of gating the sink for good."""
2128    session = _make_session(tmp_path)
2129    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
2130    item.claim()
2131    item.write(b"x" * 4096)
2132    assert session._retained_bytes() == 4096
2133    # the stream is gone, then the cut closes the channel
2134    item.release()
2135    item.close()
2136    assert session._retained_bytes() == 0
2137
2138
2139async def test_a_channel_still_being_read_keeps_its_tail(tmp_path: Path) -> None:
2140    """Closing the playing item at a cut must not discard what its stream is still owed."""
2141    session = _make_session(tmp_path)
2142    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
2143    item.claim()
2144    item.write(b"tail" * 4)
2145    item.close()
2146    assert item.buffered == 16
2147    assert b"".join([chunk async for chunk in item.read()]) == b"tail" * 4
2148
2149
2150@contextmanager
2151def _capture_holding(
2152    session: _SoloistSession, *, fifo_bytes: int, reader_bytes: int | None
2153) -> Iterator[None]:
2154    """
2155    Give the session a real capture FIFO and a reader holding the given amounts.
2156
2157    A real pipe is used so the byte count comes from the same ioctl the backend
2158    relies on. Pass ``reader_bytes=None`` for a reader whose buffer cannot be read.
2159    """
2160    read_fd, write_fd = os.pipe()
2161    try:
2162        if fifo_bytes:
2163            os.write(write_fd, bytes(fifo_bytes))
2164        pipe = MagicMock()
2165        pipe.fileno.return_value = read_fd
2166        transport = MagicMock()
2167        transport.get_extra_info.return_value = pipe
2168        session._transport = transport
2169        reader = MagicMock(spec=[]) if reader_bytes is None else MagicMock()
2170        if reader_bytes is not None:
2171            reader._buffer = bytearray(reader_bytes)
2172        session._reader = reader
2173        yield
2174    finally:
2175        session._transport = None
2176        session._reader = None
2177        os.close(read_fd)
2178        os.close(write_fd)
2179
2180
2181def _make_provider(tmp_path: Path, setup_data: dict[str, Any] | None = None) -> SpotifyProvider:
2182    """Return a SpotifyProvider (bypassing __init__) with the given setup_data."""
2183    prov = object.__new__(SpotifyProvider)
2184    config = MagicMock(instance_id="spotify--test")
2185    config.get_value = MagicMock(return_value=None)
2186    config.values = {}
2187    prov.config = config
2188    prov.manifest = MagicMock(domain="spotify")
2189    prov.logger = MagicMock()
2190    prov.available = True
2191    mass = MagicMock()
2192    mass.storage_path = str(tmp_path / "storage")
2193    mass.cache_path = str(tmp_path / "cache")
2194    # get_setup_value reads the live setup_data blob from the store
2195    mass.config.get = MagicMock(return_value=setup_data or {})
2196    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
2197    # the store keeps values encrypted; decrypt is an identity map for the test
2198    mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
2199    prov.mass = mass
2200    return prov
2201
2202
2203def _make_backend(tmp_path: Path, setup_data: dict[str, Any] | None = None) -> SoloistBackend:
2204    """Return a SoloistBackend on a mocked provider."""
2205    return SoloistBackend(_make_provider(tmp_path, setup_data))
2206
2207
2208def _make_session(tmp_path: Path, queue_id: str | None = "player1") -> _SoloistSession:
2209    """Return a session with its process/sink/client replaced by mocks."""
2210    session = _SoloistSession(_make_backend(tmp_path), queue_id)
2211    session._sink = AsyncMock()
2212    session._client = AsyncMock()
2213    session._proc = MagicMock(returncode=None)
2214    # a session under test is past the engine's login and has claimed the
2215    # Connect device, unless a test says otherwise
2216    session._logged_in = True
2217    session._was_active = True
2218    return session
2219
2220
2221def _streamdetails_for(
2222    *,
2223    queue_id: str | None = "player1",
2224    uri: str = TRACK_A,
2225    media_type: MediaType = MediaType.TRACK,
2226) -> StreamDetails:
2227    """Return stream details for a Spotify item served by the test instance."""
2228    return StreamDetails(
2229        provider="spotify--test",
2230        item_id=uri.rsplit(":", 1)[1],
2231        audio_format=AudioFormat(content_type=ContentType.PCM_S16LE),
2232        media_type=media_type,
2233        queue_id=queue_id,
2234    )
2235
2236
2237def _make_item(tmp_path: Path, uri: str) -> _ItemAudio:
2238    """Return a bare item channel on a mocked session."""
2239    return _ItemAudio(uri, _make_session(tmp_path))
2240
2241
2242def _queue_item(uri: str, streamdetails: Any = None) -> MagicMock:
2243    """Return a queue item stand-in for a Spotify track on the test instance."""
2244    item_id = uri.rsplit(":", 1)[1]
2245    media_item = MagicMock(media_type=MediaType.TRACK, provider="spotify--test", item_id=item_id)
2246    media_item.provider_mappings = []
2247    return MagicMock(
2248        media_item=media_item,
2249        queue_item_id=f"qi-{item_id}",
2250        streamdetails=streamdetails,
2251    )
2252
2253
2254async def _wait_for(predicate: Callable[[], bool], timeout: float = 2.0) -> None:
2255    """Wait until the predicate holds, so a background task can get there."""
2256    loop = asyncio.get_running_loop()
2257    deadline = loop.time() + timeout
2258    while loop.time() < deadline:
2259        if predicate():
2260            return
2261        await asyncio.sleep(0.01)
2262    raise AssertionError("condition not met within timeout")
2263
2264
2265def _client_of(session: _SoloistSession) -> AsyncMock:
2266    """Return the session's mocked WebSocket client."""
2267    return cast("AsyncMock", session._client)
2268
2269
2270def _sink_of(session: _SoloistSession) -> AsyncMock:
2271    """Return the session's mocked capture sink."""
2272    return cast("AsyncMock", session._sink)
2273
2274
2275def _queues_of(session: _SoloistSession) -> MagicMock:
2276    """Return the mocked player_queues controller the session consults."""
2277    return cast("MagicMock", session.mass.player_queues)
2278
2279
2280def _auth_event(*, logged_in: bool, is_active: bool = True) -> SoloistEvent:
2281    """Return an auth_state event with the given login and active-device state."""
2282    return SoloistEvent(
2283        type="auth_state",
2284        data=SoloistAuthState(logged_in=logged_in, is_active=is_active),
2285        raw={},
2286    )
2287
2288
2289def _device_event(*, is_active: bool) -> SoloistEvent:
2290    """Return a device_changed event with the given active-device state."""
2291    return SoloistEvent(
2292        type="device_changed", data=SoloistDeviceChanged(is_active=is_active), raw={}
2293    )
2294
2295
2296def _playback_event(status: str, position_ms: int = 0) -> SoloistEvent:
2297    """Return a playback_state event for the current item with the given status."""
2298    return SoloistEvent(
2299        type="playback_state",
2300        data=SoloistPlaybackState(
2301            status=status,
2302            item=SoloistEntity(uri=TRACK_A, entity_type="track"),
2303            position=SoloistPosition(position_ms=position_ms, timestamp_ms=0),
2304        ),
2305        raw={},
2306    )
2307
2308
2309def _install_fake_binary_manager(monkeypatch: pytest.MonkeyPatch) -> None:
2310    """Replace the shared binary manager so no download or exec is attempted."""
2311    manager = MagicMock()
2312    manager.ensure_fresh = AsyncMock(return_value=Path("/nonexistent/soloist"))
2313    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", MagicMock(return_value=manager))
2314