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