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 time
16from collections.abc import AsyncGenerator, Callable
17from contextlib import suppress
18from pathlib import Path
19from typing import Any, cast
20from unittest.mock import AsyncMock, MagicMock, patch
21
22import pytest
23from music_assistant_models.enums import MediaType
24from music_assistant_models.errors import AudioError, LoginFailed
25
26from music_assistant.helpers.pulse_capture import CAPTURE_SAMPLE_RATE
27from music_assistant.models.music_provider import ProviderStreamLimitError
28from music_assistant.providers.spotify.backends import soloist as soloist_backend
29from music_assistant.providers.spotify.backends.soloist import (
30    _BYTES_PER_SECOND,
31    _FRAME_BYTES,
32    _IDLE_TIMEOUT_S,
33    _MAX_LEAD_TRIM_S,
34    SoloistBackend,
35    _CaptureShaper,
36    _ItemAudio,
37    _SoloistSession,
38    _trim_lead_silence,
39)
40from music_assistant.providers.spotify.constants import (
41    CONF_SOLOIST_API_KEY,
42    CONF_SOLOIST_CONSENT,
43    CONF_SOLOIST_SESSION_DIR,
44    SOLOIST_DATA_DIR_NAME,
45)
46from music_assistant.providers.spotify.helpers import soloist_session_present
47from music_assistant.providers.spotify.provider import SpotifyProvider
48from music_assistant.providers.spotify_connect.soloist.runtime import (
49    WS_ADDR_FILE,
50    WS_PORT_FILE,
51    SoloistAuthState,
52    SoloistEntity,
53    SoloistError,
54    SoloistEvent,
55    SoloistOptionsChanged,
56    SoloistPlaybackOptions,
57    SoloistPlaybackState,
58    SoloistPosition,
59    SoloistTrackChanged,
60    SoloistVolumeChanged,
61)
62
63TRACK_A = "spotify:track:aaa"
64TRACK_B = "spotify:track:bbb"
65
66
67def test_trim_drops_an_all_zero_chunk_within_the_bound() -> None:
68    """A pure-silence chunk inside the trim budget is dropped entirely."""
69    chunk = b"\x00" * 1024
70    trimmed, skipped = _trim_lead_silence(chunk, 0)
71    assert trimmed == b""
72    assert skipped == 1024
73
74
75def test_trim_keeps_frame_alignment_when_audio_starts_mid_chunk() -> None:
76    """Audio starting mid-chunk is cut on a sample-frame boundary."""
77    # audio starts one byte into the third frame: the trim must keep that frame whole
78    chunk = b"\x00" * (_FRAME_BYTES * 2 + 1) + b"\x01" * 64
79    trimmed, skipped = _trim_lead_silence(chunk, 0)
80    assert skipped == _FRAME_BYTES * 2
81    assert len(trimmed) % _FRAME_BYTES == 1  # the partial frame's remainder is preserved
82    assert trimmed.endswith(b"\x01" * 64)
83
84
85def test_trim_passes_silence_through_once_the_bound_is_exceeded() -> None:
86    """Beyond the trim budget, silence is genuine content and is delivered."""
87    chunk = b"\x00" * 1024
88    trimmed, skipped = _trim_lead_silence(chunk, int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND))
89    assert trimmed == chunk
90    assert skipped == 0
91
92
93def test_seek_is_confirmed_only_within_tolerance(tmp_path: Path) -> None:
94    """A position report confirms a seek only once it reaches the tolerance window."""
95    item = _make_item(tmp_path, TRACK_A)
96    item.seek_target_ms = 60_000
97    item.observe_position(50_000)
98    assert not item.seek_confirmed.is_set()
99    item.observe_position(58_500)
100    assert item.seek_confirmed.is_set()
101
102
103def test_small_seek_target_is_not_confirmed_by_a_pre_seek_zero_report(tmp_path: Path) -> None:
104    """A position-0 report before the seek lands cannot confirm a small target."""
105    item = _make_item(tmp_path, TRACK_A)
106    item.seek_target_ms = 1_500
107    item.observe_position(0)
108    assert not item.seek_confirmed.is_set()
109    item.observe_position(1_500)
110    assert item.seek_confirmed.is_set()
111
112
113def test_position_never_regresses_and_stops_at_the_cut(tmp_path: Path) -> None:
114    """The furthest position is kept, and reports after the cut belong to the next item."""
115    item = _make_item(tmp_path, TRACK_A)
116    item.observe_position(120_000)
117    # the engine's stop/idle snapshot at the end of an item reports position 0
118    item.observe_position(0)
119    assert item.last_position_ms == 120_000
120    item.close()
121    item.observe_position(5_000)
122    assert item.last_position_ms == 120_000
123
124
125async def test_item_stream_ends_where_the_session_moves_on(tmp_path: Path) -> None:
126    """An item's audio ends at the track change, and the next item's begins there."""
127    session = _make_session(tmp_path)
128    item_a = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
129    session._current = item_a
130    item_a.started.set()
131    item_a.claim()
132    item_a.write(b"a" * 16)
133    await session._observe_current(TRACK_B, 200_000)
134    item_a.write(b"late" * 4)  # written after the cut: goes nowhere
135    chunks = [chunk async for chunk in item_a.read()]
136    assert b"".join(chunks) == b"a" * 16
137    # the next item exists, carries the duration and now receives the audio
138    item_b = session._items[TRACK_B]
139    assert session.current is item_b
140    assert item_b.duration_ms == 200_000
141
142
143async def test_the_engines_restored_state_does_not_cut_a_pending_item(
144    tmp_path: Path,
145) -> None:
146    """A daemon reports the item it restored before playing ours; that is not a boundary."""
147    session = _make_session(tmp_path)
148    requested = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
149    session._current = requested
150    requested.claim()
151    # the engine announces the state it came up with, which is someone else's item
152    await session._observe_current("spotify:track:restored", 152_000)
153    # closing our item here would end its stream before it delivered anything
154    assert requested._closed is False
155    assert requested.started.is_set() is False
156    # ... and the restored item is never offered as an item's audio
157    assert session.item_for("spotify:track:restored") is None
158    # then ours starts for real, and picks up from there
159    await session._observe_current(TRACK_A, 200_000)
160    assert session.current is requested
161    assert requested.started.is_set() is True
162    requested.write(b"\x01" * 32)
163    requested.close()
164    assert b"".join([chunk async for chunk in requested.read()]) == b"\x01" * 32
165
166
167async def test_audio_read_before_the_stream_opens_is_kept(tmp_path: Path) -> None:
168    """Audio captured before an item's stream opens is buffered, not dropped."""
169    session = _make_session(tmp_path)
170    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
171    session._current = item
172    item.write(b"head" * 8)
173    item.claim()
174    item.close()
175    chunks = [chunk async for chunk in item.read()]
176    assert b"".join(chunks) == b"head" * 8
177
178
179async def test_a_channel_is_only_ever_served_once(tmp_path: Path) -> None:
180    """A consumed channel cannot be replayed, so the item needs a fresh session."""
181    session = _make_session(tmp_path)
182    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
183    item.started.set()
184    assert session.item_for(TRACK_A) is item
185    item.claim()
186    item.close()
187    item.release()
188    # this is what a queue holding the same track twice, or repeat wrapping back
189    # to the top, asks for: it must not be handed a drained channel
190    assert session.item_for(TRACK_A) is None
191
192
193async def test_an_abandoned_channel_cannot_be_continued(tmp_path: Path) -> None:
194    """A stream abandoned mid-item cannot resume where it left off either."""
195    session = _make_session(tmp_path)
196    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
197    item.started.set()
198    item.claim()
199    item.release()
200    assert session.item_for(TRACK_A) is None
201
202
203async def test_a_stuck_item_fails_instead_of_streaming_forever(tmp_path: Path) -> None:
204    """An item that runs far past its duration without a track change fails."""
205    session = _make_session(tmp_path)
206    item = _ItemAudio(TRACK_A, session)
207    item.duration_ms = 1_000
208    item.claim()
209    limit = item._overrun_limit()
210    assert limit is not None
211    item.write(b"\x01" * (limit + _FRAME_BYTES))
212    with pytest.raises(AudioError, match="never moved on"):
213        async for _ in item.read():
214            pass
215
216
217async def test_the_first_logged_out_snapshot_is_not_a_lost_pairing(tmp_path: Path) -> None:
218    """A daemon reports logged_in=False until it has restored its session."""
219    session = _make_session(tmp_path)
220    session._logged_in = None
221    await session._handle_event(_auth_event(logged_in=False))
222    # failing here would break every playback on a perfectly good pairing
223    assert session.usable is True
224    await session._handle_event(_auth_event(logged_in=True))
225    assert session.usable is True
226
227
228async def test_losing_an_established_login_fails_the_session(tmp_path: Path) -> None:
229    """A login that goes away mid-session is real, and ends the session."""
230    session = _make_session(tmp_path)
231    await session._handle_event(_auth_event(logged_in=True))
232    await session._handle_event(_auth_event(logged_in=False))
233    assert session.usable is False
234    assert session._error == "the session was logged out"
235
236
237async def test_buffering_gates_the_sink_once_demand_started(tmp_path: Path) -> None:
238    """Once PCM demand started, playing runs the sink and buffering suspends it again."""
239    session = _make_session(tmp_path)
240    session._demand_started = True
241    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
242    session._pending.append(TRACK_B)
243    sink = _sink_of(session)
244    # the sink is created suspended, so there is nothing to suspend yet
245    await session._handle_event(_playback_event("buffering"))
246    sink.suspend.assert_not_awaited()
247    await session._handle_event(_playback_event("playing"))
248    sink.resume.assert_awaited_once()
249    assert session._current is not None
250    assert session._current.playing_seen is True
251    # the engine stalling on a rebuffer keeps that silence out of the PCM
252    await session._handle_event(_playback_event("buffering"))
253    sink.suspend.assert_awaited_once()
254
255
256async def test_sink_is_not_gated_before_demand_started(tmp_path: Path) -> None:
257    """Buffering/playing before PCM demand leave the (still suspended) sink alone."""
258    session = _make_session(tmp_path)
259    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
260    sink = _sink_of(session)
261    await session._handle_event(_playback_event("buffering"))
262    await session._handle_event(_playback_event("playing"))
263    sink.suspend.assert_not_awaited()
264    sink.resume.assert_not_awaited()
265    # the status is recorded either way, so session start can decide when to resume
266    assert session._current.status == "playing"
267
268
269@pytest.mark.parametrize("end_status", ["stopped", "idle", "paused"])
270async def test_the_last_item_is_drained_rather_than_cut(tmp_path: Path, end_status: str) -> None:
271    """However the engine reports the end of a run, the last item drains and closes."""
272    session = _make_session(tmp_path)
273    session._demand_started = True
274    session._sink_running = True
275    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
276    item.duration_ms = 1_000
277    item.last_position_ms = 1_000
278    one_second = 1_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
279    sink = _sink_of(session)
280    await session._handle_event(_playback_event(end_status, position_ms=1_000))
281    # the sink stays open for now, so audio still in the FIFO can arrive...
282    sink.suspend.assert_not_awaited()
283    assert item.draining is True
284    assert item._closed is False
285    # ... but only that item's own audio is taken, never the padding silence the
286    # sink keeps rendering afterwards
287    item.write(b"\x01" * one_second)
288    item.write(b"\x00" * 4096)
289    assert item.buffered == one_second
290    await _wait_for(lambda: item._closed)
291    sink.suspend.assert_awaited_once()
292
293
294async def test_an_app_pause_midway_through_the_last_item_is_not_the_end(
295    tmp_path: Path,
296) -> None:
297    """Pausing in the Spotify app halfway through the last track must not truncate it."""
298    session = _make_session(tmp_path)
299    session._demand_started = True
300    session._sink_running = True
301    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
302    item.duration_ms = 200_000
303    item.last_position_ms = 90_000
304    await session._handle_event(_playback_event("paused", position_ms=90_000))
305    assert item.draining is False
306    assert item._closed is False
307    # treated as interference instead: the sink is gated and playback resumed
308    _sink_of(session).suspend.assert_awaited_once()
309    _client_of(session).resume.assert_awaited_once()
310
311
312async def test_a_resumed_item_cancels_its_tail_drain(tmp_path: Path) -> None:
313    """An armed drain is undone when the engine turns out to have been rebuffering."""
314    session = _make_session(tmp_path)
315    session._demand_started = True
316    session._sink_running = True
317    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
318    item.duration_ms = 200_000
319    item.last_position_ms = 199_000
320    await session._handle_event(_playback_event("stopped", position_ms=199_000))
321    armed = item.draining
322    await session._handle_event(_playback_event("playing", position_ms=199_500))
323    assert armed is True
324    assert item.draining is False
325    assert item._closed is False
326    assert item.drain_task is None
327
328
329async def test_the_cushion_is_capped_by_suspending_the_sink(tmp_path: Path) -> None:
330    """Undelivered audio is handed back as backpressure rather than piling up."""
331    session = _make_session(tmp_path)
332    session._demand_started = True
333    session._sink_running = True
334    session._engine_playing = True
335    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
336    item.claim()
337    sink = _sink_of(session)
338    await session._apply_sink_state()
339    sink.suspend.assert_not_awaited()
340    # the engine has run this far ahead of what the player has taken
341    item.write(b"\x01" * int((soloist_backend._MAX_RETAINED_S + 1) * _BYTES_PER_SECOND))
342    await session._apply_sink_state()
343    sink.suspend.assert_awaited_once()
344    assert session._backpressured is True
345    # and it comes back once the player has drained enough of it
346    item._buffered = int(soloist_backend._RESUME_RETAINED_S * _BYTES_PER_SECOND) - 1
347    await session._apply_sink_state()
348    sink.resume.assert_awaited_once()
349    assert session._backpressured is False
350
351
352async def test_a_pause_with_more_queued_suspends_the_sink(tmp_path: Path) -> None:
353    """A pause while another item is queued behind is ordinary interference, not the end."""
354    session = _make_session(tmp_path)
355    session._demand_started = True
356    session._sink_running = True
357    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
358    session._pending.append(TRACK_B)
359    await session._handle_event(_playback_event("paused"))
360    _sink_of(session).suspend.assert_awaited_once()
361
362
363async def test_nothing_is_sent_before_the_websocket_is_up(
364    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
365) -> None:
366    """Commands travel over the events socket: a published endpoint is not enough."""
367    monkeypatch.setattr(soloist_backend, "_STARTUP_TIMEOUT_S", 0.05)
368    session = _make_session(tmp_path)
369    client = _client_of(session)
370    # the endpoint file exists, but the events task has not connected yet
371    client.connected = False
372    endpoint_published = asyncio.Event()
373    endpoint_published.set()
374    with pytest.raises(AudioError, match="did not connect and log in"):
375        await session._play(TRACK_A, 0, endpoint_published)
376    client.activate.assert_not_awaited()
377    client.play.assert_not_awaited()
378
379
380async def test_nothing_is_sent_before_the_engine_has_logged_in(
381    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
382) -> None:
383    """The engine drops commands sent before it has restored its session."""
384    monkeypatch.setattr(soloist_backend, "_STARTUP_TIMEOUT_S", 0.05)
385    session = _make_session(tmp_path)
386    client = _client_of(session)
387    client.connected = True
388    # connected, but the engine has not announced its login yet
389    session._logged_in = None
390    endpoint_published = asyncio.Event()
391    endpoint_published.set()
392    with pytest.raises(AudioError, match="did not connect and log in"):
393        await session._play(TRACK_A, 0, endpoint_published)
394    client.activate.assert_not_awaited()
395    client.play.assert_not_awaited()
396
397
398async def test_startup_activates_before_it_plays(
399    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
400) -> None:
401    """A fresh daemon has to become the active device before it is told to play."""
402    session = _make_session(tmp_path)
403    client = _client_of(session)
404    client.connected = True
405    monkeypatch.setattr(session, "_await_item_ready", AsyncMock())
406    endpoint_published = asyncio.Event()
407    endpoint_published.set()
408    item = await session._play(TRACK_A, 0, endpoint_published)
409    assert item.uri == TRACK_A
410    client.activate.assert_awaited_once_with(await_result=True)
411    client.play.assert_awaited_once_with(TRACK_A)
412
413
414async def test_a_refused_start_command_reports_soloist(tmp_path: Path) -> None:
415    """A dropped start command surfaces as a Soloist error, not a raw client one."""
416    session = _make_session(tmp_path)
417    client = _client_of(session)
418    client.connected = True
419    client.activate.side_effect = SoloistError("websocket is not connected")
420    endpoint_published = asyncio.Event()
421    endpoint_published.set()
422    with pytest.raises(AudioError, match="Spotify Soloist would not start"):
423        await session._play(TRACK_A, 0, endpoint_published)
424
425
426async def test_the_engine_is_told_not_to_shuffle_or_repeat(
427    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
428) -> None:
429    """MA owns the order, and a repeating engine would never reach the item fed behind."""
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    await session._play(TRACK_A, 0, endpoint_published)
437    client.set_shuffle.assert_awaited_once_with(False)
438    client.set_repeat_context.assert_awaited_once_with(False)
439    client.set_repeat_track.assert_awaited_once_with(False)
440
441
442async def test_repeat_turned_on_from_the_app_is_pinned_back_off(tmp_path: Path) -> None:
443    """Repeat enabled in the Spotify app is undone before it can loop the item."""
444    session = _make_session(tmp_path)
445    await session._handle_event(
446        SoloistEvent(
447            type="options_changed",
448            data=SoloistOptionsChanged(
449                options=SoloistPlaybackOptions(shuffle=True, repeat="track")
450            ),
451            raw={},
452        )
453    )
454    client = _client_of(session)
455    client.set_shuffle.assert_awaited_once_with(False)
456    client.set_repeat_track.assert_awaited_once_with(False)
457    client.set_repeat_context.assert_awaited_once_with(False)
458    # options that are already off are left alone
459    client.set_shuffle.reset_mock()
460    await session._handle_event(
461        SoloistEvent(
462            type="options_changed",
463            data=SoloistOptionsChanged(options=SoloistPlaybackOptions()),
464            raw={},
465        )
466    )
467    client.set_shuffle.assert_not_awaited()
468
469
470async def test_a_busy_data_directory_is_reported_as_such(tmp_path: Path) -> None:
471    """A daemon left over from an earlier run is named, not reported as a generic failure."""
472    session = _make_session(tmp_path)
473    # the daemon's own parting complaint, which is all it gives (it exits with 1)
474    session._data_dir_busy = True
475    with pytest.raises(AudioError, match="Another Spotify Soloist session is still running"):
476        session._raise_startup_error("exited before playback started", TRACK_A)
477
478
479async def test_the_busy_marker_is_picked_up_from_the_daemon_output(tmp_path: Path) -> None:
480    """The marker is read off the daemon's stdout, with the API key still redacted."""
481    session = _make_session(tmp_path)
482    proc = MagicMock()
483    lines = [
484        'Error: another session is running for data directory "/data/x/soloist-data".',
485        "Stop the running session before starting soloist again.",
486    ]
487
488    async def _iter_stdout() -> AsyncGenerator[str]:
489        for line in lines:
490            yield line
491
492    proc.iter_stdout = _iter_stdout
493    await session._log_output(proc)
494    assert session._data_dir_busy is True
495
496
497async def test_a_pairing_that_never_logs_in_routes_through_setup(
498    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
499) -> None:
500    """A session that cannot log in sends the user back to setup, not a per-track error."""
501    session = _make_session(tmp_path)
502    unload_with_error = MagicMock()
503    monkeypatch.setattr(session.backend.provider, "unload_with_error", unload_with_error)
504    await session._handle_event(_auth_event(logged_in=False))
505    with pytest.raises(LoginFailed) as err:
506        session._raise_startup_error("timed out waiting for playback to start", TRACK_A)
507    assert err.value.translation_key == "soloist_pairing_required"
508    # ... and the provider is taken out of service, so the user is asked to redo setup
509    unload_with_error.assert_called_once()
510
511
512async def test_a_login_that_never_happened_is_not_confused_with_another_failure(
513    tmp_path: Path,
514) -> None:
515    """An unrelated failure keeps its own message even before any login was reported."""
516    session = _make_session(tmp_path)
517    session._fail("the capture sink was lost mid-stream")
518    with pytest.raises(AudioError, match="capture sink was lost"):
519        session._raise_startup_error("exited before playback started", TRACK_A)
520
521
522def test_a_seeked_item_only_expects_what_is_left_of_it(tmp_path: Path) -> None:
523    """A seeked item delivers the remainder, so its targets are based on that."""
524    session = _make_session(tmp_path)
525    item = _ItemAudio(TRACK_A, session)
526    item.duration_ms = 200_000
527    full = 200_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
528    assert item._duration_bytes() == full
529    item.seek_target_ms = 150_000
530    remainder = 50_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
531    assert item._duration_bytes() == remainder
532    # so the tail drain has a target it can actually reach
533    item.start_tail_drain()
534    item.write(b"\x01" * remainder)
535    assert item.tail_complete is True
536    # and the padding silence after it is refused
537    item.write(b"\x00" * 4096)
538    assert item.buffered == remainder
539
540
541def test_the_lead_trim_never_exceeds_its_budget() -> None:
542    """Silence beyond the budget is content, including where audio starts mid-chunk."""
543    budget = int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND)
544    # already at the budget, with a chunk whose silence runs well past it
545    chunk = b"\x00" * 4096 + b"\x01" * 64
546    trimmed, skipped = _trim_lead_silence(chunk, budget - _FRAME_BYTES)
547    assert skipped == _FRAME_BYTES
548    assert len(trimmed) == len(chunk) - _FRAME_BYTES
549
550
551async def test_a_dying_log_reader_fails_the_session(tmp_path: Path) -> None:
552    """Nothing else drains the daemon's stdout, so a dead reader must not go unnoticed."""
553    session = _make_session(tmp_path)
554
555    async def _boom() -> None:
556        raise RuntimeError("reader blew up")
557
558    session._log_task = asyncio.create_task(_boom())
559    session._log_task.add_done_callback(session._task_done)
560    await asyncio.sleep(0)
561    await _wait_for(lambda: not session.usable)
562    assert session._error is not None
563    assert "reader blew up" in session._error
564
565
566async def test_feeding_never_replaces_a_channel_already_in_use(tmp_path: Path) -> None:
567    """If the engine reaches the fed item first, its live channel must survive."""
568    session = _make_session(tmp_path, queue_id="player1")
569    streamdetails = MagicMock()
570    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
571    queues = _queues_of(session)
572    queues.get.return_value = MagicMock(current_index=0)
573    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
574    queues.get_next_item.return_value = _queue_item(TRACK_B)
575
576    async def _engine_gets_there_first(_uri: str, **_kwargs: Any) -> None:
577        # the events task advances to the fed item while the command is in flight
578        await session._observe_current(TRACK_B, 200_000)
579
580    _client_of(session).add_to_queue.side_effect = _engine_gets_there_first
581    await session.feed_after(streamdetails, TRACK_A)
582    live = session.current
583    assert live is not None
584    assert live.uri == TRACK_B
585    # the channel the reader is writing to is the one a stream will be handed
586    assert session._items[TRACK_B] is live
587    assert session.item_for(TRACK_B) is live
588    # and it is not queued as pending, because it already started
589    assert session.has_pending is False
590
591
592async def test_seeking_the_playing_item_restarts_the_session(
593    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
594) -> None:
595    """
596    A seek re-opens the item that is playing, which is a restart of the session.
597
598    A realtime source has not captured anything past the play position, so any
599    forward seek lands outside the buffer and comes back here.
600    """
601    backend = _make_backend(tmp_path)
602    backend._server = MagicMock()
603    backend._binary = Path("/nonexistent/soloist")
604    session = _SoloistSession(backend, "player1")
605    backend._session = session
606    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
607    item.started.set()
608    # its own stream is still attached when the seek re-opens it
609    item.claim()
610    stopped = AsyncMock()
611    monkeypatch.setattr(session, "stop", stopped)
612    _install_fake_binary_manager(monkeypatch)
613    monkeypatch.setattr(
614        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
615    )
616    with pytest.raises(AudioError, match="spawn"):
617        await backend._acquire(TRACK_A, 90, "player1")
618    stopped.assert_awaited_once()
619
620
621@pytest.mark.parametrize(
622    ("requested", "other_queue"),
623    [
624        # another player, whatever it asks for - including the very track this
625        # session is in the middle of delivering
626        pytest.param(TRACK_B, "player2", id="other_player"),
627        pytest.param(TRACK_A, "player2", id="other_player_same_track"),
628        # an early fetch across a boundary this session does not drive, such as a
629        # podcast episode or audiobook chapter
630        pytest.param(TRACK_B, "player1", id="unstitched_boundary"),
631    ],
632)
633async def test_a_session_in_use_is_never_cut_short(
634    tmp_path: Path, requested: str, other_queue: str
635) -> None:
636    """
637    An item the session cannot serve must not stop one it is still delivering.
638
639    Reported as capacity, so a speculative prepare gives up softly.
640    """
641    backend = _make_backend(tmp_path)
642    backend._server = MagicMock()
643    backend._binary = Path("/nonexistent/soloist")
644    session = _SoloistSession(backend, "player1")
645    backend._session = session
646    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
647    item.started.set()
648    item.claim()
649    # the session really is playing TRACK_A, so a same-track request from another
650    # player cannot be mistaken for a seek
651    session._current = item
652    with pytest.raises(ProviderStreamLimitError) as err:
653        await backend._acquire(requested, 0, other_queue)
654    # a stream-limit error so the item is not marked unplayable, but the message
655    # is about the session, not the provider's source-stream budget
656    assert err.value.limit == 1
657    assert err.value.translation_key == "soloist_session_busy"
658    # the session that was playing is untouched
659    assert backend._session is session
660    assert session.usable is True
661
662
663async def test_a_session_nobody_reads_is_replaced_for_another_item(
664    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
665) -> None:
666    """Once the other item has been released, the same request gets the session."""
667    backend = _make_backend(tmp_path)
668    backend._server = MagicMock()
669    backend._binary = Path("/nonexistent/soloist")
670    session = _SoloistSession(backend, "player1")
671    backend._session = session
672    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
673    item.started.set()
674    item.claim()
675    item.close()
676    item.release()
677    _install_fake_binary_manager(monkeypatch)
678    monkeypatch.setattr(
679        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
680    )
681    monkeypatch.setattr(session, "stop", AsyncMock())
682    with pytest.raises(AudioError, match="spawn"):
683        await backend._acquire(TRACK_B, 0, "player1")
684
685
686async def test_a_replacement_waits_for_the_old_daemon_to_be_gone(
687    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
688) -> None:
689    """The engine refuses to start while another daemon still holds its data dir."""
690    backend = _make_backend(tmp_path)
691    backend._server = MagicMock()
692    backend._binary = Path("/nonexistent/soloist")
693    session = _SoloistSession(backend, "player1")
694    backend._session = session
695    order: list[str] = []
696
697    async def _slow_stop() -> None:
698        order.append("stop-start")
699        await asyncio.sleep(0.05)
700        order.append("stop-done")
701
702    monkeypatch.setattr(session, "stop", _slow_stop)
703    _install_fake_binary_manager(monkeypatch)
704
705    async def _spawn(_self: Any, _uri: str, _seek: int) -> None:
706        order.append("spawn")
707        raise AudioError("spawn")
708
709    monkeypatch.setattr(soloist_backend._SoloistSession, "start", _spawn)
710    # the session failed, so its teardown is under way when the next item arrives
711    discard = asyncio.create_task(backend.discard_session(session))
712    await asyncio.sleep(0)
713    with pytest.raises(AudioError, match="spawn"):
714        await backend._acquire(TRACK_B, 0, "player1")
715    await discard
716    assert order == ["stop-start", "stop-done", "spawn"]
717
718
719async def test_an_idle_session_is_taken_over(
720    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
721) -> None:
722    """A session nobody is reading is replaced instead of blocking another player."""
723    backend = _make_backend(tmp_path)
724    backend._server = MagicMock()
725    backend._binary = Path("/nonexistent/soloist")
726    session = _SoloistSession(backend, "player1")
727    session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
728    backend._session = session
729    stopped = AsyncMock()
730    monkeypatch.setattr(session, "stop", stopped)
731    _install_fake_binary_manager(monkeypatch)
732    # the replacement spawn is out of scope here; only the takeover decision is
733    monkeypatch.setattr(
734        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
735    )
736    with pytest.raises(AudioError, match="spawn"):
737        await backend._acquire(TRACK_B, 0, "player2")
738    stopped.assert_awaited_once()
739
740
741def test_a_dead_session_task_fails_the_session(tmp_path: Path) -> None:
742    """A session task that dies of an unexpected error takes the session with it."""
743    session = _make_session(tmp_path)
744    task: Any = MagicMock()
745    task.cancelled.return_value = False
746    task.exception.return_value = RuntimeError("reader blew up")
747    session._task_done(task)
748    assert session.usable is False
749    assert session._error is not None
750    assert "reader blew up" in session._error
751
752
753def test_a_cancelled_session_task_is_not_a_failure(tmp_path: Path) -> None:
754    """Teardown cancels the session's tasks; that must not be reported as an error."""
755    session = _make_session(tmp_path)
756    task: Any = MagicMock()
757    task.cancelled.return_value = True
758    session._task_done(task)
759    assert session.usable is True
760
761
762async def test_failed_sink_control_fails_the_session(tmp_path: Path) -> None:
763    """A failed suspend/resume fails the session instead of leaking stall silence."""
764    session = _make_session(tmp_path)
765    session._demand_started = True
766    session._sink_running = True
767    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
768    session._pending.append(TRACK_B)
769    _sink_of(session).suspend.side_effect = RuntimeError("pactl failed")
770    await session._handle_event(_playback_event("buffering"))
771    assert session._error is not None
772    assert "capture sink control failed" in session._error
773
774
775async def test_app_pause_is_fought_with_a_resume(tmp_path: Path) -> None:
776    """A pause from the Spotify app is undone: this session has no user-facing pause."""
777    session = _make_session(tmp_path)
778    session._demand_started = True
779    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
780    session._pending.append(TRACK_B)
781    await session._handle_event(_playback_event("paused"))
782    _client_of(session).resume.assert_awaited_once()
783
784
785async def test_app_volume_change_is_pinned_back_to_unity(tmp_path: Path) -> None:
786    """An off-unity volume set from the Spotify app is pinned back to 100."""
787    session = _make_session(tmp_path)
788    await session._handle_event(
789        SoloistEvent(type="volume_changed", data=SoloistVolumeChanged(volume=40), raw={})
790    )
791    _client_of(session).set_volume.assert_awaited_once_with(100)
792    _client_of(session).set_volume.reset_mock()
793    await session._handle_event(
794        SoloistEvent(type="volume_changed", data=SoloistVolumeChanged(volume=100), raw={})
795    )
796    _client_of(session).set_volume.assert_not_awaited()
797
798
799async def test_track_change_signals_the_queue_when_it_matches_the_next_item(
800    tmp_path: Path,
801) -> None:
802    """Reaching a fed item tells the queue to start filling that item's buffer."""
803    session = _make_session(tmp_path, queue_id="player1")
804    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
805    session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
806    queues = _queues_of(session)
807    queues.get.return_value = MagicMock(next_item=_queue_item(TRACK_B), current_index=0)
808    await session._handle_event(
809        SoloistEvent(
810            type="track_changed",
811            data=SoloistTrackChanged(item=SoloistEntity(uri=TRACK_B, entity_type="track")),
812            raw={},
813        )
814    )
815    queues.prepare_next_audio_buffer.assert_called_once_with("player1")
816
817
818async def test_track_change_to_another_item_signals_nothing(tmp_path: Path) -> None:
819    """An item the queue is not asking for next must not trigger a prebuffer."""
820    session = _make_session(tmp_path, queue_id="player1")
821    session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
822    queues = _queues_of(session)
823    queues.get.return_value = MagicMock(next_item=_queue_item(TRACK_B), current_index=0)
824    await session._handle_event(
825        SoloistEvent(
826            type="track_changed",
827            data=SoloistTrackChanged(
828                item=SoloistEntity(uri="spotify:track:surprise", entity_type="track")
829            ),
830            raw={},
831        )
832    )
833    queues.prepare_next_audio_buffer.assert_not_called()
834
835
836async def test_the_follower_of_the_streamed_item_is_fed(tmp_path: Path) -> None:
837    """The item after the one being streamed is handed to the engine."""
838    session = _make_session(tmp_path, queue_id="player1")
839    streamdetails = MagicMock()
840    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
841    follower = _queue_item(TRACK_B)
842    queues = _queues_of(session)
843    queues.get.return_value = MagicMock(current_index=3)
844    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 3 else None
845    queues.get_next_item.return_value = follower
846    await session.feed_after(streamdetails, TRACK_A)
847    _client_of(session).add_to_queue.assert_awaited_once_with(TRACK_B)
848    assert TRACK_B in session._items
849    assert session.has_pending is True
850
851
852async def test_an_item_the_queue_resolved_elsewhere_is_not_fed(tmp_path: Path) -> None:
853    """A track the queue will stream from another provider must not be queued here."""
854    session = _make_session(tmp_path, queue_id="player1")
855    streamdetails = MagicMock()
856    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
857    # same track, but the queue already picked a different provider for it
858    follower = _queue_item(TRACK_B, streamdetails=MagicMock(provider="tidal--x"))
859    queues = _queues_of(session)
860    queues.get.return_value = MagicMock(current_index=0)
861    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
862    queues.get_next_item.return_value = follower
863    await session.feed_after(streamdetails, TRACK_A)
864    _client_of(session).add_to_queue.assert_not_awaited()
865
866
867async def test_skipping_to_the_fed_item_keeps_the_session(tmp_path: Path) -> None:
868    """A next-track lands on the item already fed, so the engine jumps instead of respawning."""
869    backend = _make_backend(tmp_path)
870    backend._server = MagicMock()
871    backend._binary = Path("/nonexistent/soloist")
872    session = _SoloistSession(backend, "player1")
873    session._client = AsyncMock()
874    session._logged_in = True
875    backend._session = session
876    playing = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
877    playing.started.set()
878    # fed one ahead and not reached yet, which is where a next-track goes
879    fed = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
880    session._pending.append(TRACK_B)
881
882    async def _engine_gets_there(**_kwargs: Any) -> None:
883        await session._observe_current(TRACK_B, 200_000)
884
885    _client_of(session).skip_next.side_effect = _engine_gets_there
886    got_session, got_item = await backend._acquire(TRACK_B, 0, "player1")
887    # the same session, no respawn, and the item that was already queued
888    assert got_session is session
889    assert got_item is fed
890    assert backend._session is session
891    _client_of(session).skip_next.assert_awaited_once()
892
893
894async def test_a_skip_drops_what_the_old_track_left_in_the_pipeline(tmp_path: Path) -> None:
895    """
896    Audio already rendered when the skip is issued belongs to the track left behind.
897
898    The sink and the FIFO hold a few hundred milliseconds of it, which would
899    otherwise be written to the head of the item skipped to - audible as a blip
900    of the previous track.
901    """
902    session = _make_session(tmp_path)
903    leaving = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
904    leaving.started.set()
905    target = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
906    session._pending.append(TRACK_B)
907    captured: list[bytes] = []
908
909    async def _engine_gets_there(**_kwargs: Any) -> None:
910        # the pipeline still holds the old track while the command is in flight
911        session._write_if_wanted(b"\x01" * 32)
912        await session._observe_current(TRACK_B, 200_000)
913        # from here on the audio really is the new item's
914        session._write_if_wanted(b"\x02" * 32)
915
916    _client_of(session).skip_next.side_effect = _engine_gets_there
917    await session.skip_to(target)
918    captured.extend(target._chunks)
919    assert b"".join(captured) == b"\x02" * 32
920    assert session._discard_until is None
921
922
923async def test_a_skip_the_engine_never_reaches_fails(
924    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
925) -> None:
926    """A skip that does not land is an error, not a wait for the track to end."""
927    monkeypatch.setattr(soloist_backend, "_STARTUP_TIMEOUT_S", 0.05)
928    session = _make_session(tmp_path)
929    fed = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
930    session._pending.append(TRACK_B)
931    with pytest.raises(AudioError, match="did not reach"):
932        await session.skip_to(fed)
933
934
935async def test_a_fed_item_the_engine_has_not_reached_is_not_served(tmp_path: Path) -> None:
936    """Skipping to an already-fed item must not hand over a channel that fills later."""
937    session = _make_session(tmp_path)
938    fed = session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
939    # fed, but the engine is still on the previous track
940    assert session.item_for(TRACK_B) is None
941    # once the engine gets there it is servable
942    await session._observe_current(TRACK_B, 200_000)
943    assert session.item_for(TRACK_B) is fed
944
945
946def test_the_shaper_only_emits_whole_frames() -> None:
947    """A read that ends mid-frame must never split a frame across two items."""
948    shaper = soloist_backend._CaptureShaper()
949    # the session's first bytes are infrastructure silence, and are dropped
950    assert shaper.shape(b"\x00" * 4096) == b""
951    # a mis-aligned read emits whole frames and carries the remainder
952    first = shaper.shape(b"\x01" * (_FRAME_BYTES + 3))
953    assert len(first) == _FRAME_BYTES
954    # which is then completed by the next read, losing nothing
955    second = shaper.shape(b"\x02" * (_FRAME_BYTES - 3))
956    assert len(second) == _FRAME_BYTES
957    assert second[:3] == b"\x01" * 3
958    # an aligned read passes straight through
959    assert shaper.shape(b"\x03" * _FRAME_BYTES) == b"\x03" * _FRAME_BYTES
960
961
962def test_the_shaper_trims_lead_silence_only_once() -> None:
963    """Silence after the audio has started is content, not pre-roll."""
964    shaper = soloist_backend._CaptureShaper()
965    assert shaper.shape(b"\x01" * _FRAME_BYTES) == b"\x01" * _FRAME_BYTES
966    silence = b"\x00" * _FRAME_BYTES
967    assert shaper.shape(silence) == silence
968
969
970async def test_only_whole_sample_frames_are_handed_over(
971    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
972) -> None:
973    """A read that ends mid-frame must not split a frame across two items."""
974    session = _make_session(tmp_path)
975    item = session._items[TRACK_A] = session._current = _ItemAudio(TRACK_A, session)
976    item.started.set()
977    item.claim()
978    session._demand_started = True
979    session._sink_running = True
980    # two reads that are each mis-aligned but whole together
981    reads = [b"\x01" * (_FRAME_BYTES + 3), b"\x02" * (_FRAME_BYTES - 3), b""]
982    reader = MagicMock()
983
984    async def _read(_size: int) -> bytes:
985        return reads.pop(0) if reads else b""
986
987    reader.read = _read
988    session._reader = reader
989    monkeypatch.setattr(soloist_backend, "_PACE_RATE", 1000.0)
990    await session._read_capture()
991    # every write was frame-aligned, and no byte was lost
992    assert item.buffered % _FRAME_BYTES == 0
993    assert item.buffered == _FRAME_BYTES * 2
994
995
996async def test_an_already_known_item_is_not_fed_twice(tmp_path: Path) -> None:
997    """An item the session already plays or was fed is not queued again."""
998    session = _make_session(tmp_path, queue_id="player1")
999    session._items[TRACK_B] = _ItemAudio(TRACK_B, session)
1000    streamdetails = MagicMock()
1001    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1002    queues = _queues_of(session)
1003    queues.get.return_value = MagicMock(current_index=0)
1004    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1005    queues.get_next_item.return_value = _queue_item(TRACK_B)
1006    await session.feed_after(streamdetails, TRACK_A)
1007    _client_of(session).add_to_queue.assert_not_awaited()
1008
1009
1010async def test_only_tracks_are_fed_ahead(tmp_path: Path) -> None:
1011    """A podcast episode or audiobook chapter is played on its own, never stitched."""
1012    session = _make_session(tmp_path, queue_id="player1")
1013    await session.feed_after(MagicMock(), "spotify:episode:xyz")
1014    _client_of(session).add_to_queue.assert_not_awaited()
1015
1016
1017async def test_a_non_spotify_follower_is_not_fed(tmp_path: Path) -> None:
1018    """The run simply ends where the queue leaves this provider."""
1019    session = _make_session(tmp_path, queue_id="player1")
1020    streamdetails = MagicMock()
1021    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1022    follower = MagicMock(
1023        media_item=MagicMock(media_type=MediaType.TRACK, provider="tidal--x"), streamdetails=None
1024    )
1025    follower.media_item.provider_mappings = []
1026    queues = _queues_of(session)
1027    queues.get.return_value = MagicMock(current_index=0)
1028    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1029    queues.get_next_item.return_value = follower
1030    await session.feed_after(streamdetails, TRACK_A)
1031    _client_of(session).add_to_queue.assert_not_awaited()
1032
1033
1034async def test_a_library_item_is_fed_through_its_spotify_mapping(tmp_path: Path) -> None:
1035    """A library track is fed with the item id this provider instance knows it by."""
1036    session = _make_session(tmp_path, queue_id="player1")
1037    streamdetails = MagicMock()
1038    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1039    follower = MagicMock(
1040        media_item=MagicMock(media_type=MediaType.TRACK, provider="library", item_id="42"),
1041        streamdetails=None,
1042    )
1043    follower.media_item.provider_mappings = [
1044        MagicMock(provider_instance="other--y", item_id="wrong"),
1045        MagicMock(provider_instance="spotify--test", item_id="bbb"),
1046    ]
1047    queues = _queues_of(session)
1048    queues.get.return_value = MagicMock(current_index=0)
1049    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1050    queues.get_next_item.return_value = follower
1051    await session.feed_after(streamdetails, TRACK_A)
1052    _client_of(session).add_to_queue.assert_awaited_once_with(TRACK_B)
1053
1054
1055@pytest.mark.parametrize(
1056    ("provider_option", "player_setting", "expected"),
1057    [
1058        (True, "enabled", True),
1059        # the player's own switch decides first: off means nobody normalizes,
1060        # not that the job passes to Spotify
1061        (True, "disabled", False),
1062        (False, "enabled", False),
1063        (False, "disabled", False),
1064    ],
1065)
1066def test_who_normalizes_needs_both_switches(
1067    tmp_path: Path,
1068    monkeypatch: pytest.MonkeyPatch,
1069    provider_option: bool,
1070    player_setting: str,
1071    expected: bool,
1072) -> None:
1073    """The engine normalizes only when the provider option and the player agree."""
1074    session = _make_session(tmp_path, queue_id="player1")
1075    monkeypatch.setattr(
1076        type(session.backend.provider),
1077        "spotify_normalization_configured",
1078        property(lambda _self: provider_option),
1079    )
1080    cast("MagicMock", session.mass.config).get_effective_player_queue_config_value = MagicMock(
1081        return_value=player_setting
1082    )
1083    assert session._engine_normalization_enabled() is expected
1084
1085
1086def test_a_running_session_answers_for_what_the_engine_is_doing(tmp_path: Path) -> None:
1087    """
1088    The engine reads its settings at startup, so a later toggle must not split them.
1089
1090    Otherwise the streams core would start normalizing on top of audio the engine
1091    is still normalizing, or stop while it no longer is.
1092    """
1093    backend = _make_backend(tmp_path)
1094    provider = backend.provider
1095    # nothing playing yet: the configuration is all there is to go on
1096    before_any_session = backend.session_normalizes
1097    session = _SoloistSession(backend, "player1")
1098    session.engine_normalizes = True
1099    backend._session = session
1100    while_playing = backend.session_normalizes
1101    # ... and a session that has been torn down no longer speaks for the engine
1102    session._stopped = True
1103    after_teardown = backend.session_normalizes
1104    assert before_any_session is None
1105    assert while_playing is True
1106    assert after_teardown is None
1107    assert provider.delivers_normalized_audio is provider.spotify_normalization_configured
1108
1109
1110def test_crossfade_comes_from_the_queue_preference(tmp_path: Path) -> None:
1111    """The queue's crossfade setting is handed to the engine, in milliseconds."""
1112    session = _make_session(tmp_path, queue_id="player1")
1113    _queues_of(session).get.return_value = MagicMock(queue_id="player1", crossfade_enabled=True)
1114    cast("MagicMock", session.mass.config).get_raw_core_config_value = MagicMock(return_value=6)
1115    assert session._queue_crossfade_ms() == 6000
1116
1117
1118def test_crossfade_off_is_zero(tmp_path: Path) -> None:
1119    """A queue with crossfade disabled gets an explicit zero (which clears the pref)."""
1120    session = _make_session(tmp_path, queue_id="player1")
1121    _queues_of(session).get.return_value = MagicMock(crossfade_enabled=False)
1122    assert session._queue_crossfade_ms() == 0
1123
1124
1125def test_no_queue_means_no_crossfade(tmp_path: Path) -> None:
1126    """Without a queue to read the preference from, the engine gets no crossfade."""
1127    session = _make_session(tmp_path, queue_id=None)
1128    assert session._queue_crossfade_ms() == 0
1129
1130
1131async def test_short_delivery_is_rejected_as_incomplete(tmp_path: Path) -> None:
1132    """PCM that stops well short of the item's duration is rejected."""
1133    session = _make_session(tmp_path)
1134    item = _ItemAudio(TRACK_A, session)
1135    item.playing_seen = True
1136    item.duration_ms = 200_000
1137    item.last_position_ms = 100_000
1138    with pytest.raises(AudioError, match="incomplete"):
1139        await session.validate_item(item)
1140
1141
1142async def test_a_crossfade_shortfall_is_tolerated(tmp_path: Path) -> None:
1143    """With crossfade the engine reports the item short by design; that is not a failure."""
1144    session = _make_session(tmp_path)
1145    session.crossfade_ms = 12_000
1146    item = _ItemAudio(TRACK_A, session)
1147    item.playing_seen = True
1148    item.duration_ms = 200_000
1149    # 12s of crossfade plus the ordinary tolerance
1150    item.last_position_ms = 200_000 - 21_000
1151    await session.validate_item(item)
1152
1153
1154async def test_missing_position_is_rejected_as_incomplete(tmp_path: Path) -> None:
1155    """Without any position report there is no evidence the item played out."""
1156    session = _make_session(tmp_path)
1157    item = _ItemAudio(TRACK_A, session)
1158    item.playing_seen = True
1159    item.duration_ms = 200_000
1160    with pytest.raises(AudioError, match="incomplete"):
1161        await session.validate_item(item)
1162
1163
1164async def test_short_item_cannot_pass_at_position_zero(tmp_path: Path) -> None:
1165    """The tolerance never spans a whole item, so a short item cannot pass unplayed."""
1166    session = _make_session(tmp_path)
1167    item = _ItemAudio(TRACK_A, session)
1168    item.playing_seen = True
1169    item.duration_ms = 8_000
1170    item.last_position_ms = 0
1171    with pytest.raises(AudioError, match="incomplete"):
1172        await session.validate_item(item)
1173
1174
1175async def test_an_item_that_never_played_is_rejected(tmp_path: Path) -> None:
1176    """An item the engine never reported playing is a failure, whatever was delivered."""
1177    session = _make_session(tmp_path)
1178    item = _ItemAudio(TRACK_A, session)
1179    item.duration_ms = 200_000
1180    item.last_position_ms = 200_000
1181    with pytest.raises(AudioError, match="never started playing"):
1182        await session.validate_item(item)
1183
1184
1185async def test_a_duration_less_item_is_not_judged(tmp_path: Path) -> None:
1186    """Without a duration there is nothing to judge completeness against."""
1187    session = _make_session(tmp_path)
1188    item = _ItemAudio(TRACK_A, session)
1189    item.playing_seen = True
1190    await session.validate_item(item)
1191
1192
1193def test_an_unread_session_expires(tmp_path: Path) -> None:
1194    """A session no item stream reads from is ended so its daemon does not linger."""
1195    session = _make_session(tmp_path)
1196    session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1197    session._expire_idle()
1198    assert session._idle_since is not None
1199    assert session.usable is True
1200    session._idle_since = time.monotonic() - _IDLE_TIMEOUT_S - 1
1201    session._expire_idle()
1202    assert session.usable is False
1203
1204
1205def test_a_session_being_read_never_expires(tmp_path: Path) -> None:
1206    """An item stream reading the session keeps it alive indefinitely."""
1207    session = _make_session(tmp_path)
1208    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1209    item.claim()
1210    session._idle_since = time.monotonic() - _IDLE_TIMEOUT_S * 10
1211    session._expire_idle()
1212    assert session.usable is True
1213
1214
1215def test_pre_roll_silence_is_dropped_a_whole_frame_at_a_time() -> None:
1216    """
1217    Trimming pre-roll must leave the audio on the session's frame grid.
1218
1219    A FIFO read is not always a whole number of frames, and dropping a partial
1220    one would shift every sample that follows for the rest of the session.
1221    """
1222    shaper = _CaptureShaper()
1223    # pre-roll that ends mid-frame: the real audio starts at byte 1024
1224    assert shaper.shape(b"\x00" * 1021) == b""
1225    audio = bytes(range(1, 9)) * 4
1226    shaped = shaper.shape(b"\x00" * 3 + audio)
1227    assert shaped == audio
1228    assert shaper._lead_skipped % _FRAME_BYTES == 0
1229
1230
1231async def test_a_refused_skip_does_not_leave_the_audio_discarded(tmp_path: Path) -> None:
1232    """
1233    A skip that never landed must not keep the session dropping its audio.
1234
1235    The marker silences everything the session captures, so a command that
1236    failed has to clear it on the way out.
1237    """
1238    session = _make_session(tmp_path)
1239    client = cast("MagicMock", session._client)
1240    client.skip_next = AsyncMock(side_effect=TimeoutError)
1241    item = _ItemAudio(TRACK_B, session)
1242
1243    with pytest.raises(AudioError, match="would not skip"):
1244        await session.skip_to(item)
1245
1246    assert session._discard_until is None
1247
1248
1249async def test_a_daemon_that_will_not_die_is_reported_and_released(tmp_path: Path) -> None:
1250    """A close that could not terminate the daemon still finishes the teardown."""
1251    session = _make_session(tmp_path)
1252    proc = cast("MagicMock", session._proc)
1253    proc.close = AsyncMock()
1254    # AsyncProcess.close() gives up after a handful of kill attempts
1255    proc.returncode = None
1256    with patch.object(session.logger, "warning") as warning:
1257        await session.stop()
1258    assert warning.called
1259    assert session._teardown_done is True
1260    assert session._proc is None
1261
1262
1263async def test_a_cancelled_teardown_still_closes_the_daemon(tmp_path: Path) -> None:
1264    """
1265    A cancelled teardown must leave the retry something to close.
1266
1267    Dropping the references first is how a daemon survives to hold the data
1268    directory, which every later session is then refused for.
1269    """
1270    session = _make_session(tmp_path)
1271    proc = cast("MagicMock", session._proc)
1272    sink = cast("AsyncMock", session._sink)
1273
1274    async def _never_returns() -> None:
1275        await asyncio.Event().wait()
1276
1277    proc.close = _never_returns
1278    task = asyncio.create_task(session.stop())
1279    await asyncio.sleep(0.01)
1280    task.cancel()
1281    with suppress(asyncio.CancelledError):
1282        await task
1283    # the teardown did not finish, so nothing was dropped and it can be redone
1284    unfinished = session._teardown_done
1285    kept_proc = session._proc
1286    kept_sink = session._sink
1287    proc.close = AsyncMock()
1288    proc.returncode = 0
1289    await session.stop()
1290    assert unfinished is False
1291    assert kept_proc is proc
1292    assert kept_sink is sink
1293    assert session._teardown_done is True
1294    assert session._proc is None
1295    assert session._sink is None
1296    proc.close.assert_awaited()
1297    sink.unload.assert_awaited()
1298
1299
1300def test_a_failed_session_is_torn_down(tmp_path: Path) -> None:
1301    """A session that fails is discarded, so its daemon does not keep playing to nobody."""
1302    session = _make_session(tmp_path)
1303    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1304    item.claim()
1305    session._fail("audio stalled")
1306    assert session.usable is False
1307    # every waiting item is released and the teardown is scheduled
1308    assert item._closed is True
1309    # a startup wait must not sit out its timeout on a session that already failed
1310    assert item.started.is_set() is True
1311    discard = cast("MagicMock", session.mass.create_task)
1312    discard.assert_called_once_with(session.backend.discard_session, session)
1313    # a second failure does not queue a second teardown
1314    session._fail("and again")
1315    assert session._error == "audio stalled"
1316    assert discard.call_count == 1
1317
1318
1319async def test_an_item_the_engine_skipped_past_fails_instead_of_hanging(
1320    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1321) -> None:
1322    """A claimed channel the engine never reaches gives up rather than blocking forever."""
1323    monkeypatch.setattr(soloist_backend, "_READ_SLICE_S", 0.01)
1324    monkeypatch.setattr(soloist_backend, "_STALL_TIMEOUT_S", 0.05)
1325    session = _make_session(tmp_path)
1326    item = session._items[TRACK_A] = _ItemAudio(TRACK_A, session)
1327    item.claim()
1328    # the engine is playing something else, so nothing is ever written here
1329    session._items["spotify:track:other"] = session._current = _ItemAudio(
1330        "spotify:track:other", session
1331    )
1332    with pytest.raises(AudioError, match="no audio"):
1333        async for _ in item.read():
1334            pass
1335
1336
1337async def test_adopt_paired_session_copies_into_the_canonical_dir(
1338    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1339) -> None:
1340    """A session paired by the setup flow is adopted into the per-instance data dir."""
1341    storage = tmp_path / "storage"
1342    pending = storage / "spotify" / "pairing" / "flow1"
1343    pending.mkdir(parents=True)
1344    (pending / "session.bin").write_bytes(b"session")
1345    prov = _make_provider(tmp_path, {CONF_SOLOIST_SESSION_DIR: "spotify/pairing/flow1"})
1346    update_setup_data = MagicMock()
1347    monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
1348    backend = SoloistBackend(prov)
1349    await backend._adopt_paired_session()
1350    canonical = storage / "spotify" / "spotify--test" / SOLOIST_DATA_DIR_NAME
1351    assert (canonical / "session.bin").read_bytes() == b"session"
1352    # a copy, not a move: the flow-private source must survive a failed
1353    # provider load so the setup flow can retry (the flow removes it at its end)
1354    assert (pending / "session.bin").exists()
1355    update_setup_data.assert_called_once_with(CONF_SOLOIST_SESSION_DIR, None)
1356
1357
1358def test_the_engine_is_told_not_to_normalize(tmp_path: Path) -> None:
1359    """MA normalizes this audio itself, so the engine's own normalization is switched off."""
1360    backend = _make_backend(tmp_path)
1361    prefs = backend._data_dir / "settings" / "Users" / "alice-user" / "prefs"
1362    prefs.parent.mkdir(parents=True)
1363    prefs.write_text("some.engine.key=1\n", encoding="utf-8")
1364    backend._prepare_data_dir(8000, normalize=False)
1365    content = prefs.read_text(encoding="utf-8").splitlines()
1366    assert "some.engine.key=1" in content
1367    assert "audio.normalize_v2=false" in content
1368    assert "audio.crossfade_v2=true" in content
1369    assert "audio.crossfade.time_v2=8000" in content
1370    # the ceiling is stated rather than left to the engine's own default
1371    assert "audio.play_bitrate_enumeration=5" in content
1372    assert "audio.play_bitrate_non_metered_enumeration=5" in content
1373    assert "audio.play_bitrate_non_metered_migrated=true" in content
1374
1375
1376def test_disabling_crossfade_writes_the_boolean(tmp_path: Path) -> None:
1377    """Crossfade off is written explicitly, so a stale 'on' cannot survive."""
1378    backend = _make_backend(tmp_path)
1379    prefs = backend._data_dir / "settings" / "prefs"
1380    prefs.parent.mkdir(parents=True)
1381    prefs.write_text("audio.crossfade_v2=true\naudio.crossfade.time_v2=8000\n", encoding="utf-8")
1382    backend._prepare_data_dir(0, normalize=False)
1383    content = prefs.read_text(encoding="utf-8").splitlines()
1384    assert "audio.crossfade_v2=false" in content
1385    assert not any(line.startswith("audio.crossfade.time_v2") for line in content)
1386
1387
1388async def test_setup_requires_an_api_key(tmp_path: Path) -> None:
1389    """Without a stored API key the user must be sent back through the setup flow."""
1390    backend = _make_backend(tmp_path)
1391    with pytest.raises(LoginFailed) as err:
1392        await backend.setup()
1393    assert err.value.translation_key == "soloist_pairing_required"
1394
1395
1396async def test_setup_requires_a_paired_session(
1397    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1398) -> None:
1399    """An API key without a paired session also routes back to the setup flow."""
1400    backend = _make_backend(tmp_path, {CONF_SOLOIST_API_KEY: "k" * 20, CONF_SOLOIST_CONSENT: True})
1401    _install_fake_binary_manager(monkeypatch)
1402    with pytest.raises(LoginFailed) as err:
1403        await backend.setup()
1404    assert err.value.translation_key == "soloist_pairing_required"
1405
1406
1407async def test_streaming_without_setup_is_refused(tmp_path: Path) -> None:
1408    """A backend whose setup never ran refuses to stream instead of half-starting."""
1409    backend = _make_backend(tmp_path)
1410    with pytest.raises(AudioError, match="not started"):
1411        async for _ in backend.stream_spotify_uri(TRACK_A):
1412            pass
1413
1414
1415def test_session_present_detection(tmp_path: Path) -> None:
1416    """Only a dir holding something besides the WS endpoint files counts as paired."""
1417    data_dir = tmp_path / "soloist-data"
1418    assert soloist_session_present(data_dir) is False
1419    data_dir.mkdir()
1420    (data_dir / WS_ADDR_FILE).write_text("127.0.0.1", encoding="utf-8")
1421    (data_dir / WS_PORT_FILE).write_text("1234", encoding="utf-8")
1422    assert soloist_session_present(data_dir) is False
1423    (data_dir / "session.bin").write_bytes(b"x")
1424    assert soloist_session_present(data_dir) is True
1425
1426
1427def _make_provider(tmp_path: Path, setup_data: dict[str, Any] | None = None) -> SpotifyProvider:
1428    """Return a SpotifyProvider (bypassing __init__) with the given setup_data."""
1429    prov = object.__new__(SpotifyProvider)
1430    config = MagicMock(instance_id="spotify--test")
1431    config.get_value = MagicMock(return_value=None)
1432    config.values = {}
1433    prov.config = config
1434    prov.manifest = MagicMock(domain="spotify")
1435    prov.logger = MagicMock()
1436    prov.available = True
1437    mass = MagicMock()
1438    mass.storage_path = str(tmp_path / "storage")
1439    mass.cache_path = str(tmp_path / "cache")
1440    # get_setup_value reads the live setup_data blob from the store
1441    mass.config.get = MagicMock(return_value=setup_data or {})
1442    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
1443    # the store keeps values encrypted; decrypt is an identity map for the test
1444    mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
1445    prov.mass = mass
1446    return prov
1447
1448
1449def _make_backend(tmp_path: Path, setup_data: dict[str, Any] | None = None) -> SoloistBackend:
1450    """Return a SoloistBackend on a mocked provider."""
1451    return SoloistBackend(_make_provider(tmp_path, setup_data))
1452
1453
1454def _make_session(tmp_path: Path, queue_id: str | None = "player1") -> _SoloistSession:
1455    """Return a session with its process/sink/client replaced by mocks."""
1456    session = _SoloistSession(_make_backend(tmp_path), queue_id)
1457    session._sink = AsyncMock()
1458    session._client = AsyncMock()
1459    session._proc = MagicMock(returncode=None)
1460    # a session under test is past the engine's login unless a test says otherwise
1461    session._logged_in = True
1462    return session
1463
1464
1465def _make_item(tmp_path: Path, uri: str) -> _ItemAudio:
1466    """Return a bare item channel on a mocked session."""
1467    return _ItemAudio(uri, _make_session(tmp_path))
1468
1469
1470def _queue_item(uri: str, streamdetails: Any = None) -> MagicMock:
1471    """Return a queue item stand-in for a Spotify track on the test instance."""
1472    item_id = uri.rsplit(":", 1)[1]
1473    media_item = MagicMock(media_type=MediaType.TRACK, provider="spotify--test", item_id=item_id)
1474    media_item.provider_mappings = []
1475    return MagicMock(
1476        media_item=media_item,
1477        queue_item_id=f"qi-{item_id}",
1478        streamdetails=streamdetails,
1479    )
1480
1481
1482async def _wait_for(predicate: Callable[[], bool], timeout: float = 2.0) -> None:
1483    """Wait until the predicate holds, so a background task can get there."""
1484    loop = asyncio.get_running_loop()
1485    deadline = loop.time() + timeout
1486    while loop.time() < deadline:
1487        if predicate():
1488            return
1489        await asyncio.sleep(0.01)
1490    raise AssertionError("condition not met within timeout")
1491
1492
1493def _client_of(session: _SoloistSession) -> AsyncMock:
1494    """Return the session's mocked WebSocket client."""
1495    return cast("AsyncMock", session._client)
1496
1497
1498def _sink_of(session: _SoloistSession) -> AsyncMock:
1499    """Return the session's mocked capture sink."""
1500    return cast("AsyncMock", session._sink)
1501
1502
1503def _queues_of(session: _SoloistSession) -> MagicMock:
1504    """Return the mocked player_queues controller the session consults."""
1505    return cast("MagicMock", session.mass.player_queues)
1506
1507
1508def _auth_event(*, logged_in: bool) -> SoloistEvent:
1509    """Return an auth_state event with the given login state."""
1510    return SoloistEvent(
1511        type="auth_state",
1512        data=SoloistAuthState(logged_in=logged_in, is_active=False),
1513        raw={},
1514    )
1515
1516
1517def _playback_event(status: str, position_ms: int = 0) -> SoloistEvent:
1518    """Return a playback_state event for the current item with the given status."""
1519    return SoloistEvent(
1520        type="playback_state",
1521        data=SoloistPlaybackState(
1522            status=status,
1523            item=SoloistEntity(uri=TRACK_A, entity_type="track"),
1524            position=SoloistPosition(position_ms=position_ms, timestamp_ms=0),
1525        ),
1526        raw={},
1527    )
1528
1529
1530def _install_fake_binary_manager(monkeypatch: pytest.MonkeyPatch) -> None:
1531    """Replace the shared binary manager so no download or exec is attempted."""
1532    manager = MagicMock()
1533    manager.ensure_fresh = AsyncMock(return_value=Path("/nonexistent/soloist"))
1534    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", MagicMock(return_value=manager))
1535