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