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.controllers.streams.audio_buffer import BUFFER_READY_TIMEOUT
30from music_assistant.helpers.pulse_capture import CAPTURE_SAMPLE_RATE
31from music_assistant.models.music_provider import ProviderStreamLimitError
32from music_assistant.providers.spotify.backends import soloist as soloist_backend
33from music_assistant.providers.spotify.backends.soloist import (
34    _BYTES_PER_SECOND,
35    _FRAME_BYTES,
36    _IDLE_TIMEOUT_S,
37    _ITEM_OVERRUN_S,
38    _JUMP_TIMEOUT_S,
39    _MAX_APP_PAUSE_RESUMES,
40    _MAX_LEAD_TRIM_S,
41    _READ_CHUNK_SIZE,
42    SoloistAppControl,
43    SoloistAppControlError,
44    SoloistBackend,
45    _CaptureShaper,
46    _ItemAudio,
47    _SoloistSession,
48    _trim_lead_silence,
49)
50from music_assistant.providers.spotify.constants import (
51    CONF_SOLOIST_API_KEY,
52    CONF_SOLOIST_CONSENT,
53    CONF_SOLOIST_SESSION_DIR,
54    SOLOIST_DATA_DIR_NAME,
55    SOLOIST_DEVICE_NAME,
56)
57from music_assistant.providers.spotify.helpers import soloist_session_present
58from music_assistant.providers.spotify.provider import SpotifyProvider
59from music_assistant.providers.spotify_connect.provider import DEFAULT_PUBLISH_NAME
60from music_assistant.providers.spotify_connect.soloist.runtime import (
61    WS_ADDR_FILE,
62    WS_PORT_FILE,
63    SoloistAuthState,
64    SoloistDeviceChanged,
65    SoloistEntity,
66    SoloistError,
67    SoloistEvent,
68    SoloistOptionsChanged,
69    SoloistPlaybackOptions,
70    SoloistPlaybackState,
71    SoloistPosition,
72    SoloistPositionSync,
73    SoloistTrackChanged,
74    SoloistVolumeChanged,
75)
76
77TRACK_A = "spotify:track:aaa"
78TRACK_B = "spotify:track:bbb"
79TRACK_C = "spotify:track:ccc"
80
81
82def test_trim_drops_an_all_zero_chunk_within_the_bound() -> None:
83    """A pure-silence chunk inside the trim budget is dropped entirely."""
84    chunk = b"\x00" * 1024
85    trimmed, skipped = _trim_lead_silence(chunk, 0)
86    assert trimmed == b""
87    assert skipped == 1024
88
89
90def test_trim_keeps_frame_alignment_when_audio_starts_mid_chunk() -> None:
91    """Audio starting mid-chunk is cut on a sample-frame boundary."""
92    # audio starts one byte into the third frame: the trim must keep that frame whole
93    chunk = b"\x00" * (_FRAME_BYTES * 2 + 1) + b"\x01" * 64
94    trimmed, skipped = _trim_lead_silence(chunk, 0)
95    assert skipped == _FRAME_BYTES * 2
96    assert len(trimmed) % _FRAME_BYTES == 1  # the partial frame's remainder is preserved
97    assert trimmed.endswith(b"\x01" * 64)
98
99
100def test_trim_passes_silence_through_once_the_bound_is_exceeded() -> None:
101    """Beyond the trim budget, silence is genuine content and is delivered."""
102    chunk = b"\x00" * 1024
103    trimmed, skipped = _trim_lead_silence(chunk, int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND))
104    assert trimmed == chunk
105    assert skipped == 0
106
107
108def test_seek_is_confirmed_only_within_tolerance(tmp_path: Path) -> None:
109    """A position report confirms a seek only once it reaches the tolerance window."""
110    item = _make_item(tmp_path, TRACK_A)
111    item.arm_seek(60_000)
112    item.observe_position(50_000)
113    assert not item.seek_confirmed.is_set()
114    item.observe_position(58_500)
115    assert item.seek_confirmed.is_set()
116    assert item.started_at_ms == 58_500
117
118
119def test_small_seek_target_is_not_confirmed_by_a_pre_seek_zero_report(tmp_path: Path) -> None:
120    """A position-0 report before the seek lands cannot confirm a small target."""
121    item = _make_item(tmp_path, TRACK_A)
122    item.arm_seek(1_500)
123    item.observe_position(0)
124    assert not item.seek_confirmed.is_set()
125    item.observe_position(1_500)
126    assert item.seek_confirmed.is_set()
127
128
129def test_a_small_seek_is_confirmed_without_a_report_of_exactly_zero(tmp_path: Path) -> None:
130    """A target inside the tolerance window has no room below it to be anchored on."""
131    item = _make_item(tmp_path, TRACK_A)
132    # the engine restored this item a second in, so the seek is short enough
133    # that no report can fall below its tolerance window
134    item.observe_position(1_200)
135    item.arm_seek(2_000)
136    item.observe_position(400)
137    item.observe_position(2_000)
138    assert item.seek_confirmed.is_set()
139
140
141def test_the_restored_position_of_the_same_item_cannot_confirm_a_seek(tmp_path: Path) -> None:
142    """The state a fresh session restores does not pass for the seek landing."""
143    item = _make_item(tmp_path, TRACK_A)
144    item.duration_ms = 176_000
145    # the engine restores the account's last state: this very item, sitting at
146    # the position the seek is aiming for
147    item.observe_position(117_000)
148    item.arm_seek(117_000)
149    item.observe_position(117_000)
150    assert not item.seek_confirmed.is_set()
151    # only once the engine has reloaded the track does its seek count
152    item.observe_position(0)
153    item.observe_position(117_000)
154    assert item.seek_confirmed.is_set()
155
156
157def test_a_backward_seek_is_confirmed_below_where_the_engine_was(tmp_path: Path) -> None:
158    """Seeking back into an item confirms on the target, not on where it came from."""
159    item = _make_item(tmp_path, TRACK_A)
160    # the engine restored this item well past the point being seeked back to
161    item.observe_position(117_000)
162    item.arm_seek(30_000)
163    item.observe_position(0)
164    item.observe_position(30_000)
165    assert item.seek_confirmed.is_set()
166    assert item.started_at_ms == 30_000
167
168
169def test_position_never_regresses_and_stops_at_the_cut(tmp_path: Path) -> None:
170    """The furthest position is kept, and reports after the cut belong to the next item."""
171    item = _make_item(tmp_path, TRACK_A)
172    item.observe_position(120_000)
173    # the engine's stop/idle snapshot at the end of an item reports position 0
174    item.observe_position(0)
175    assert item.last_position_ms == 120_000
176    item.close()
177    item.observe_position(5_000)
178    assert item.last_position_ms == 120_000
179
180
181async def test_item_stream_ends_where_the_session_moves_on(tmp_path: Path) -> None:
182    """An item's audio ends at the track change, and the next item's begins there."""
183    session = _make_session(tmp_path)
184    item_a = session._open_channel(TRACK_A)
185    session._current = item_a
186    item_a.started.set()
187    item_a.claim()
188    item_a.write(b"a" * 16)
189    await session._observe_current(TRACK_B, 200_000, track_changed=True)
190    item_a.write(b"late" * 4)  # written after the cut: goes nowhere
191    chunks = [chunk async for chunk in item_a.read()]
192    assert b"".join(chunks) == b"a" * 16
193    # the next item exists, carries the duration and now receives the audio
194    item_b = session.current
195    assert item_b is not None
196    assert item_b.uri == TRACK_B
197    assert item_b.duration_ms == 200_000
198
199
200async def test_the_engines_restored_state_does_not_cut_a_pending_item(
201    tmp_path: Path,
202) -> None:
203    """A daemon reports the item it restored before playing ours; that is not a boundary."""
204    session = _make_session(tmp_path)
205    requested = session._open_channel(TRACK_A)
206    session._current = requested
207    requested.claim()
208    # the engine announces the state it came up with, which is someone else's item
209    await session._observe_current("spotify:track:restored", 152_000, track_changed=False)
210    # closing our item here would end its stream before it delivered anything
211    assert requested._closed is False
212    assert requested.started.is_set() is False
213    # ... and the restored item is never offered as an item's audio
214    assert session.item_for("spotify:track:restored") is None
215    # then ours starts for real, and picks up from there
216    await session._observe_current(TRACK_A, 200_000, track_changed=True)
217    assert session.current is requested
218    assert requested.started.is_set() is True
219    requested.write(b"\x01" * 32)
220    requested.close()
221    assert b"".join([chunk async for chunk in requested.read()]) == b"\x01" * 32
222
223
224async def test_leaving_the_engines_restored_item_is_not_a_takeover(tmp_path: Path) -> None:
225    """The restored item is part-way through a track, and we are about to leave it."""
226    session = _make_session(tmp_path)
227    # as _play leaves it: the channel exists, its stream is not reading it yet
228    requested = session._open_channel(TRACK_A)
229    session._current = requested
230    await session._observe_current("spotify:track:restored", 152_000, track_changed=False)
231    restored = session.current
232    assert restored is not None
233    restored.observe_position(20_000)
234
235    # our own play() lands and the engine leaves the restored item for ours
236    await session._observe_current(TRACK_A, 200_000, track_changed=True)
237    assert session.usable is True
238    assert session.current is requested
239
240
241async def test_audio_read_before_the_stream_opens_is_kept(tmp_path: Path) -> None:
242    """Audio captured before an item's stream opens is buffered, not dropped."""
243    session = _make_session(tmp_path)
244    item = session._open_channel(TRACK_A)
245    session._current = item
246    item.write(b"head" * 8)
247    item.claim()
248    item.close()
249    chunks = [chunk async for chunk in item.read()]
250    assert b"".join(chunks) == b"head" * 8
251
252
253async def test_a_channel_is_only_ever_served_once(tmp_path: Path) -> None:
254    """A consumed channel cannot be replayed, so the item needs a fresh session."""
255    session = _make_session(tmp_path)
256    item = session._current = session._open_channel(TRACK_A)
257    item.started.set()
258    assert session.item_for(TRACK_A) is item
259    item.claim()
260    item.close()
261    item.release()
262    # this is what a queue holding the same track twice, or repeat wrapping back
263    # to the top, asks for: it must not be handed a drained channel
264    assert session.item_for(TRACK_A) is None
265
266
267async def test_an_abandoned_channel_cannot_be_continued(tmp_path: Path) -> None:
268    """A stream abandoned mid-item cannot resume where it left off either."""
269    session = _make_session(tmp_path)
270    item = session._current = session._open_channel(TRACK_A)
271    item.started.set()
272    item.claim()
273    item.release()
274    assert session.item_for(TRACK_A) is None
275
276
277async def test_a_stuck_item_fails_instead_of_streaming_forever(tmp_path: Path) -> None:
278    """An item that runs far past its duration without a track change fails."""
279    session = _make_session(tmp_path)
280    item = _ItemAudio(TRACK_A, session)
281    item.duration_ms = 1_000
282    item.claim()
283    limit = item._overrun_limit()
284    assert limit is not None
285    item.write(b"\x01" * (limit + _FRAME_BYTES))
286    with pytest.raises(AudioError, match="never moved on"):
287        async for _ in item.read():
288            pass
289
290
291async def test_the_first_logged_out_snapshot_is_not_a_lost_pairing(tmp_path: Path) -> None:
292    """A daemon reports logged_in=False until it has restored its session."""
293    session = _make_session(tmp_path)
294    session._logged_in = None
295    session._was_active = False
296    await session._handle_event(_auth_event(logged_in=False, is_active=False))
297    # failing here would break every playback on a perfectly good pairing
298    assert session.usable is True
299    await session._handle_event(_auth_event(logged_in=True, is_active=False))
300    assert session.usable is True
301
302
303async def test_losing_an_established_login_fails_the_session(tmp_path: Path) -> None:
304    """A login that goes away mid-session is real, and ends the session."""
305    session = _make_session(tmp_path)
306    await session._handle_event(_auth_event(logged_in=True))
307    await session._handle_event(_auth_event(logged_in=False))
308    assert session.usable is False
309    assert session._error == "the session was logged out"
310
311
312async def test_buffering_gates_the_sink_once_demand_started(tmp_path: Path) -> None:
313    """Once PCM demand started, playing runs the sink and buffering suspends it again."""
314    session = _make_session(tmp_path)
315    session._demand_started = True
316    session._current = session._open_channel(TRACK_A)
317    _feed(session, TRACK_B)
318    sink = _sink_of(session)
319    # the sink is created suspended, so there is nothing to suspend yet
320    await session._handle_event(_playback_event("buffering"))
321    sink.suspend.assert_not_awaited()
322    await session._handle_event(_playback_event("playing"))
323    sink.resume.assert_awaited_once()
324    assert session._current is not None
325    assert session._current.playing_seen is True
326    # the engine stalling on a rebuffer keeps that silence out of the PCM
327    await session._handle_event(_playback_event("buffering"))
328    sink.suspend.assert_awaited_once()
329
330
331async def test_sink_is_not_gated_before_demand_started(tmp_path: Path) -> None:
332    """Buffering/playing before PCM demand leave the (still suspended) sink alone."""
333    session = _make_session(tmp_path)
334    session._current = session._open_channel(TRACK_A)
335    sink = _sink_of(session)
336    await session._handle_event(_playback_event("buffering"))
337    await session._handle_event(_playback_event("playing"))
338    sink.suspend.assert_not_awaited()
339    sink.resume.assert_not_awaited()
340    # the status is recorded either way, so session start can decide when to resume
341    assert session._current.status == "playing"
342
343
344@pytest.mark.parametrize("end_status", ["stopped", "idle", "paused"])
345async def test_the_last_item_is_drained_rather_than_cut(tmp_path: Path, end_status: str) -> None:
346    """However the engine reports the end of a run, the last item drains and closes."""
347    session = _make_session(tmp_path)
348    session._demand_started = True
349    session._sink_running = True
350    item = session._current = session._open_channel(TRACK_A)
351    item.duration_ms = 1_000
352    item.last_position_ms = 1_000
353    one_second = 1_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
354    sink = _sink_of(session)
355    await session._handle_event(_playback_event(end_status, position_ms=1_000))
356    # the sink stays open for now, so audio still in the FIFO can arrive...
357    sink.suspend.assert_not_awaited()
358    assert item.draining is True
359    assert item._closed is False
360    # ... but only that item's own audio is taken, never the padding silence the
361    # sink keeps rendering afterwards
362    item.write(b"\x01" * one_second)
363    item.write(b"\x00" * 4096)
364    assert item.buffered == one_second
365    await _wait_for(lambda: item._closed)
366    sink.suspend.assert_awaited_once()
367
368
369async def test_an_app_pause_midway_through_the_last_item_is_not_the_end(
370    tmp_path: Path,
371) -> None:
372    """Pausing in the Spotify app halfway through the last track must not truncate it."""
373    session = _make_session(tmp_path)
374    session._demand_started = True
375    session._sink_running = True
376    item = session._current = session._open_channel(TRACK_A)
377    item.duration_ms = 200_000
378    item.last_position_ms = 90_000
379    await session._handle_event(_playback_event("paused", position_ms=90_000))
380    assert item.draining is False
381    assert item._closed is False
382    # treated as interference instead: the sink is gated and playback resumed
383    _sink_of(session).suspend.assert_awaited_once()
384    _client_of(session).resume.assert_awaited_once()
385
386
387async def test_a_resumed_item_cancels_its_tail_drain(tmp_path: Path) -> None:
388    """An armed drain is undone when the engine turns out to have been rebuffering."""
389    session = _make_session(tmp_path)
390    session._demand_started = True
391    session._sink_running = True
392    item = session._current = session._open_channel(TRACK_A)
393    item.duration_ms = 200_000
394    item.last_position_ms = 199_000
395    await session._handle_event(_playback_event("stopped", position_ms=199_000))
396    armed = item.draining
397    await session._handle_event(_playback_event("playing", position_ms=199_500))
398    assert armed is True
399    assert item.draining is False
400    assert item._closed is False
401    assert item.drain_task is None
402
403
404async def test_the_cushion_is_capped_by_suspending_the_sink(tmp_path: Path) -> None:
405    """Undelivered audio is handed back as backpressure rather than piling up."""
406    session = _make_session(tmp_path)
407    session._demand_started = True
408    session._sink_running = True
409    session._engine_playing = True
410    item = session._current = session._open_channel(TRACK_A)
411    item.claim()
412    sink = _sink_of(session)
413    await session._apply_sink_state()
414    sink.suspend.assert_not_awaited()
415    # the engine has run this far ahead of what the player has taken
416    item.write(b"\x01" * int((soloist_backend._MAX_RETAINED_S + 1) * _BYTES_PER_SECOND))
417    await session._apply_sink_state()
418    sink.suspend.assert_awaited_once()
419    assert session._backpressured is True
420    # and it comes back once the player has drained enough of it
421    item._buffered = int(soloist_backend._RESUME_RETAINED_S * _BYTES_PER_SECOND) - 1
422    await session._apply_sink_state()
423    sink.resume.assert_awaited_once()
424    assert session._backpressured is False
425
426
427async def test_a_pause_with_more_queued_suspends_the_sink(tmp_path: Path) -> None:
428    """A pause while another item is queued behind is ordinary interference, not the end."""
429    session = _make_session(tmp_path)
430    session._demand_started = True
431    session._sink_running = True
432    session._current = session._open_channel(TRACK_A)
433    _feed(session, TRACK_B)
434    await session._handle_event(_playback_event("paused"))
435    _sink_of(session).suspend.assert_awaited_once()
436
437
438async def test_nothing_is_sent_before_the_websocket_is_up(
439    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
440) -> None:
441    """Commands travel over the events socket: a published endpoint is not enough."""
442    monkeypatch.setattr(soloist_backend, "_STARTUP_TIMEOUT_S", 0.05)
443    session = _make_session(tmp_path)
444    client = _client_of(session)
445    # the endpoint file exists, but the events task has not connected yet
446    client.connected = False
447    endpoint_published = asyncio.Event()
448    endpoint_published.set()
449    with pytest.raises(AudioError, match="did not connect and log in"):
450        await session._play(TRACK_A, 0, endpoint_published)
451    client.activate.assert_not_awaited()
452    client.play.assert_not_awaited()
453
454
455async def test_nothing_is_sent_before_the_engine_has_logged_in(
456    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
457) -> None:
458    """The engine drops commands sent before it has restored its session."""
459    monkeypatch.setattr(soloist_backend, "_STARTUP_TIMEOUT_S", 0.05)
460    session = _make_session(tmp_path)
461    client = _client_of(session)
462    client.connected = True
463    # connected, but the engine has not announced its login yet
464    session._logged_in = None
465    endpoint_published = asyncio.Event()
466    endpoint_published.set()
467    with pytest.raises(AudioError, match="did not connect and log in"):
468        await session._play(TRACK_A, 0, endpoint_published)
469    client.activate.assert_not_awaited()
470    client.play.assert_not_awaited()
471
472
473async def test_startup_activates_before_it_plays(
474    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
475) -> None:
476    """A fresh daemon has to become the active device before it is told to play."""
477    session = _make_session(tmp_path)
478    client = _client_of(session)
479    client.connected = True
480    monkeypatch.setattr(session, "_await_item_ready", AsyncMock())
481    endpoint_published = asyncio.Event()
482    endpoint_published.set()
483    item = await session._play(TRACK_A, 0, endpoint_published)
484    assert item.uri == TRACK_A
485    client.activate.assert_awaited_once_with(await_result=True)
486    client.play.assert_awaited_once_with(TRACK_A)
487
488
489async def test_a_takeover_between_activate_and_play_stops_the_start(tmp_path: Path) -> None:
490    """Playing here would claim the device straight back off wherever the user moved to."""
491    session = _make_session(tmp_path)
492    session._was_active = False
493    client = _client_of(session)
494
495    async def _take_over(*_args: Any, **_kwargs: Any) -> None:
496        session._observe_active_device(is_active=False)
497
498    client.set_repeat_track.side_effect = _take_over
499    ready = asyncio.Event()
500    ready.set()
501    with pytest.raises(SoloistAppControlError):
502        await session._play(TRACK_A, 0, ready)
503    client.play.assert_not_awaited()
504
505
506async def test_a_refused_start_command_reports_soloist(tmp_path: Path) -> None:
507    """A dropped start command surfaces as a Soloist error, not a raw client one."""
508    session = _make_session(tmp_path)
509    client = _client_of(session)
510    client.connected = True
511    client.activate.side_effect = SoloistError("websocket is not connected")
512    endpoint_published = asyncio.Event()
513    endpoint_published.set()
514    with pytest.raises(AudioError, match="Spotify Soloist would not start"):
515        await session._play(TRACK_A, 0, endpoint_published)
516
517
518async def test_the_engine_is_told_not_to_shuffle_or_repeat(
519    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
520) -> None:
521    """MA owns the order, and a repeating engine would never reach the item fed behind."""
522    session = _make_session(tmp_path)
523    client = _client_of(session)
524    client.connected = True
525    monkeypatch.setattr(session, "_await_item_ready", AsyncMock())
526    endpoint_published = asyncio.Event()
527    endpoint_published.set()
528    await session._play(TRACK_A, 0, endpoint_published)
529    client.set_shuffle.assert_awaited_once_with(False)
530    client.set_repeat_context.assert_awaited_once_with(False)
531    client.set_repeat_track.assert_awaited_once_with(False)
532
533
534async def test_repeat_turned_on_from_the_app_is_pinned_back_off(tmp_path: Path) -> None:
535    """Repeat enabled in the Spotify app is undone before it can loop the item."""
536    session = _make_session(tmp_path)
537    await session._handle_event(
538        SoloistEvent(
539            type="options_changed",
540            data=SoloistOptionsChanged(
541                options=SoloistPlaybackOptions(shuffle=True, repeat="track")
542            ),
543            raw={},
544        )
545    )
546    client = _client_of(session)
547    client.set_shuffle.assert_awaited_once_with(False)
548    client.set_repeat_track.assert_awaited_once_with(False)
549    client.set_repeat_context.assert_awaited_once_with(False)
550    # options that are already off are left alone
551    client.set_shuffle.reset_mock()
552    await session._handle_event(
553        SoloistEvent(
554            type="options_changed",
555            data=SoloistOptionsChanged(options=SoloistPlaybackOptions()),
556            raw={},
557        )
558    )
559    client.set_shuffle.assert_not_awaited()
560
561
562async def test_a_busy_data_directory_is_reported_as_such(tmp_path: Path) -> None:
563    """A daemon left over from an earlier run is named, not reported as a generic failure."""
564    session = _make_session(tmp_path)
565    # the daemon's own parting complaint, which is all it gives (it exits with 1)
566    session._data_dir_busy = True
567    with pytest.raises(AudioError, match="Another Spotify Soloist session is still running"):
568        session._raise_startup_error("exited before playback started", TRACK_A)
569
570
571async def test_the_busy_marker_is_picked_up_from_the_daemon_output(tmp_path: Path) -> None:
572    """The marker is read off the daemon's stdout, with the API key still redacted."""
573    session = _make_session(tmp_path)
574    await session._log_output(
575        _stdout_of(
576            'Error: another session is running for data directory "/data/x/soloist-data".',
577            "Stop the running session before starting soloist again.",
578        )
579    )
580    assert session._data_dir_busy is True
581
582
583async def test_a_lost_pairing_is_caught_the_moment_the_daemon_reports_it(
584    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
585) -> None:
586    """A daemon advertising for pairing fails the session at once, not on a timeout."""
587    session = _make_session(tmp_path)
588    session._logged_in = None
589    unload_with_error = MagicMock()
590    monkeypatch.setattr(session.backend.provider, "unload_with_error", unload_with_error)
591    await session._log_output(
592        _stdout_of('waiting for login - connect to "X" from your Spotify app')
593    )
594    assert session._unpaired is True
595    # the buffer gives up on the audio long before the startup budget runs out, so
596    # the session has to fail while an item is still waiting on it
597    assert session._error is not None
598    with pytest.raises(LoginFailed) as err:
599        session._raise_startup_error("did not connect and log in", TRACK_A)
600    assert err.value.translation_key == "soloist_pairing_required"
601    unload_with_error.assert_called_once()
602
603
604async def test_a_lost_pairing_fails_the_item_without_waiting_for_the_endpoint(
605    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
606) -> None:
607    """The item fails on the lost pairing, not on the endpoint that is no longer coming."""
608    session = _make_session(tmp_path)
609    session._logged_in = None
610    monkeypatch.setattr(session.backend.provider, "unload_with_error", MagicMock())
611    await session._log_output(
612        _stdout_of('waiting for login - connect to "X" from your Spotify app')
613    )
614    # the endpoint never appears, so the wait for it must not swallow the failure:
615    # sitting it out would outlast the queue's own patience for the audio
616    with pytest.raises(LoginFailed) as err:
617        async with asyncio.timeout(5):
618            await session._play(TRACK_A, 0, asyncio.Event())
619    assert err.value.translation_key == "soloist_pairing_required"
620
621
622async def test_a_daemon_still_restoring_its_session_is_left_alone(tmp_path: Path) -> None:
623    """The engine advertises for pairing while restoring too; the stored session decides."""
624    session = _make_session(tmp_path)
625    data_dir = session.backend._data_dir
626    (data_dir / "settings" / "Users" / "spotify-user-user").mkdir(parents=True)
627    await session._log_output(
628        _stdout_of('waiting for login - connect to "X" from your Spotify app')
629    )
630    assert session._unpaired is False
631    assert session._error is None
632
633
634async def test_a_pairing_that_never_logs_in_routes_through_setup(
635    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
636) -> None:
637    """A session that cannot log in sends the user back to setup, not a per-track error."""
638    session = _make_session(tmp_path)
639    unload_with_error = MagicMock()
640    monkeypatch.setattr(session.backend.provider, "unload_with_error", unload_with_error)
641    await session._handle_event(_auth_event(logged_in=False))
642    with pytest.raises(LoginFailed) as err:
643        session._raise_startup_error("timed out waiting for playback to start", TRACK_A)
644    assert err.value.translation_key == "soloist_pairing_required"
645    # ... and the provider is taken out of service, so the user is asked to redo setup
646    unload_with_error.assert_called_once()
647
648
649async def test_a_login_that_never_happened_is_not_confused_with_another_failure(
650    tmp_path: Path,
651) -> None:
652    """An unrelated failure keeps its own message even before any login was reported."""
653    session = _make_session(tmp_path)
654    session._fail("the capture sink was lost mid-stream")
655    with pytest.raises(AudioError, match="capture sink was lost"):
656        session._raise_startup_error("exited before playback started", TRACK_A)
657
658
659def test_a_seeked_item_only_expects_what_is_left_of_it(tmp_path: Path) -> None:
660    """A seeked item delivers the remainder, so its targets are based on that."""
661    session = _make_session(tmp_path)
662    item = _ItemAudio(TRACK_A, session)
663    item.duration_ms = 200_000
664    full = 200_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
665    assert item._duration_bytes() == full
666    item.seek_target_ms = 150_000
667    remainder = 50_000 * CAPTURE_SAMPLE_RATE // 1000 * _FRAME_BYTES
668    assert item._duration_bytes() == remainder
669    # so the tail drain has a target it can actually reach
670    item.start_tail_drain()
671    item.write(b"\x01" * remainder)
672    assert item.tail_complete is True
673    # and the padding silence after it is refused
674    item.write(b"\x00" * 4096)
675    assert item.buffered == remainder
676
677
678@pytest.mark.parametrize(
679    ("target_ms", "reports"),
680    [
681        # seeking the restored item to where the engine already was
682        (117_000, (117_000, 0, 117_000)),
683        # seeking back into it, where every report lands below where it was
684        (30_000, (117_000, 0, 30_000)),
685    ],
686)
687async def test_the_seek_retries_until_the_engine_reports_the_target(
688    tmp_path: Path, target_ms: int, reports: tuple[int, ...]
689) -> None:
690    """A seek dropped while the track loads is re-sent until a report confirms it."""
691    session = _make_session(tmp_path)
692    item = _ItemAudio(TRACK_A, session)
693    client = cast("Any", session._client)
694    # the engine restored this item part-way in, before the seek goes out
695    item.observe_position(117_000)
696
697    async def _report_positions() -> None:
698        for position_ms in reports:
699            await asyncio.sleep(0)
700            item.observe_position(position_ms)
701
702    with (
703        patch.object(soloist_backend, "_SEEK_RETRY_INTERVAL_S", 0.01),
704        # bounded so a regression fails fast instead of sitting out the real budget
705        patch.object(soloist_backend, "_SEEK_CONFIRM_TIMEOUT_S", 1.0),
706    ):
707        reporter = asyncio.create_task(_report_positions())
708        await session._cold_seek(client, item, target_ms)
709        await reporter
710    assert item.seek_confirmed.is_set()
711    assert item.started_at_ms == target_ms
712    assert client.seek.await_count >= 1
713
714
715async def test_a_seek_that_only_ever_sees_the_restored_position_fails(tmp_path: Path) -> None:
716    """A seek nothing confirms fails loudly rather than streaming from elsewhere."""
717    session = _make_session(tmp_path)
718    item = _ItemAudio(TRACK_A, session)
719    # the engine sits at the restored position and never reloads the track
720    item.observe_position(117_000)
721    with (
722        patch.object(soloist_backend, "_SEEK_RETRY_INTERVAL_S", 0.01),
723        patch.object(soloist_backend, "_SEEK_CONFIRM_TIMEOUT_S", 0.05),
724        pytest.raises(AudioError, match="did not confirm seeking"),
725    ):
726        await session._cold_seek(cast("Any", session._client), item, 117_000)
727
728
729async def test_a_seek_the_engine_ignored_does_not_cut_the_item_short(tmp_path: Path) -> None:
730    """An item the engine plays from its start is bounded by its full duration."""
731    session = _make_session(tmp_path)
732    item = _ItemAudio(TRACK_A, session)
733    item.duration_ms = 176_000
734    item.arm_seek(117_000)
735    item.claim()
736    # the engine never made the seek and is playing the item from its start, so
737    # the audio it delivers runs well past what the seeked remainder would allow
738    item.observe_position(80_000)
739    item.write(b"\x01" * (89 * _BYTES_PER_SECOND))
740    item.close()
741    delivered = 0
742    async for chunk in item.read():
743        delivered += len(chunk)
744    assert delivered == 89 * _BYTES_PER_SECOND
745
746
747async def test_a_seek_that_landed_still_bounds_the_item_at_its_remainder(tmp_path: Path) -> None:
748    """An item the engine really seeked into stays bounded by what is left of it."""
749    session = _make_session(tmp_path)
750    item = _ItemAudio(TRACK_A, session)
751    item.duration_ms = 176_000
752    item.arm_seek(117_000)
753    item.observe_position(0)
754    item.observe_position(117_000)
755    assert item.seek_confirmed.is_set()
756    assert item._overrun_limit() == 59 * _BYTES_PER_SECOND + int(
757        _ITEM_OVERRUN_S * _BYTES_PER_SECOND
758    )
759    # later reports do not move the latch, which would shrink the bound
760    item.observe_position(150_000)
761    assert item.started_at_ms == 117_000
762    # so it still fails once it runs that far past the seek point
763    item.claim()
764    item.write(b"\x01" * (89 * _BYTES_PER_SECOND))
765    with pytest.raises(AudioError, match="never moved on"):
766        async for _ in item.read():
767            pass
768
769
770def test_the_lead_trim_never_exceeds_its_budget() -> None:
771    """Silence beyond the budget is content, including where audio starts mid-chunk."""
772    budget = int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND)
773    # already at the budget, with a chunk whose silence runs well past it
774    chunk = b"\x00" * 4096 + b"\x01" * 64
775    trimmed, skipped = _trim_lead_silence(chunk, budget - _FRAME_BYTES)
776    assert skipped == _FRAME_BYTES
777    assert len(trimmed) == len(chunk) - _FRAME_BYTES
778
779
780async def test_a_dying_log_reader_fails_the_session(tmp_path: Path) -> None:
781    """Nothing else drains the daemon's stdout, so a dead reader must not go unnoticed."""
782    session = _make_session(tmp_path)
783
784    async def _boom() -> None:
785        raise RuntimeError("reader blew up")
786
787    session._log_task = asyncio.create_task(_boom())
788    session._log_task.add_done_callback(session._task_done)
789    await asyncio.sleep(0)
790    await _wait_for(lambda: not session.usable)
791    assert session._error is not None
792    assert "reader blew up" in session._error
793
794
795async def test_feeding_never_replaces_a_channel_already_in_use(tmp_path: Path) -> None:
796    """If the engine reaches the fed item first, its live channel must survive."""
797    session = _make_session(tmp_path, queue_id="player1")
798    streamed = _streamed(session)
799    streamdetails = MagicMock()
800    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
801    queues = _queues_of(session)
802    queues.get.return_value = MagicMock(current_index=0)
803    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
804    queues.get_next_item.return_value = _queue_item(TRACK_B)
805
806    async def _engine_gets_there_first(_uri: str, **_kwargs: Any) -> None:
807        # the events task advances to the fed item while the command is in flight
808        await session._observe_current(TRACK_B, 200_000, track_changed=True)
809
810    _client_of(session).add_to_queue.side_effect = _engine_gets_there_first
811    await session.feed_after(streamdetails, streamed)
812    live = session.current
813    assert live is not None
814    assert live.uri == TRACK_B
815    # the channel the reader is writing to is the one a stream will be handed
816    assert [item for item in session._channels if item.uri == TRACK_B] == [live]
817    assert session.item_for(TRACK_B) is live
818    # and it is not queued as pending, because it already started
819    assert session.has_pending is False
820
821
822async def test_a_seek_the_session_cannot_take_restarts_it(
823    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
824) -> None:
825    """
826    A seek the running session cannot serve falls back to restarting it.
827
828    A realtime source has not captured anything past the play position, so any
829    forward seek lands outside the buffer and comes back here; the session is
830    seeked in place when it can be, and replaced when it cannot.
831    """
832    backend = _make_backend(tmp_path)
833    backend._server = MagicMock()
834    backend._binary = Path("/nonexistent/soloist")
835    session = _SoloistSession(backend, "player1")
836    backend._session = session
837    item = session._current = session._open_channel(TRACK_A)
838    item.started.set()
839    # its own stream is still attached when the seek re-opens it
840    item.claim()
841    stopped = AsyncMock()
842    monkeypatch.setattr(session, "stop", stopped)
843    _install_fake_binary_manager(monkeypatch)
844    monkeypatch.setattr(
845        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
846    )
847    with pytest.raises(AudioError, match="spawn"):
848        await backend._acquire(TRACK_A, 90, "player1")
849    stopped.assert_awaited_once()
850
851
852@pytest.mark.parametrize(
853    ("requested", "other_queue"),
854    [
855        # another player, whatever it asks for - including the very track this
856        # session is in the middle of delivering
857        pytest.param(TRACK_B, "player2", id="other_player"),
858        pytest.param(TRACK_A, "player2", id="other_player_same_track"),
859        # an early fetch across a boundary this session does not drive, such as a
860        # podcast episode or audiobook chapter
861        pytest.param(TRACK_B, "player1", id="unstitched_boundary"),
862    ],
863)
864async def test_a_session_in_use_is_never_cut_short(
865    tmp_path: Path, requested: str, other_queue: str
866) -> None:
867    """
868    An item the session cannot serve must not stop one it is still delivering.
869
870    Reported as capacity, so a speculative prepare gives up softly.
871    """
872    backend = _make_backend(tmp_path)
873    backend._server = MagicMock()
874    backend._binary = Path("/nonexistent/soloist")
875    session = _SoloistSession(backend, "player1")
876    backend._session = session
877    item = session._open_channel(TRACK_A)
878    item.started.set()
879    item.claim()
880    # the session really is playing TRACK_A, so a same-track request from another
881    # player cannot be mistaken for a seek
882    session._current = item
883    with pytest.raises(ProviderStreamLimitError) as err:
884        await backend._acquire(requested, 0, other_queue)
885    # a stream-limit error so the item is not marked unplayable, but the message
886    # is about the session, not the provider's source-stream budget
887    assert err.value.limit == 1
888    assert err.value.translation_key == "soloist_session_busy"
889    # the session that was playing is untouched
890    assert backend._session is session
891    assert session.usable is True
892
893
894async def test_a_session_nobody_reads_is_replaced_for_another_item(
895    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
896) -> None:
897    """Once the other item has been released, the same request gets the session."""
898    backend = _make_backend(tmp_path)
899    backend._server = MagicMock()
900    backend._binary = Path("/nonexistent/soloist")
901    session = _SoloistSession(backend, "player1")
902    backend._session = session
903    item = session._open_channel(TRACK_A)
904    item.started.set()
905    item.claim()
906    item.close()
907    item.release()
908    _install_fake_binary_manager(monkeypatch)
909    monkeypatch.setattr(
910        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
911    )
912    monkeypatch.setattr(session, "stop", AsyncMock())
913    with pytest.raises(AudioError, match="spawn"):
914        await backend._acquire(TRACK_B, 0, "player1")
915
916
917async def test_a_replacement_waits_for_the_old_daemon_to_be_gone(
918    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
919) -> None:
920    """The engine refuses to start while another daemon still holds its data dir."""
921    backend = _make_backend(tmp_path)
922    backend._server = MagicMock()
923    backend._binary = Path("/nonexistent/soloist")
924    session = _SoloistSession(backend, "player1")
925    backend._session = session
926    order: list[str] = []
927
928    async def _slow_stop() -> None:
929        order.append("stop-start")
930        await asyncio.sleep(0.05)
931        order.append("stop-done")
932
933    monkeypatch.setattr(session, "stop", _slow_stop)
934    _install_fake_binary_manager(monkeypatch)
935
936    async def _spawn(_self: Any, _uri: str, _seek: int) -> None:
937        order.append("spawn")
938        raise AudioError("spawn")
939
940    monkeypatch.setattr(soloist_backend._SoloistSession, "start", _spawn)
941    # the session failed, so its teardown is under way when the next item arrives
942    discard = asyncio.create_task(backend.discard_session(session))
943    await asyncio.sleep(0)
944    with pytest.raises(AudioError, match="spawn"):
945        await backend._acquire(TRACK_B, 0, "player1")
946    await discard
947    assert order == ["stop-start", "stop-done", "spawn"]
948
949
950async def test_an_idle_session_is_taken_over(
951    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
952) -> None:
953    """A session nobody is reading is replaced instead of blocking another player."""
954    backend = _make_backend(tmp_path)
955    backend._server = MagicMock()
956    backend._binary = Path("/nonexistent/soloist")
957    session = _SoloistSession(backend, "player1")
958    session._open_channel(TRACK_A)
959    backend._session = session
960    stopped = AsyncMock()
961    monkeypatch.setattr(session, "stop", stopped)
962    _install_fake_binary_manager(monkeypatch)
963    # the replacement spawn is out of scope here; only the takeover decision is
964    monkeypatch.setattr(
965        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
966    )
967    with pytest.raises(AudioError, match="spawn"):
968        await backend._acquire(TRACK_B, 0, "player2")
969    stopped.assert_awaited_once()
970
971
972def test_a_dead_session_task_fails_the_session(tmp_path: Path) -> None:
973    """A session task that dies of an unexpected error takes the session with it."""
974    session = _make_session(tmp_path)
975    task: Any = MagicMock()
976    task.cancelled.return_value = False
977    task.exception.return_value = RuntimeError("reader blew up")
978    session._task_done(task)
979    assert session.usable is False
980    assert session._error is not None
981    assert "reader blew up" in session._error
982
983
984def test_a_cancelled_session_task_is_not_a_failure(tmp_path: Path) -> None:
985    """Teardown cancels the session's tasks; that must not be reported as an error."""
986    session = _make_session(tmp_path)
987    task: Any = MagicMock()
988    task.cancelled.return_value = True
989    session._task_done(task)
990    assert session.usable is True
991
992
993async def test_failed_sink_control_fails_the_session(tmp_path: Path) -> None:
994    """A failed suspend/resume fails the session instead of leaking stall silence."""
995    session = _make_session(tmp_path)
996    session._demand_started = True
997    session._sink_running = True
998    session._current = session._open_channel(TRACK_A)
999    _feed(session, TRACK_B)
1000    _sink_of(session).suspend.side_effect = RuntimeError("pactl failed")
1001    await session._handle_event(_playback_event("buffering"))
1002    assert session._error is not None
1003    assert "capture sink control failed" in session._error
1004
1005
1006async def test_app_pause_is_fought_with_a_resume(tmp_path: Path) -> None:
1007    """A pause from the Spotify app is undone: this session has no user-facing pause."""
1008    session = _make_session(tmp_path)
1009    session._demand_started = True
1010    session._current = session._open_channel(TRACK_A)
1011    _feed(session, TRACK_B)
1012    await session._handle_event(_playback_event("paused"))
1013    _client_of(session).resume.assert_awaited_once()
1014
1015
1016async def test_an_app_pause_is_only_undone_so_many_times(tmp_path: Path) -> None:
1017    """Someone who keeps pausing means it: the session gives up instead of fighting on."""
1018    session = _make_session(tmp_path)
1019    session._demand_started = True
1020    session._current = session._open_channel(TRACK_A)
1021    _feed(session, TRACK_B)
1022    for _ in range(_MAX_APP_PAUSE_RESUMES):
1023        await session._handle_event(_playback_event("playing"))
1024        await session._handle_event(_playback_event("paused"))
1025    assert _client_of(session).resume.await_count == _MAX_APP_PAUSE_RESUMES
1026    assert session._error is None
1027
1028    await session._handle_event(_playback_event("playing"))
1029    await session._handle_event(_playback_event("paused"))
1030    assert _client_of(session).resume.await_count == _MAX_APP_PAUSE_RESUMES
1031    assert session.usable is False
1032    assert session._app_control is SoloistAppControl.PAUSED
1033
1034
1035async def test_one_pause_reported_twice_counts_once(tmp_path: Path) -> None:
1036    """A repeated snapshot of the same pause is not a new pause."""
1037    session = _make_session(tmp_path)
1038    session._demand_started = True
1039    session._current = session._open_channel(TRACK_A)
1040    _feed(session, TRACK_B)
1041    for _ in range(_MAX_APP_PAUSE_RESUMES + 2):
1042        await session._handle_event(_playback_event("paused"))
1043    assert session.usable is True
1044
1045
1046async def test_the_pause_budget_resets_on_the_next_item(tmp_path: Path) -> None:
1047    """Each item gets its own budget; pausing one track does not spend the next one's."""
1048    session = _make_session(tmp_path)
1049    session._demand_started = True
1050    session._current = session._open_channel(TRACK_A)
1051    _feed(session, TRACK_B)
1052    for _ in range(_MAX_APP_PAUSE_RESUMES):
1053        await session._handle_event(_playback_event("playing"))
1054        await session._handle_event(_playback_event("paused"))
1055    await session._observe_current(TRACK_B, 200_000, track_changed=True)
1056    assert session._app_pauses == 0
1057
1058
1059async def test_a_pause_is_not_undone_once_the_device_is_gone(tmp_path: Path) -> None:
1060    """A bare resume on a device Spotify no longer routes to would play to nobody."""
1061    session = _make_session(tmp_path)
1062    session._demand_started = True
1063    session._was_active = False
1064    session._current = session._open_channel(TRACK_A)
1065    _feed(session, TRACK_B)
1066    await session._handle_event(_playback_event("paused"))
1067    _client_of(session).resume.assert_not_awaited()
1068
1069
1070async def test_losing_the_active_device_ends_the_session(tmp_path: Path) -> None:
1071    """Playback moved to another device from the Spotify app: this session is over."""
1072    session = _make_session(tmp_path)
1073    await session._handle_event(_device_event(is_active=False))
1074    assert session.usable is False
1075    assert session._app_control is SoloistAppControl.TOOK_OVER
1076
1077
1078async def test_a_takeover_reported_on_the_auth_state_ends_the_session(tmp_path: Path) -> None:
1079    """The active-device state also rides on auth_state, and counts the same there."""
1080    session = _make_session(tmp_path)
1081    await session._handle_event(_auth_event(logged_in=True, is_active=False))
1082    assert session.usable is False
1083
1084
1085async def test_an_inactive_device_before_activation_is_not_a_takeover(tmp_path: Path) -> None:
1086    """A fresh daemon is inactive until the session claims it; that is not a takeover."""
1087    session = _make_session(tmp_path)
1088    session._was_active = False
1089    await session._handle_event(_device_event(is_active=False))
1090    await session._handle_event(_auth_event(logged_in=True, is_active=False))
1091    assert session.usable is True
1092
1093    # nor does a respawned daemon reporting the session Spotify still has for
1094    # the account: only the status _play claimed is followed
1095    await session._handle_event(_device_event(is_active=True))
1096    await session._handle_event(_device_event(is_active=False))
1097    assert session.usable is True
1098
1099
1100async def test_a_reconnect_snapshot_keeps_an_active_session_alive(tmp_path: Path) -> None:
1101    """The events connection re-snapshots after a drop; that is not a device change."""
1102    session = _make_session(tmp_path)
1103    await session._handle_event(_auth_event(logged_in=True, is_active=True))
1104    await session._handle_event(_device_event(is_active=True))
1105    assert session.usable is True
1106
1107
1108async def test_the_playback_snapshots_active_flag_is_ignored(tmp_path: Path) -> None:
1109    """It is optional and rides on deltas, so only the dedicated reports are followed."""
1110    session = _make_session(tmp_path)
1111    session._demand_started = True
1112    session._current = session._open_channel(TRACK_A)
1113    await session._handle_event(
1114        SoloistEvent(
1115            type="playback_changed",
1116            data=SoloistPlaybackState(status="playing", is_active=False),
1117            raw={},
1118        )
1119    )
1120    assert session.usable is True
1121
1122
1123async def test_backpressure_does_not_spend_the_pause_budget(tmp_path: Path) -> None:
1124    """A sink suspended to cap the cushion is our doing, not the user pausing."""
1125    session = _make_session(tmp_path)
1126    session._demand_started = True
1127    session._engine_playing = True
1128    session._backpressured = True
1129    session._current = session._open_channel(TRACK_A)
1130    _feed(session, TRACK_B)
1131    await session._handle_event(_playback_event("paused"))
1132    _client_of(session).resume.assert_not_awaited()
1133    assert session._app_pauses == 0
1134
1135
1136async def test_a_lost_login_is_not_reported_as_a_takeover(tmp_path: Path) -> None:
1137    """Losing the login wins over the inactive device it brings with it."""
1138    session = _make_session(tmp_path)
1139    await session._handle_event(_auth_event(logged_in=False, is_active=False))
1140    assert session.usable is False
1141    assert session._app_control is None
1142
1143
1144async def test_a_track_started_from_the_app_ends_the_session(tmp_path: Path) -> None:
1145    """The engine pulled off an item part-way through is the app playing something else."""
1146    session = _make_session(tmp_path)
1147    item = session._open_channel(TRACK_A)
1148    item.duration_ms = 200_000
1149    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1150    item.observe_position(20_000)
1151
1152    await session._observe_current("spotify:track:theirs", 180_000, track_changed=True)
1153    assert session.usable is False
1154    assert session._app_control is SoloistAppControl.TOOK_OVER
1155    assert session.current is item
1156
1157
1158async def test_a_track_played_earlier_started_from_the_app_ends_the_session(
1159    tmp_path: Path,
1160) -> None:
1161    """A known uri is no exemption: only the item fed behind this one is where we sent it."""
1162    session = _make_session(tmp_path)
1163    played = session._open_channel(TRACK_B)
1164    played.spent = True
1165    item = session._open_channel(TRACK_A)
1166    item.duration_ms = 200_000
1167    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1168    item.observe_position(20_000)
1169
1170    await session._observe_current(TRACK_B, 180_000, track_changed=True)
1171    assert session.usable is False
1172    assert session._app_control is SoloistAppControl.TOOK_OVER
1173
1174
1175async def test_skipping_from_the_app_to_the_fed_item_is_followed(tmp_path: Path) -> None:
1176    """The queue moves to that same track, so following the engine keeps the two in step."""
1177    session = _make_session(tmp_path)
1178    item = session._open_channel(TRACK_A)
1179    item.duration_ms = 200_000
1180    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1181    item.observe_position(20_000)
1182    fed = _feed(session, TRACK_B)
1183
1184    await session._observe_current(TRACK_B, 180_000, track_changed=True)
1185    assert session.usable is True
1186    assert session.current is fed
1187    assert session.item_for(TRACK_B) is fed
1188
1189
1190async def test_a_skip_leaves_the_outgoing_stream_nothing_to_report(tmp_path: Path) -> None:
1191    """A jump Music Assistant asked for cuts the outgoing item, which is not starving."""
1192    session = _make_session(tmp_path)
1193    item = session._open_channel(TRACK_A)
1194    item.duration_ms = 200_000
1195    item.playing_seen = True
1196    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1197    item.observe_position(20_000)
1198    fed = _feed(session, TRACK_B)
1199    # what skip_to arms before telling the engine to jump
1200    session._discard_until = fed
1201
1202    await session._observe_current(TRACK_B, 180_000, track_changed=True)
1203    assert session.current is fed
1204    assert item.superseded
1205    await session.validate_item(item)
1206
1207
1208async def test_a_boundary_the_engine_drove_is_still_judged(tmp_path: Path) -> None:
1209    """Nobody asked the engine to leave this item, so what it delivered still counts."""
1210    session = _make_session(tmp_path)
1211    item = session._open_channel(TRACK_A)
1212    item.duration_ms = 200_000
1213    item.playing_seen = True
1214    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1215    item.observe_position(20_000)
1216    _feed(session, TRACK_B)
1217
1218    await session._observe_current(TRACK_B, 180_000, track_changed=True)
1219    assert item.superseded is False
1220    with pytest.raises(AudioError, match="incomplete"):
1221        await session.validate_item(item)
1222
1223
1224async def test_a_takeover_snapshot_stops_pinning_volume_and_options(tmp_path: Path) -> None:
1225    """Once the app has the session, the rest of its snapshot must not reach the daemon."""
1226    session = _make_session(tmp_path)
1227    session._demand_started = True
1228    item = session._open_channel(TRACK_A)
1229    item.duration_ms = 200_000
1230    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1231    item.observe_position(20_000)
1232
1233    await session._handle_event(
1234        SoloistEvent(
1235            type="playback_changed",
1236            data=SoloistPlaybackState(
1237                status="playing",
1238                item=SoloistEntity(uri="spotify:track:theirs", entity_type="track"),
1239                volume=40,
1240                options=SoloistPlaybackOptions(shuffle=True, repeat="context"),
1241            ),
1242            raw={},
1243        )
1244    )
1245    assert session.usable is False
1246    _client_of(session).set_volume.assert_not_awaited()
1247    _client_of(session).set_shuffle.assert_not_awaited()
1248
1249
1250async def test_the_engine_moving_on_at_a_track_end_is_not_a_takeover(tmp_path: Path) -> None:
1251    """An unasked-for item the engine reaches at a boundary is its own autoplay."""
1252    session = _make_session(tmp_path)
1253    item = session._open_channel(TRACK_A)
1254    item.duration_ms = 200_000
1255    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1256    item.observe_position(200_000)
1257
1258    await session._observe_current("spotify:track:autoplay", 180_000, track_changed=True)
1259    assert session.usable is True
1260    assert session.item_for("spotify:track:autoplay") is None
1261
1262
1263async def test_an_ended_item_says_what_the_app_did(tmp_path: Path) -> None:
1264    """The item's stream fails with the takeover, not a generic session error."""
1265    session = _make_session(tmp_path)
1266    item = session._open_channel(TRACK_A)
1267    await session._handle_event(_device_event(is_active=False))
1268    with pytest.raises(SoloistAppControlError) as err:
1269        await session.validate_item(item)
1270    assert err.value.translation_key == SoloistAppControl.TOOK_OVER.value
1271    assert isinstance(err.value, ProviderStreamLimitError)
1272
1273
1274async def test_a_session_being_torn_down_does_not_hold_off_the_next_one(tmp_path: Path) -> None:
1275    """Teardown pauses the daemon; that must not read as the user pausing."""
1276    session = _make_session(tmp_path)
1277    session._demand_started = True
1278    session._current = session._open_channel(TRACK_A)
1279    _feed(session, TRACK_B)
1280    session._stopped = True
1281    for _ in range(_MAX_APP_PAUSE_RESUMES + 1):
1282        await session._handle_event(_playback_event("playing"))
1283        await session._handle_event(_playback_event("paused"))
1284    await session._handle_event(_device_event(is_active=False))
1285    session.backend._raise_if_app_controlled()
1286    _client_of(session).resume.assert_not_awaited()
1287
1288
1289async def test_no_session_is_started_while_the_app_holds_the_last_one(tmp_path: Path) -> None:
1290    """A replacement would claim the Connect device straight back off the user."""
1291    backend = _make_backend(tmp_path)
1292    backend._note_app_control(SoloistAppControl.TOOK_OVER)
1293    with pytest.raises(SoloistAppControlError):
1294        await backend._acquire(TRACK_A, 0, "player1")
1295
1296
1297async def test_the_hold_on_a_new_session_expires(tmp_path: Path) -> None:
1298    """Coming back to Music Assistant later plays again without any fuss."""
1299    backend = _make_backend(tmp_path)
1300    backend._note_app_control(SoloistAppControl.TOOK_OVER)
1301    backend._app_control_until = time.monotonic() - 1
1302    backend._raise_if_app_controlled()
1303    assert backend._held_by_app() is None
1304
1305
1306async def test_an_audiobook_gives_up_on_capacity_instead_of_burning_chapters(
1307    tmp_path: Path,
1308) -> None:
1309    """Skipping ahead would cost the audiobook its availability and the caller its retry."""
1310    provider = _make_provider(tmp_path)
1311    calls: list[str] = []
1312
1313    async def _refuse(uri: str, *_args: Any, **_kwargs: Any) -> AsyncGenerator[bytes]:
1314        calls.append(uri)
1315        for _ in ():  # never yields; only makes this an async generator
1316            yield b""
1317        raise SoloistAppControlError(provider, SoloistAppControl.TOOK_OVER)
1318
1319    provider.backend = MagicMock(stream_spotify_uri=_refuse)
1320    streamdetails = MagicMock(
1321        media_type=MediaType.AUDIOBOOK,
1322        data={"chapters": [TRACK_A, TRACK_B, "spotify:track:ccc"], "chapters_data": []},
1323    )
1324
1325    with pytest.raises(SoloistAppControlError):
1326        async for _ in provider.get_audio_stream(streamdetails):
1327            pass
1328    # the first chapter's refusal ends it: no chapter is skipped over
1329    assert calls == [TRACK_A]
1330
1331
1332def test_the_playback_device_is_named_apart_from_the_connect_one() -> None:
1333    """Two identically named devices in the Spotify app is what causes the takeovers."""
1334    assert SOLOIST_DEVICE_NAME != DEFAULT_PUBLISH_NAME
1335
1336
1337async def test_app_volume_change_is_pinned_back_to_unity(tmp_path: Path) -> None:
1338    """An off-unity volume set from the Spotify app is pinned back to 100."""
1339    session = _make_session(tmp_path)
1340    await session._handle_event(
1341        SoloistEvent(type="volume_changed", data=SoloistVolumeChanged(volume=40), raw={})
1342    )
1343    _client_of(session).set_volume.assert_awaited_once_with(100)
1344    _client_of(session).set_volume.reset_mock()
1345    await session._handle_event(
1346        SoloistEvent(type="volume_changed", data=SoloistVolumeChanged(volume=100), raw={})
1347    )
1348    _client_of(session).set_volume.assert_not_awaited()
1349
1350
1351async def test_track_change_signals_the_queue_when_it_matches_the_next_item(
1352    tmp_path: Path,
1353) -> None:
1354    """Reaching a fed item tells the queue to start filling that item's buffer."""
1355    session = _make_session(tmp_path, queue_id="player1")
1356    session._current = session._open_channel(TRACK_A)
1357    session._open_channel(TRACK_B)
1358    queues = _queues_of(session)
1359    queues.get.return_value = MagicMock(next_item=_queue_item(TRACK_B), current_index=0)
1360    await session._handle_event(
1361        SoloistEvent(
1362            type="track_changed",
1363            data=SoloistTrackChanged(item=SoloistEntity(uri=TRACK_B, entity_type="track")),
1364            raw={},
1365        )
1366    )
1367    queues.prepare_next_audio_buffer.assert_called_once_with("player1")
1368
1369
1370async def test_track_change_to_another_item_signals_nothing(tmp_path: Path) -> None:
1371    """An item the queue is not asking for next must not trigger a prebuffer."""
1372    session = _make_session(tmp_path, queue_id="player1")
1373    session._current = session._open_channel(TRACK_A)
1374    queues = _queues_of(session)
1375    queues.get.return_value = MagicMock(next_item=_queue_item(TRACK_B), current_index=0)
1376    await session._handle_event(
1377        SoloistEvent(
1378            type="track_changed",
1379            data=SoloistTrackChanged(
1380                item=SoloistEntity(uri="spotify:track:surprise", entity_type="track")
1381            ),
1382            raw={},
1383        )
1384    )
1385    queues.prepare_next_audio_buffer.assert_not_called()
1386
1387
1388async def test_the_follower_of_the_streamed_item_is_fed(tmp_path: Path) -> None:
1389    """The item after the one being streamed is handed to the engine."""
1390    session = _make_session(tmp_path, queue_id="player1")
1391    streamed = _streamed(session)
1392    streamdetails = MagicMock()
1393    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1394    follower = _queue_item(TRACK_B)
1395    queues = _queues_of(session)
1396    queues.get.return_value = MagicMock(current_index=3)
1397    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 3 else None
1398    queues.get_next_item.return_value = follower
1399    await session.feed_after(streamdetails, streamed)
1400    _client_of(session).add_to_queue.assert_awaited_once_with(TRACK_B)
1401    assert session.pending_item(TRACK_B) is not None
1402    assert session.has_pending is True
1403
1404
1405async def test_repeating_one_track_does_not_feed_the_engine(tmp_path: Path) -> None:
1406    """Repeat-one replays the item from the buffer the queue holds, so nothing is queued."""
1407    session = _make_session(tmp_path, queue_id="player1")
1408    streamed = _streamed(session)
1409    streamdetails = MagicMock(provider="spotify--test")
1410    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1411    queues = _queues_of(session)
1412    queues.get.return_value = MagicMock(current_index=0)
1413    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1414    # repeat-one names the item being streamed as its own follower
1415    queues.get_next_item.return_value = playing
1416    assert await session.feed_after(streamdetails, streamed) is False
1417    _client_of(session).add_to_queue.assert_not_awaited()
1418
1419
1420async def test_an_item_the_queue_resolved_elsewhere_is_not_fed(tmp_path: Path) -> None:
1421    """A track the queue will stream from another provider must not be queued here."""
1422    session = _make_session(tmp_path, queue_id="player1")
1423    streamed = _streamed(session)
1424    streamdetails = MagicMock()
1425    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1426    # same track, but the queue already picked a different provider for it
1427    follower = _queue_item(TRACK_B, streamdetails=MagicMock(provider="tidal--x"))
1428    queues = _queues_of(session)
1429    queues.get.return_value = MagicMock(current_index=0)
1430    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1431    queues.get_next_item.return_value = follower
1432    await session.feed_after(streamdetails, streamed)
1433    _client_of(session).add_to_queue.assert_not_awaited()
1434
1435
1436async def test_skipping_to_the_fed_item_keeps_the_session(tmp_path: Path) -> None:
1437    """A next-track lands on the item already fed, so the engine jumps instead of respawning."""
1438    backend = _make_backend(tmp_path)
1439    backend._server = MagicMock()
1440    backend._binary = Path("/nonexistent/soloist")
1441    session = _SoloistSession(backend, "player1")
1442    session._client = AsyncMock()
1443    session._logged_in = True
1444    backend._session = session
1445    playing = session._current = session._open_channel(TRACK_A)
1446    playing.started.set()
1447    # fed one ahead and not reached yet, which is where a next-track goes
1448    fed = _feed(session, TRACK_B)
1449
1450    async def _engine_gets_there(**_kwargs: Any) -> None:
1451        await session._observe_current(TRACK_B, 200_000, track_changed=True)
1452
1453    _client_of(session).skip_next.side_effect = _engine_gets_there
1454    got_session, got_item = await backend._acquire(TRACK_B, 0, "player1")
1455    # the same session, no respawn, and the item that was already queued
1456    assert got_session is session
1457    assert got_item is fed
1458    assert backend._session is session
1459    _client_of(session).skip_next.assert_awaited_once()
1460
1461
1462async def test_a_repeated_track_keeps_the_session(tmp_path: Path) -> None:
1463    """The second occurrence of a track is served by the session that played the first."""
1464    backend = _make_backend(tmp_path)
1465    backend._server = MagicMock()
1466    backend._binary = Path("/nonexistent/soloist")
1467    session = _SoloistSession(backend, "player1")
1468    session._client = AsyncMock()
1469    backend._session = session
1470    # the first occurrence has been delivered and the second was fed behind it
1471    first = _streamed(session)
1472    first.release()
1473    second = _feed(session, TRACK_A)
1474
1475    async def _engine_gets_there(**_kwargs: Any) -> None:
1476        await session._observe_current(TRACK_A, 200_000, track_changed=True)
1477
1478    _client_of(session).skip_next.side_effect = _engine_gets_there
1479    got_session, got_item = await backend._acquire(TRACK_A, 0, "player1")
1480    assert got_session is session
1481    assert got_item is second
1482    assert backend._session is session
1483
1484
1485async def test_a_next_item_the_session_was_not_fed_is_queued_and_skipped_to(
1486    tmp_path: Path,
1487) -> None:
1488    """A queue reordered after the feed is served by sending the engine on, not by respawning."""
1489    backend = _make_backend(tmp_path)
1490    backend._server = MagicMock()
1491    backend._binary = Path("/nonexistent/soloist")
1492    session = _SoloistSession(backend, "player1")
1493    session._client = AsyncMock()
1494    session._engine_playing = True
1495    backend._session = session
1496    # the engine moved on into the item it was fed, which the queue no longer wants
1497    stale = session._current = session._open_channel(TRACK_B)
1498    stale.started.set()
1499
1500    async def _engine_gets_there(**_kwargs: Any) -> None:
1501        await session._observe_current(TRACK_C, 200_000, track_changed=True)
1502
1503    _client_of(session).skip_next.side_effect = _engine_gets_there
1504    got_session, got_item = await backend._acquire(TRACK_C, 0, "player1")
1505    assert got_session is session
1506    assert got_item.uri == TRACK_C
1507    assert got_item.claimed is True
1508    _client_of(session).add_to_queue.assert_awaited_once_with(TRACK_C)
1509    _client_of(session).skip_next.assert_awaited_once()
1510    assert stale._closed is True
1511
1512
1513async def test_the_item_the_engine_is_on_is_never_jumped_to(tmp_path: Path) -> None:
1514    """A jump steps past the item, so the engine is never sent to what it already plays."""
1515    session = _make_session(tmp_path)
1516    session._engine_playing = True
1517    _streamed(session, TRACK_A).release()
1518    assert await session.feed_and_skip_to(TRACK_A) is None
1519    _client_of(session).add_to_queue.assert_not_awaited()
1520    _client_of(session).skip_next.assert_not_awaited()
1521
1522
1523async def test_a_session_delivering_an_item_is_never_sent_to_another(tmp_path: Path) -> None:
1524    """A jump would cut short the item being delivered, so capacity is reported instead."""
1525    backend = _make_backend(tmp_path)
1526    backend._server = MagicMock()
1527    backend._binary = Path("/nonexistent/soloist")
1528    session = _SoloistSession(backend, "player1")
1529    session._client = AsyncMock()
1530    session._engine_playing = True
1531    backend._session = session
1532    # an item the engine is on and a stream is reading
1533    _streamed(session, TRACK_B)
1534    with pytest.raises(ProviderStreamLimitError):
1535        await backend._acquire(TRACK_C, 0, "player1")
1536    _client_of(session).add_to_queue.assert_not_awaited()
1537    _client_of(session).skip_next.assert_not_awaited()
1538    assert backend._session is session
1539    assert session.usable is True
1540
1541
1542@pytest.mark.parametrize(
1543    "blocker",
1544    [
1545        # one skip steps one entry, so anything queued behind would be landed on
1546        "something_queued",
1547        # a stopped engine has nothing to skip out of
1548        "engine_stopped",
1549        # the engine would not take the jump
1550        "refused",
1551    ],
1552)
1553async def test_a_session_that_cannot_be_sent_on_is_replaced(
1554    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, blocker: str
1555) -> None:
1556    """Where the engine's transport cannot reach the item, a fresh session serves it."""
1557    backend = _make_backend(tmp_path)
1558    backend._server = MagicMock()
1559    backend._binary = Path("/nonexistent/soloist")
1560    session = _SoloistSession(backend, "player1")
1561    session._client = AsyncMock()
1562    session._engine_playing = blocker != "engine_stopped"
1563    backend._session = session
1564    # the engine is on an item whose audio has been handed over already
1565    _streamed(session, TRACK_B).release()
1566    if blocker == "something_queued":
1567        _feed(session, TRACK_A)
1568    if blocker == "refused":
1569        _client_of(session).skip_next.side_effect = SoloistError("no")
1570    _install_fake_binary_manager(monkeypatch)
1571    monkeypatch.setattr(
1572        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
1573    )
1574    monkeypatch.setattr(session, "stop", AsyncMock())
1575    with pytest.raises(AudioError, match="spawn"):
1576        await backend._acquire(TRACK_C, 0, "player1")
1577    if blocker != "refused":
1578        _client_of(session).add_to_queue.assert_not_awaited()
1579
1580
1581async def test_a_jump_to_the_fed_item_that_misses_is_replaced(
1582    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1583) -> None:
1584    """A jump the engine will not take costs a respawn, not the item."""
1585    backend = _make_backend(tmp_path)
1586    backend._server = MagicMock()
1587    backend._binary = Path("/nonexistent/soloist")
1588    session = _SoloistSession(backend, "player1")
1589    session._client = AsyncMock()
1590    backend._session = session
1591    _streamed(session).release()
1592    _feed(session, TRACK_B)
1593    _client_of(session).skip_next.side_effect = SoloistError("no")
1594    _install_fake_binary_manager(monkeypatch)
1595    monkeypatch.setattr(
1596        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
1597    )
1598    monkeypatch.setattr(session, "stop", AsyncMock())
1599    with pytest.raises(AudioError, match="spawn"):
1600        await backend._acquire(TRACK_B, 0, "player1")
1601
1602
1603async def test_a_skip_drops_what_arrives_while_the_command_is_in_flight(
1604    tmp_path: Path,
1605) -> None:
1606    """
1607    Audio captured between the skip command and the engine's answer is dropped.
1608
1609    Only covers the marker's own window; what the pipeline still holds when the
1610    answer arrives is measured at the cut instead.
1611    """
1612    session = _make_session(tmp_path)
1613    leaving = session._current = session._open_channel(TRACK_A)
1614    leaving.started.set()
1615    target = _feed(session, TRACK_B)
1616    captured: list[bytes] = []
1617
1618    async def _engine_gets_there(**_kwargs: Any) -> None:
1619        # the pipeline still holds the old track while the command is in flight
1620        session._write_if_wanted(b"\x01" * 32)
1621        await session._observe_current(TRACK_B, 200_000, track_changed=True)
1622        # from here on the audio really is the new item's
1623        session._write_if_wanted(b"\x02" * 32)
1624
1625    _client_of(session).skip_next.side_effect = _engine_gets_there
1626    await session.skip_to(target)
1627    captured.extend(target._chunks)
1628    assert b"".join(captured) == b"\x02" * 32
1629    assert session._discard_until is None
1630
1631
1632async def test_a_skip_the_engine_never_reaches_fails(
1633    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1634) -> None:
1635    """A skip that does not land is an error, not a wait for the track to end."""
1636    monkeypatch.setattr(soloist_backend, "_JUMP_TIMEOUT_S", 0.05)
1637    session = _make_session(tmp_path)
1638    fed = _feed(session, TRACK_B)
1639    with pytest.raises(AudioError, match="did not reach"):
1640        await session.skip_to(fed)
1641
1642
1643def test_a_jump_gives_up_while_the_queue_is_still_waiting() -> None:
1644    """A jump that will not land has to fail in time for a fresh session to serve the item."""
1645    assert _JUMP_TIMEOUT_S < BUFFER_READY_TIMEOUT
1646
1647
1648async def test_a_fed_item_the_engine_has_not_reached_is_not_served(tmp_path: Path) -> None:
1649    """Skipping to an already-fed item must not hand over a channel that fills later."""
1650    session = _make_session(tmp_path)
1651    # the engine is still on the track before it
1652    _streamed(session, TRACK_A)
1653    fed = _feed(session, TRACK_B)
1654    assert session.item_for(TRACK_B) is None
1655    # once the engine gets there it is servable
1656    await session._observe_current(TRACK_B, 200_000, track_changed=True)
1657    assert session.item_for(TRACK_B) is fed
1658
1659
1660def test_the_shaper_only_emits_whole_frames() -> None:
1661    """A read that ends mid-frame must never split a frame across two items."""
1662    shaper = soloist_backend._CaptureShaper()
1663    # the session's first bytes are infrastructure silence, and are dropped
1664    assert shaper.shape(b"\x00" * 4096) == b""
1665    # a mis-aligned read emits whole frames and carries the remainder
1666    first = shaper.shape(b"\x01" * (_FRAME_BYTES + 3))
1667    assert len(first) == _FRAME_BYTES
1668    # which is then completed by the next read, losing nothing
1669    second = shaper.shape(b"\x02" * (_FRAME_BYTES - 3))
1670    assert len(second) == _FRAME_BYTES
1671    assert second[:3] == b"\x01" * 3
1672    # an aligned read passes straight through
1673    assert shaper.shape(b"\x03" * _FRAME_BYTES) == b"\x03" * _FRAME_BYTES
1674
1675
1676def test_the_shaper_trims_lead_silence_only_once() -> None:
1677    """Silence after the audio has started is content, not pre-roll."""
1678    shaper = soloist_backend._CaptureShaper()
1679    assert shaper.shape(b"\x01" * _FRAME_BYTES) == b"\x01" * _FRAME_BYTES
1680    silence = b"\x00" * _FRAME_BYTES
1681    assert shaper.shape(silence) == silence
1682
1683
1684async def test_only_whole_sample_frames_are_handed_over(
1685    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1686) -> None:
1687    """A read that ends mid-frame must not split a frame across two items."""
1688    session = _make_session(tmp_path)
1689    item = session._current = session._open_channel(TRACK_A)
1690    item.started.set()
1691    item.claim()
1692    session._demand_started = True
1693    session._sink_running = True
1694    # two reads that are each mis-aligned but whole together
1695    reads = [b"\x01" * (_FRAME_BYTES + 3), b"\x02" * (_FRAME_BYTES - 3), b""]
1696    reader = MagicMock()
1697
1698    async def _read(_size: int) -> bytes:
1699        return reads.pop(0) if reads else b""
1700
1701    reader.read = _read
1702    session._reader = reader
1703    monkeypatch.setattr(soloist_backend, "_PACE_RATE", 1000.0)
1704    await session._read_capture()
1705    # every write was frame-aligned, and no byte was lost
1706    assert item.buffered % _FRAME_BYTES == 0
1707    assert item.buffered == _FRAME_BYTES * 2
1708
1709
1710@pytest.mark.parametrize("state", ["fed", "reached", "being_read"])
1711async def test_an_already_known_item_is_not_fed_twice(tmp_path: Path, state: str) -> None:
1712    """An item the session was fed, or has already moved on to, is not queued again."""
1713    session = _make_session(tmp_path, queue_id="player1")
1714    streamed = _streamed(session)
1715    known = _feed(session, TRACK_B)
1716    if state != "fed":
1717        # the engine got there while the stream is still reading the item before it
1718        await session._observe_current(TRACK_B, 200_000, track_changed=True)
1719        assert session.current is known
1720    if state == "being_read":
1721        # and its own stream opened, which spends the channel
1722        known.claim()
1723    streamdetails = MagicMock()
1724    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1725    queues = _queues_of(session)
1726    queues.get.return_value = MagicMock(current_index=0)
1727    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1728    queues.get_next_item.return_value = _queue_item(TRACK_B)
1729    assert await session.feed_after(streamdetails, streamed) is True
1730    _client_of(session).add_to_queue.assert_not_awaited()
1731
1732
1733async def test_occurrences_of_one_track_are_served_in_the_order_they_were_fed(
1734    tmp_path: Path,
1735) -> None:
1736    """A track queued three times in a row hands each occurrence its own channel."""
1737    session = _make_session(tmp_path)
1738    first = _streamed(session)
1739    second = _feed(session, TRACK_A)
1740    third = _feed(session, TRACK_A)
1741    # the engine has not moved yet, so the next occurrence is the one fed first
1742    assert session.pending_item(TRACK_A) is second
1743    first.duration_ms = 200_000
1744    first.observe_position(199_000)
1745    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1746    assert session.current is second
1747    assert session.pending_item(TRACK_A) is third
1748    second.claim()
1749    second.duration_ms = 200_000
1750    second.observe_position(199_000)
1751    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1752    assert session.current is third
1753    assert session.pending_item(TRACK_A) is None
1754
1755
1756async def test_a_track_played_earlier_is_not_answered_with_its_old_channel(
1757    tmp_path: Path,
1758) -> None:
1759    """A track that comes round again is the occurrence fed for it, not the one played."""
1760    session = _make_session(tmp_path)
1761    played = _feed(session, TRACK_B)
1762    await session._observe_current(TRACK_B, 200_000, track_changed=True)
1763    # the engine moves on, so that channel is over
1764    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1765    # ... and the track comes round again later in the queue
1766    again = _feed(session, TRACK_B)
1767    await session._observe_current(TRACK_B, 200_000, track_changed=True)
1768    assert session.current is again
1769    assert played.closed is True
1770
1771
1772async def test_a_channel_nothing_can_read_is_not_kept(tmp_path: Path) -> None:
1773    """A channel the session moved past, with no stream on it, stops counting against the cap."""
1774    session = _make_session(tmp_path)
1775    passed_by = _feed(session, TRACK_B)
1776    await session._observe_current(TRACK_B, 200_000, track_changed=True)
1777    session._write_if_wanted(b"\x01" * 4096)
1778    assert session._retained_bytes() == 4096
1779    # the engine moves on again and no stream ever opened this one
1780    await session._observe_current(TRACK_C, 200_000, track_changed=True)
1781    assert passed_by.closed is True
1782    session._open_channel(TRACK_A)
1783    assert passed_by not in session._channels
1784    assert session._retained_bytes() == 0
1785
1786
1787async def test_a_channel_a_stream_still_holds_is_never_dropped(tmp_path: Path) -> None:
1788    """A stream still draining an item past the cut keeps the session in use."""
1789    session = _make_session(tmp_path)
1790    reading = _streamed(session)
1791    # the engine moves on while that stream is still reading the item
1792    await session._observe_current(TRACK_B, 200_000, track_changed=True)
1793    assert reading.closed is True
1794    assert reading.claimed is True
1795    session._open_channel(TRACK_C)
1796    assert reading in session._channels
1797    assert session.in_use is True
1798
1799
1800async def test_the_channel_the_engine_is_on_is_always_kept(tmp_path: Path) -> None:
1801    """The last item of a run is closed by its own drain, but the session is still on it."""
1802    session = _make_session(tmp_path)
1803    last = _streamed(session)
1804    last.release()
1805    last.close()
1806    session._open_channel(TRACK_B)
1807    assert session.current is last
1808    assert last in session._channels
1809
1810
1811async def test_a_cancelled_jump_ends_the_session_without_blaming_the_spotify_app(
1812    tmp_path: Path,
1813) -> None:
1814    """A jump nobody is waiting for any more ends the session, but is not a takeover."""
1815    session = _make_session(tmp_path)
1816    session._engine_playing = True
1817    playing = _streamed(session, TRACK_A)
1818    playing.release()
1819    # part-way through, so an arrival nobody asked for would read as a takeover
1820    playing.duration_ms = 200_000
1821    playing.observe_position(20_000)
1822
1823    async def _gives_up(**_kwargs: Any) -> None:
1824        raise asyncio.CancelledError
1825
1826    _client_of(session).skip_next.side_effect = _gives_up
1827    with pytest.raises(asyncio.CancelledError):
1828        await session.feed_and_skip_to(TRACK_B)
1829    # the jump cannot be accounted for any more, so the session ends
1830    assert session.usable is False
1831    # ... but the engine still getting there is not the app taking over, which
1832    # would hold off every session that follows
1833    await session._observe_current(TRACK_B, 200_000, track_changed=True)
1834    assert session._app_control is None
1835    assert session.backend._held_by_app() is None
1836
1837
1838async def test_a_drained_channel_is_not_served(tmp_path: Path) -> None:
1839    """The last item of a run ends its channel when its audio is done; it serves nothing after."""
1840    session = _make_session(tmp_path)
1841    item = session._current = session._open_channel(TRACK_A)
1842    item.started.set()
1843    assert session.item_for(TRACK_A) is item
1844    # nothing follows it, so the session drains the item and closes it
1845    item.close()
1846    assert session.item_for(TRACK_A) is None
1847
1848
1849async def test_a_channel_the_session_moved_past_is_not_served(tmp_path: Path) -> None:
1850    """A channel the session left behind holds only part of its item, so it is never handed out."""
1851    session = _make_session(tmp_path)
1852    played_past = _feed(session, TRACK_B)
1853    await session._observe_current(TRACK_B, 200_000, track_changed=True)
1854    # servable while the engine is on it and no stream has taken it
1855    assert session.item_for(TRACK_B) is played_past
1856    # the engine moves on again before any stream opened it
1857    await session._observe_current(TRACK_C, 200_000, track_changed=True)
1858    assert played_past.closed is True
1859    assert session.item_for(TRACK_B) is None
1860
1861
1862async def test_a_repeated_track_is_fed_a_channel_of_its_own(tmp_path: Path) -> None:
1863    """A track that follows itself is queued again, not answered with the channel in use."""
1864    session = _make_session(tmp_path, queue_id="player1")
1865    streamed = _streamed(session)
1866    streamdetails = MagicMock()
1867    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
1868    queues = _queues_of(session)
1869    queues.get.return_value = MagicMock(current_index=0)
1870    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
1871    # the very same track once more, as a queue item of its own
1872    queues.get_next_item.return_value = _queue_item(TRACK_A, queue_item_id="qi-again")
1873    assert await session.feed_after(streamdetails, streamed) is True
1874    _client_of(session).add_to_queue.assert_awaited_once_with(TRACK_A)
1875    second = session.pending_item(TRACK_A)
1876    assert second is not None
1877    assert second is not streamed
1878
1879
1880async def test_a_repeated_track_moves_on_at_its_track_change(tmp_path: Path) -> None:
1881    """The boundary between two occurrences of one track is reported under the same uri."""
1882    session = _make_session(tmp_path)
1883    first = _streamed(session)
1884    first.duration_ms = 200_000
1885    first.observe_position(199_000)
1886    second = _feed(session, TRACK_A)
1887    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1888    assert session.current is second
1889    assert first._closed is True
1890
1891
1892async def test_a_state_report_does_not_move_a_repeated_track_on(tmp_path: Path) -> None:
1893    """Only a track change crosses that boundary; a state report says where the engine is."""
1894    session = _make_session(tmp_path)
1895    first = _streamed(session)
1896    first.duration_ms = 200_000
1897    first.observe_position(199_000)
1898    second = _feed(session, TRACK_A)
1899    await session._observe_current(TRACK_A, 200_000, track_changed=False)
1900    assert session.current is first
1901    assert session.pending_item(TRACK_A) is second
1902
1903
1904async def test_the_track_change_event_moves_a_repeated_track_on(tmp_path: Path) -> None:
1905    """The track_changed event is the one that carries a repeat across its boundary."""
1906    session = _make_session(tmp_path)
1907    first = _streamed(session)
1908    first.duration_ms = 200_000
1909    first.observe_position(199_000)
1910    second = _feed(session, TRACK_A)
1911    await session._handle_event(_current_item_event("track_changed", TRACK_A, 200_000))
1912    assert session.current is second
1913    assert first.closed is True
1914
1915
1916@pytest.mark.parametrize("event_type", ["playback_state", "playback_changed"])
1917async def test_a_state_event_does_not_move_a_repeated_track_on(
1918    tmp_path: Path, event_type: str
1919) -> None:
1920    """A snapshot near the end of the first occurrence describes it, it does not end it."""
1921    session = _make_session(tmp_path)
1922    first = _streamed(session)
1923    first.duration_ms = 200_000
1924    first.observe_position(199_000)
1925    second = _feed(session, TRACK_A)
1926    await session._handle_event(_current_item_event(event_type, TRACK_A, 200_000))
1927    assert session.current is first
1928    assert first.closed is False
1929    assert session.pending_item(TRACK_A) is second
1930
1931
1932@pytest.mark.parametrize("event_type", ["track_changed", "playback_state", "playback_changed"])
1933async def test_an_event_naming_another_item_cuts_at_the_boundary(
1934    tmp_path: Path, event_type: str
1935) -> None:
1936    """Whichever event reports the move, the item being left ends and the next one takes over."""
1937    session = _make_session(tmp_path)
1938    first = _streamed(session)
1939    first.duration_ms = 200_000
1940    first.observe_position(20_000)
1941    second = _feed(session, TRACK_B)
1942    await session._handle_event(_current_item_event(event_type, TRACK_B, 180_000))
1943    # the engine left the previous item part-way through, but for one it was fed:
1944    # the queue moving on, not the Spotify app taking the session over
1945    assert session.usable is True
1946    assert session.current is second
1947    assert second.duration_ms == 180_000
1948    assert first.closed is True
1949
1950
1951async def test_a_position_report_tells_the_current_item_where_the_engine_is(
1952    tmp_path: Path,
1953) -> None:
1954    """Where the engine got to is what tells an item played out from one it was pulled off."""
1955    session = _make_session(tmp_path)
1956    item = _streamed(session)
1957    item.duration_ms = 200_000
1958    assert item.mid_play is False  # no position reported yet, so nothing to judge by
1959    await session._handle_event(_position_event(20_000))
1960    assert item.last_position_ms == 20_000
1961    assert item.mid_play is True
1962
1963
1964async def test_a_seek_in_flight_is_not_cut_short_by_a_repeat_boundary(tmp_path: Path) -> None:
1965    """A channel opened for a seek has no position of its own, which is not a played-out one."""
1966    session = _make_session(tmp_path)
1967    seeking = _streamed(session)
1968    seeking.duration_ms = 200_000
1969    _feed(session, TRACK_A)
1970    session._seeking = True
1971    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1972    assert session.current is seeking
1973    assert seeking.closed is False
1974
1975
1976async def test_a_repeat_is_not_moved_on_to_part_way_through_the_first(tmp_path: Path) -> None:
1977    """A track change reported part-way through the first occurrence is not its boundary."""
1978    session = _make_session(tmp_path)
1979    first = _streamed(session)
1980    first.duration_ms = 200_000
1981    first.observe_position(20_000)
1982    _feed(session, TRACK_A)
1983    await session._observe_current(TRACK_A, 200_000, track_changed=True)
1984    assert session.current is first
1985
1986
1987async def test_a_jump_to_a_repeat_is_followed_part_way_through(tmp_path: Path) -> None:
1988    """Skipping ahead to the second occurrence lands there even mid-track."""
1989    session = _make_session(tmp_path)
1990    first = _streamed(session)
1991    first.duration_ms = 200_000
1992    first.observe_position(20_000)
1993    second = _feed(session, TRACK_A)
1994
1995    async def _engine_gets_there(**_kwargs: Any) -> None:
1996        await session._observe_current(TRACK_A, 200_000, track_changed=True)
1997
1998    _client_of(session).skip_next.side_effect = _engine_gets_there
1999    await session.skip_to(second)
2000    assert session.current is second
2001    assert first._closed is True
2002
2003
2004async def test_only_tracks_are_fed_ahead(tmp_path: Path) -> None:
2005    """A podcast episode or audiobook chapter is played on its own, never stitched."""
2006    session = _make_session(tmp_path, queue_id="player1")
2007    await session.feed_after(MagicMock(), session._open_channel("spotify:episode:xyz"))
2008    _client_of(session).add_to_queue.assert_not_awaited()
2009
2010
2011async def test_a_non_spotify_follower_is_not_fed(tmp_path: Path) -> None:
2012    """The run simply ends where the queue leaves this provider."""
2013    session = _make_session(tmp_path, queue_id="player1")
2014    streamed = _streamed(session)
2015    streamdetails = MagicMock()
2016    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
2017    follower = MagicMock(
2018        media_item=MagicMock(media_type=MediaType.TRACK, provider="tidal--x"), streamdetails=None
2019    )
2020    follower.media_item.provider_mappings = []
2021    queues = _queues_of(session)
2022    queues.get.return_value = MagicMock(current_index=0)
2023    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
2024    queues.get_next_item.return_value = follower
2025    await session.feed_after(streamdetails, streamed)
2026    _client_of(session).add_to_queue.assert_not_awaited()
2027
2028
2029async def test_a_library_item_is_fed_through_its_spotify_mapping(tmp_path: Path) -> None:
2030    """A library track is fed with the item id this provider instance knows it by."""
2031    session = _make_session(tmp_path, queue_id="player1")
2032    streamed = _streamed(session)
2033    streamdetails = MagicMock()
2034    playing = _queue_item(TRACK_A, streamdetails=streamdetails)
2035    follower = MagicMock(
2036        media_item=MagicMock(media_type=MediaType.TRACK, provider="library", item_id="42"),
2037        streamdetails=None,
2038    )
2039    follower.media_item.provider_mappings = [
2040        MagicMock(provider_instance="other--y", item_id="wrong"),
2041        MagicMock(provider_instance="spotify--test", item_id="bbb"),
2042    ]
2043    queues = _queues_of(session)
2044    queues.get.return_value = MagicMock(current_index=0)
2045    queues.get_item.side_effect = lambda _queue_id, index: playing if index == 0 else None
2046    queues.get_next_item.return_value = follower
2047    await session.feed_after(streamdetails, streamed)
2048    _client_of(session).add_to_queue.assert_awaited_once_with(TRACK_B)
2049
2050
2051@pytest.mark.parametrize(
2052    ("provider_option", "player_setting", "expected"),
2053    [
2054        (True, "enabled", True),
2055        # the player's own switch decides first: off means nobody normalizes,
2056        # not that the job passes to Spotify
2057        (True, "disabled", False),
2058        (False, "enabled", False),
2059        (False, "disabled", False),
2060    ],
2061)
2062def test_who_normalizes_needs_both_switches(
2063    tmp_path: Path,
2064    monkeypatch: pytest.MonkeyPatch,
2065    provider_option: bool,
2066    player_setting: str,
2067    expected: bool,
2068) -> None:
2069    """The engine normalizes only when the provider option and the player agree."""
2070    session = _make_session(tmp_path, queue_id="player1")
2071    monkeypatch.setattr(
2072        type(session.backend.provider),
2073        "spotify_normalization_configured",
2074        property(lambda _self: provider_option),
2075    )
2076    cast("MagicMock", session.mass.config).get_effective_player_queue_config_value = MagicMock(
2077        return_value=player_setting
2078    )
2079    assert session._engine_normalization_enabled() is expected
2080
2081
2082def test_a_running_session_answers_for_what_the_engine_is_doing(tmp_path: Path) -> None:
2083    """
2084    The engine reads its settings at startup, so a later toggle must not split them.
2085
2086    Otherwise the streams core would start normalizing on top of audio the engine
2087    is still normalizing, or stop while it no longer is.
2088    """
2089    backend = _make_backend(tmp_path)
2090    provider = backend.provider
2091    streamdetails = _streamdetails_for(queue_id="player1")
2092    # nothing playing yet: the configuration is all there is to go on
2093    before_any_session = backend.session_normalizes(streamdetails)
2094    session = _SoloistSession(backend, "player1")
2095    session.engine_normalizes = True
2096    backend._session = session
2097    while_playing = backend.session_normalizes(streamdetails)
2098    # ... and a session that has been torn down no longer speaks for the engine
2099    session._stopped = True
2100    after_teardown = backend.session_normalizes(streamdetails)
2101    assert before_any_session is None
2102    assert while_playing is True
2103    assert after_teardown is None
2104    assert (
2105        provider.delivers_normalized_audio(streamdetails)
2106        is provider.spotify_normalization_configured
2107    )
2108
2109
2110async def test_short_delivery_is_rejected_as_incomplete(tmp_path: Path) -> None:
2111    """PCM that stops well short of the item's duration is rejected."""
2112    session = _make_session(tmp_path)
2113    item = _ItemAudio(TRACK_A, session)
2114    item.playing_seen = True
2115    item.duration_ms = 200_000
2116    item.last_position_ms = 100_000
2117    with pytest.raises(AudioError, match="incomplete"):
2118        await session.validate_item(item)
2119
2120
2121async def test_missing_position_is_rejected_as_incomplete(tmp_path: Path) -> None:
2122    """Without any position report there is no evidence the item played out."""
2123    session = _make_session(tmp_path)
2124    item = _ItemAudio(TRACK_A, session)
2125    item.playing_seen = True
2126    item.duration_ms = 200_000
2127    with pytest.raises(AudioError, match="incomplete"):
2128        await session.validate_item(item)
2129
2130
2131async def test_short_item_cannot_pass_at_position_zero(tmp_path: Path) -> None:
2132    """The tolerance never spans a whole item, so a short item cannot pass unplayed."""
2133    session = _make_session(tmp_path)
2134    item = _ItemAudio(TRACK_A, session)
2135    item.playing_seen = True
2136    item.duration_ms = 8_000
2137    item.last_position_ms = 0
2138    with pytest.raises(AudioError, match="incomplete"):
2139        await session.validate_item(item)
2140
2141
2142async def test_an_item_that_never_played_is_rejected(tmp_path: Path) -> None:
2143    """An item the engine never reported playing is a failure, whatever was delivered."""
2144    session = _make_session(tmp_path)
2145    item = _ItemAudio(TRACK_A, session)
2146    item.duration_ms = 200_000
2147    item.last_position_ms = 200_000
2148    with pytest.raises(AudioError, match="never started playing"):
2149        await session.validate_item(item)
2150
2151
2152async def test_a_duration_less_item_is_not_judged(tmp_path: Path) -> None:
2153    """Without a duration there is nothing to judge completeness against."""
2154    session = _make_session(tmp_path)
2155    item = _ItemAudio(TRACK_A, session)
2156    item.playing_seen = True
2157    await session.validate_item(item)
2158
2159
2160async def test_a_superseded_item_is_not_judged_incomplete(tmp_path: Path) -> None:
2161    """A channel cut part-way is short on purpose, so it is no evidence of starving."""
2162    session = _make_session(tmp_path)
2163    item = _ItemAudio(TRACK_A, session)
2164    item.playing_seen = True
2165    item.duration_ms = 200_000
2166    item.last_position_ms = 30_000
2167    item.close(superseded=True)
2168    await session.validate_item(item)
2169
2170
2171async def test_a_superseded_item_that_never_played_is_still_rejected(tmp_path: Path) -> None:
2172    """Cutting a channel excuses a short delivery, not one that carried nothing."""
2173    session = _make_session(tmp_path)
2174    item = _ItemAudio(TRACK_A, session)
2175    item.duration_ms = 200_000
2176    item.close(superseded=True)
2177    with pytest.raises(AudioError, match="never started playing"):
2178        await session.validate_item(item)
2179
2180
2181async def test_an_item_the_engine_moved_on_from_keeps_its_verdict(tmp_path: Path) -> None:
2182    """A later teardown must not excuse a channel the engine already starved."""
2183    session = _make_session(tmp_path)
2184    item = session._open_channel(TRACK_A)
2185    item.playing_seen = True
2186    item.duration_ms = 200_000
2187    item.last_position_ms = 30_000
2188    # the boundary the engine drove, with the teardown following behind it
2189    item.close()
2190    await session.stop()
2191    with pytest.raises(AudioError, match="incomplete"):
2192        await session.validate_item(item)
2193
2194
2195def test_an_unread_session_expires(tmp_path: Path) -> None:
2196    """A session no item stream reads from is ended so its daemon does not linger."""
2197    session = _make_session(tmp_path)
2198    session._open_channel(TRACK_A)
2199    session._expire_idle()
2200    assert session._idle_since is not None
2201    assert session.usable is True
2202    session._idle_since = time.monotonic() - _IDLE_TIMEOUT_S - 1
2203    session._expire_idle()
2204    assert session.usable is False
2205
2206
2207def test_a_session_being_read_never_expires(tmp_path: Path) -> None:
2208    """An item stream reading the session keeps it alive indefinitely."""
2209    session = _make_session(tmp_path)
2210    item = session._open_channel(TRACK_A)
2211    item.claim()
2212    session._idle_since = time.monotonic() - _IDLE_TIMEOUT_S * 10
2213    session._expire_idle()
2214    assert session.usable is True
2215
2216
2217def test_pre_roll_silence_is_dropped_a_whole_frame_at_a_time() -> None:
2218    """
2219    Trimming pre-roll must leave the audio on the session's frame grid.
2220
2221    A FIFO read is not always a whole number of frames, and dropping a partial
2222    one would shift every sample that follows for the rest of the session.
2223    """
2224    shaper = _CaptureShaper()
2225    # pre-roll that ends mid-frame: the real audio starts at byte 1024
2226    assert shaper.shape(b"\x00" * 1021) == b""
2227    audio = bytes(range(1, 9)) * 4
2228    shaped = shaper.shape(b"\x00" * 3 + audio)
2229    assert shaped == audio
2230    assert shaper._lead_skipped % _FRAME_BYTES == 0
2231
2232
2233async def test_a_refused_skip_does_not_leave_the_audio_discarded(tmp_path: Path) -> None:
2234    """
2235    A skip that never landed must not keep the session dropping its audio.
2236
2237    The marker silences everything the session captures, so a command that
2238    failed has to clear it on the way out.
2239    """
2240    session = _make_session(tmp_path)
2241    client = cast("MagicMock", session._client)
2242    client.skip_next = AsyncMock(side_effect=TimeoutError)
2243    item = _ItemAudio(TRACK_B, session)
2244
2245    with pytest.raises(AudioError, match="would not skip"):
2246        await session.skip_to(item)
2247
2248    assert session._discard_until is None
2249
2250
2251async def test_a_daemon_that_will_not_die_is_reported_and_released(tmp_path: Path) -> None:
2252    """A close that could not terminate the daemon still finishes the teardown."""
2253    session = _make_session(tmp_path)
2254    proc = cast("MagicMock", session._proc)
2255    proc.close = AsyncMock()
2256    # AsyncProcess.close() gives up after a handful of kill attempts
2257    proc.returncode = None
2258    with patch.object(session.logger, "warning") as warning:
2259        await session.stop()
2260    assert warning.called
2261    assert session._teardown_done is True
2262    assert session._proc is None
2263
2264
2265async def test_a_cancelled_teardown_still_closes_the_daemon(tmp_path: Path) -> None:
2266    """
2267    A cancelled teardown must leave the retry something to close.
2268
2269    Dropping the references first is how a daemon survives to hold the data
2270    directory, which every later session is then refused for.
2271    """
2272    session = _make_session(tmp_path)
2273    proc = cast("MagicMock", session._proc)
2274    sink = cast("AsyncMock", session._sink)
2275
2276    async def _never_returns() -> None:
2277        await asyncio.Event().wait()
2278
2279    proc.close = _never_returns
2280    task = asyncio.create_task(session.stop())
2281    await asyncio.sleep(0.01)
2282    task.cancel()
2283    with suppress(asyncio.CancelledError):
2284        await task
2285    # the teardown did not finish, so nothing was dropped and it can be redone
2286    unfinished = session._teardown_done
2287    kept_proc = session._proc
2288    kept_sink = session._sink
2289    proc.close = AsyncMock()
2290    proc.returncode = 0
2291    await session.stop()
2292    assert unfinished is False
2293    assert kept_proc is proc
2294    assert kept_sink is sink
2295    assert session._teardown_done is True
2296    assert session._proc is None
2297    assert session._sink is None
2298    proc.close.assert_awaited()
2299    sink.unload.assert_awaited()
2300
2301
2302async def test_a_teardown_leaves_the_running_stream_nothing_to_report(tmp_path: Path) -> None:
2303    """Stopping the session cuts the item being played; that is not a starved item."""
2304    session = _make_session(tmp_path)
2305    item = session._current = session._open_channel(TRACK_A)
2306    item.started.set()
2307    item.duration_ms = 260_000
2308    item.playing_seen = True
2309    item.observe_position(30_000)
2310    await session.stop()
2311
2312    assert item.superseded
2313    await session.validate_item(item)
2314
2315
2316def test_a_failed_session_is_torn_down(tmp_path: Path) -> None:
2317    """A session that fails is discarded, so its daemon does not keep playing to nobody."""
2318    session = _make_session(tmp_path)
2319    item = session._open_channel(TRACK_A)
2320    item.claim()
2321    session._fail("audio stalled")
2322    assert session.usable is False
2323    # every waiting item is released and the teardown is scheduled
2324    assert item._closed is True
2325    # a startup wait must not sit out its timeout on a session that already failed
2326    assert item.started.is_set() is True
2327    discard = cast("MagicMock", session.mass.create_task)
2328    discard.assert_called_once_with(session.backend.discard_session, session)
2329    # a second failure does not queue a second teardown
2330    session._fail("and again")
2331    assert session._error == "audio stalled"
2332    assert discard.call_count == 1
2333
2334
2335async def test_an_item_the_engine_skipped_past_fails_instead_of_hanging(
2336    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
2337) -> None:
2338    """A claimed channel the engine never reaches gives up rather than blocking forever."""
2339    monkeypatch.setattr(soloist_backend, "_READ_SLICE_S", 0.01)
2340    monkeypatch.setattr(soloist_backend, "_STALL_TIMEOUT_S", 0.05)
2341    session = _make_session(tmp_path)
2342    item = session._open_channel(TRACK_A)
2343    item.claim()
2344    # the engine is playing something else, so nothing is ever written here
2345    session._current = session._open_channel("spotify:track:other")
2346    with pytest.raises(AudioError, match="no audio"):
2347        async for _ in item.read():
2348            pass
2349
2350
2351async def test_adopt_paired_session_copies_into_the_canonical_dir(
2352    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
2353) -> None:
2354    """A session paired by the setup flow is adopted into the per-instance data dir."""
2355    storage = tmp_path / "storage"
2356    pending = storage / "spotify" / "pairing" / "flow1"
2357    pending.mkdir(parents=True)
2358    (pending / "session.bin").write_bytes(b"session")
2359    prov = _make_provider(tmp_path, {CONF_SOLOIST_SESSION_DIR: "spotify/pairing/flow1"})
2360    update_setup_data = MagicMock()
2361    monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
2362    backend = SoloistBackend(prov)
2363    await backend._adopt_paired_session()
2364    canonical = storage / "spotify" / "spotify--test" / SOLOIST_DATA_DIR_NAME
2365    assert (canonical / "session.bin").read_bytes() == b"session"
2366    # a copy, not a move: the flow-private source must survive a failed
2367    # provider load so the setup flow can retry (the flow removes it at its end)
2368    assert (pending / "session.bin").exists()
2369    update_setup_data.assert_called_once_with(CONF_SOLOIST_SESSION_DIR, None)
2370
2371
2372def test_the_engine_is_told_not_to_normalize(tmp_path: Path) -> None:
2373    """MA normalizes this audio itself, so the engine's own normalization is switched off."""
2374    backend = _make_backend(tmp_path)
2375    prefs = backend._data_dir / "settings" / "Users" / "alice-user" / "prefs"
2376    prefs.parent.mkdir(parents=True)
2377    prefs.write_text("some.engine.key=1\n", encoding="utf-8")
2378    backend._prepare_data_dir(normalize=False)
2379    content = prefs.read_text(encoding="utf-8").splitlines()
2380    assert "some.engine.key=1" in content
2381    assert "audio.normalize_v2=false" in content
2382    # MA mixes the queue's crossfade itself, so the engine's own is always off
2383    assert "audio.crossfade_v2=false" in content
2384    # the ceiling is stated rather than left to the engine's own default
2385    assert "audio.play_bitrate_enumeration=5" in content
2386    assert "audio.play_bitrate_non_metered_enumeration=5" in content
2387    assert "audio.play_bitrate_non_metered_migrated=true" in content
2388
2389
2390def test_disabling_crossfade_writes_the_boolean(tmp_path: Path) -> None:
2391    """Crossfade off is written explicitly, so a stale 'on' cannot survive."""
2392    backend = _make_backend(tmp_path)
2393    prefs = backend._data_dir / "settings" / "prefs"
2394    prefs.parent.mkdir(parents=True)
2395    prefs.write_text("audio.crossfade_v2=true\naudio.crossfade.time_v2=8000\n", encoding="utf-8")
2396    backend._prepare_data_dir(normalize=False)
2397    content = prefs.read_text(encoding="utf-8").splitlines()
2398    assert "audio.crossfade_v2=false" in content
2399    assert not any(line.startswith("audio.crossfade.time_v2") for line in content)
2400
2401
2402async def test_setup_requires_an_api_key(tmp_path: Path) -> None:
2403    """Without a stored API key the user must be sent back through the setup flow."""
2404    backend = _make_backend(tmp_path)
2405    with pytest.raises(LoginFailed) as err:
2406        await backend.setup()
2407    assert err.value.translation_key == "soloist_pairing_required"
2408
2409
2410async def test_setup_requires_a_paired_session(
2411    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
2412) -> None:
2413    """An API key without a paired session also routes back to the setup flow."""
2414    backend = _make_backend(tmp_path, {CONF_SOLOIST_API_KEY: "k" * 20, CONF_SOLOIST_CONSENT: True})
2415    _install_fake_binary_manager(monkeypatch)
2416    with pytest.raises(LoginFailed) as err:
2417        await backend.setup()
2418    assert err.value.translation_key == "soloist_pairing_required"
2419
2420
2421async def test_streaming_without_setup_is_refused(tmp_path: Path) -> None:
2422    """A backend whose setup never ran refuses to stream instead of half-starting."""
2423    backend = _make_backend(tmp_path)
2424    with pytest.raises(AudioError, match="not started"):
2425        async for _ in backend.stream_spotify_uri(TRACK_A):
2426            pass
2427
2428
2429def test_session_present_detection(tmp_path: Path) -> None:
2430    """Only the engine's per-account state counts as paired."""
2431    data_dir = tmp_path / "soloist-data"
2432    assert soloist_session_present(data_dir) is False
2433    data_dir.mkdir()
2434    (data_dir / WS_ADDR_FILE).write_text("127.0.0.1", encoding="utf-8")
2435    (data_dir / WS_PORT_FILE).write_text("1234", encoding="utf-8")
2436    assert soloist_session_present(data_dir) is False
2437    # everything a spawn leaves behind outlives the pairing it ran on: the engine
2438    # keeps its identity, lock, cache and crash handler in the data dir even
2439    # though it is given a cache dir of its own, and Music Assistant writes the
2440    # prefs there before every spawn
2441    (data_dir / "settings").mkdir()
2442    (data_dir / "settings" / "prefs").write_text("audio.normalize_v2=false\n", encoding="utf-8")
2443    (data_dir / ".device_id").write_text("6b6c2a07", encoding="utf-8")
2444    (data_dir / ".lock").write_bytes(b"")
2445    (data_dir / "cache" / "Users" / "spotify-user-user").mkdir(parents=True)
2446    (data_dir / "crashpad").mkdir()
2447    assert soloist_session_present(data_dir) is False
2448    (data_dir / "settings" / "Users" / "spotify-user-user").mkdir(parents=True)
2449    assert soloist_session_present(data_dir) is True
2450
2451
2452async def test_a_skip_drops_the_audio_still_in_flight(tmp_path: Path) -> None:
2453    """The item jumped to opens with its own audio, not the tail of the one left behind."""
2454    session = _make_session(tmp_path)
2455    left_behind = session._open_channel(TRACK_A)
2456    session._current = left_behind
2457    left_behind.started.set()
2458    left_behind.claim()
2459    jumped_to = _feed(session, TRACK_B)
2460    session._discard_until = jumped_to
2461    with _capture_holding(session, fifo_bytes=2 * _FRAME_BYTES, reader_bytes=2 * _FRAME_BYTES):
2462        await session._observe_current(TRACK_B, 200_000, track_changed=True)
2463    assert session._stale_budget == 4 * _FRAME_BYTES
2464    session._write_if_wanted(b"s" * (4 * _FRAME_BYTES))
2465    session._write_if_wanted(b"n" * (2 * _FRAME_BYTES))
2466    jumped_to.claim()
2467    jumped_to.close()
2468    assert b"".join([chunk async for chunk in jumped_to.read()]) == b"n" * (2 * _FRAME_BYTES)
2469
2470
2471async def test_a_skip_drops_the_stale_audio_across_reads(tmp_path: Path) -> None:
2472    """A budget larger than one read keeps dropping, and resumes on a frame boundary."""
2473    session = _make_session(tmp_path)
2474    session._stale_budget = 3 * _FRAME_BYTES
2475    item = session._current = session._open_channel(TRACK_A)
2476    item.claim()
2477    session._write_if_wanted(b"s" * (2 * _FRAME_BYTES))
2478    session._write_if_wanted(b"s" * _FRAME_BYTES + b"n" * _FRAME_BYTES)
2479    item.close()
2480    assert b"".join([chunk async for chunk in item.read()]) == b"n" * _FRAME_BYTES
2481
2482
2483async def test_the_marker_spends_an_earlier_jumps_budget(tmp_path: Path) -> None:
2484    """What the marker drops still counts against a budget left from an earlier jump."""
2485    session = _make_session(tmp_path)
2486    session._current = session._open_channel(TRACK_A)
2487    session._stale_budget = 4 * _FRAME_BYTES
2488    session._discard_until = _feed(session, TRACK_B)
2489    session._write_if_wanted(b"s" * (3 * _FRAME_BYTES))
2490    assert session._stale_budget == _FRAME_BYTES
2491    # a refused command leaves only what is genuinely still in flight to drop
2492    session._discard_until = None
2493    session._write_if_wanted(b"s" * _FRAME_BYTES + b"n" * _FRAME_BYTES)
2494    item = session._current
2495    item.claim()
2496    item.close()
2497    assert b"".join([chunk async for chunk in item.read()]) == b"n" * _FRAME_BYTES
2498
2499
2500async def test_a_natural_cut_keeps_the_audio_in_flight(tmp_path: Path) -> None:
2501    """Nothing is dropped without a jump: what is in flight is the continuation."""
2502    session = _make_session(tmp_path)
2503    playing = session._open_channel(TRACK_A)
2504    session._current = playing
2505    playing.started.set()
2506    playing.claim()
2507    with _capture_holding(session, fifo_bytes=4 * _FRAME_BYTES, reader_bytes=4 * _FRAME_BYTES):
2508        await session._observe_current(TRACK_B, 200_000, track_changed=True)
2509    assert session._stale_budget == 0
2510
2511
2512def test_stale_bytes_spans_both_buffers_in_whole_frames(tmp_path: Path) -> None:
2513    """The in-flight measure covers the FIFO and the reader, and never splits a frame."""
2514    session = _make_session(tmp_path)
2515    with _capture_holding(session, fifo_bytes=3 * _FRAME_BYTES + 3, reader_bytes=2 * _FRAME_BYTES):
2516        assert session._stale_bytes() == 5 * _FRAME_BYTES
2517
2518
2519def test_stale_bytes_falls_back_when_the_reader_cannot_be_sized(tmp_path: Path) -> None:
2520    """Losing the reader's internal view drops extra rather than leaving audio behind."""
2521    session = _make_session(tmp_path)
2522    with _capture_holding(session, fifo_bytes=0, reader_bytes=None):
2523        assert session._stale_bytes() == 6 * _READ_CHUNK_SIZE
2524
2525
2526async def test_a_channel_abandoned_at_the_cut_stops_holding_the_cushion(
2527    tmp_path: Path,
2528) -> None:
2529    """A skip closes the channel first and only then unwinds its stream."""
2530    session = _make_session(tmp_path)
2531    item = session._current = session._open_channel(TRACK_A)
2532    item.started.set()
2533    item.claim()
2534    item.write(b"x" * 4096)
2535    # the cut lands while the abandoned stream is still unwinding
2536    await session._observe_current(TRACK_B, 200_000, track_changed=True)
2537    assert session._retained_bytes() == 4096
2538    item.release()
2539    assert session._retained_bytes() == 0
2540
2541
2542def test_an_abandoned_channel_stops_holding_the_cushion(tmp_path: Path) -> None:
2543    """A channel skipped away from frees its buffer instead of gating the sink for good."""
2544    session = _make_session(tmp_path)
2545    item = session._open_channel(TRACK_A)
2546    item.claim()
2547    item.write(b"x" * 4096)
2548    assert session._retained_bytes() == 4096
2549    # the stream is gone, then the cut closes the channel
2550    item.release()
2551    item.close()
2552    assert session._retained_bytes() == 0
2553
2554
2555async def test_a_channel_no_stream_ever_took_stops_holding_the_cushion(
2556    tmp_path: Path,
2557) -> None:
2558    """A channel the session cuts with nothing reading it frees its buffer right away."""
2559    session = _make_session(tmp_path)
2560    item = _feed(session, TRACK_B)
2561    await session._observe_current(TRACK_B, 200_000, track_changed=True)
2562    session._write_if_wanted(b"\x01" * 4096)
2563    assert session._retained_bytes() == 4096
2564    # still the current channel, so the prune cannot be what frees the cushion
2565    item.close()
2566    assert item in session._channels
2567    assert session._retained_bytes() == 0
2568
2569
2570async def test_a_channel_still_being_read_keeps_its_tail(tmp_path: Path) -> None:
2571    """Closing the playing item at a cut must not discard what its stream is still owed."""
2572    session = _make_session(tmp_path)
2573    item = session._open_channel(TRACK_A)
2574    item.claim()
2575    item.write(b"tail" * 4)
2576    item.close()
2577    assert item.buffered == 16
2578    assert b"".join([chunk async for chunk in item.read()]) == b"tail" * 4
2579
2580
2581@contextmanager
2582def _capture_holding(
2583    session: _SoloistSession, *, fifo_bytes: int, reader_bytes: int | None
2584) -> Iterator[None]:
2585    """
2586    Give the session a real capture FIFO and a reader holding the given amounts.
2587
2588    A real pipe is used so the byte count comes from the same ioctl the backend
2589    relies on. Pass ``reader_bytes=None`` for a reader whose buffer cannot be read.
2590    """
2591    read_fd, write_fd = os.pipe()
2592    try:
2593        if fifo_bytes:
2594            os.write(write_fd, bytes(fifo_bytes))
2595        pipe = MagicMock()
2596        pipe.fileno.return_value = read_fd
2597        transport = MagicMock()
2598        transport.get_extra_info.return_value = pipe
2599        session._transport = transport
2600        reader = MagicMock(spec=[]) if reader_bytes is None else MagicMock()
2601        if reader_bytes is not None:
2602            reader._buffer = bytearray(reader_bytes)
2603        session._reader = reader
2604        yield
2605    finally:
2606        session._transport = None
2607        session._reader = None
2608        os.close(read_fd)
2609        os.close(write_fd)
2610
2611
2612def _stdout_of(*lines: str) -> MagicMock:
2613    """Return a process mock whose stdout yields the given daemon log lines."""
2614
2615    async def _iter_stdout() -> AsyncGenerator[str]:
2616        for line in lines:
2617            yield line
2618
2619    proc = MagicMock()
2620    proc.iter_stdout = _iter_stdout
2621    return proc
2622
2623
2624def _make_provider(tmp_path: Path, setup_data: dict[str, Any] | None = None) -> SpotifyProvider:
2625    """Return a SpotifyProvider (bypassing __init__) with the given setup_data."""
2626    prov = object.__new__(SpotifyProvider)
2627    config = MagicMock(instance_id="spotify--test")
2628    config.get_value = MagicMock(return_value=None)
2629    config.values = {}
2630    prov.config = config
2631    prov.manifest = MagicMock(domain="spotify")
2632    prov.logger = MagicMock()
2633    prov.available = True
2634    mass = MagicMock()
2635    mass.storage_path = str(tmp_path / "storage")
2636    mass.cache_path = str(tmp_path / "cache")
2637    # get_setup_value reads the live setup_data blob from the store
2638    mass.config.get = MagicMock(return_value=setup_data or {})
2639    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
2640    # the store keeps values encrypted; decrypt is an identity map for the test
2641    mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
2642    prov.mass = mass
2643    return prov
2644
2645
2646def _make_backend(tmp_path: Path, setup_data: dict[str, Any] | None = None) -> SoloistBackend:
2647    """Return a SoloistBackend on a mocked provider."""
2648    return SoloistBackend(_make_provider(tmp_path, setup_data))
2649
2650
2651def _make_session(tmp_path: Path, queue_id: str | None = "player1") -> _SoloistSession:
2652    """Return a session with its process/sink/client replaced by mocks."""
2653    session = _SoloistSession(_make_backend(tmp_path), queue_id)
2654    session._sink = AsyncMock()
2655    session._client = AsyncMock()
2656    session._proc = MagicMock(returncode=None)
2657    # a session under test is past the engine's login and has claimed the
2658    # Connect device, unless a test says otherwise
2659    session._logged_in = True
2660    session._was_active = True
2661    return session
2662
2663
2664def _streamdetails_for(
2665    *,
2666    queue_id: str | None = "player1",
2667    uri: str = TRACK_A,
2668    media_type: MediaType = MediaType.TRACK,
2669) -> StreamDetails:
2670    """Return stream details for a Spotify item served by the test instance."""
2671    return StreamDetails(
2672        provider="spotify--test",
2673        item_id=uri.rsplit(":", 1)[1],
2674        audio_format=AudioFormat(content_type=ContentType.PCM_S16LE),
2675        media_type=media_type,
2676        queue_id=queue_id,
2677    )
2678
2679
2680def _feed(session: _SoloistSession, uri: str) -> _ItemAudio:
2681    """Return the channel of an item handed to the engine that it has not started."""
2682    item = session._open_channel(uri)
2683    session._pending.append(item)
2684    return item
2685
2686
2687def _streamed(session: _SoloistSession, uri: str = TRACK_A) -> _ItemAudio:
2688    """Return the channel of the item the engine plays and a stream is reading."""
2689    item = session._current = session._open_channel(uri)
2690    item.started.set()
2691    item.claim()
2692    return item
2693
2694
2695def _make_item(tmp_path: Path, uri: str) -> _ItemAudio:
2696    """Return a bare item channel on a mocked session."""
2697    return _ItemAudio(uri, _make_session(tmp_path))
2698
2699
2700def _queue_item(uri: str, streamdetails: Any = None, queue_item_id: str | None = None) -> MagicMock:
2701    """Return a queue item stand-in for a Spotify track on the test instance."""
2702    item_id = uri.rsplit(":", 1)[1]
2703    media_item = MagicMock(media_type=MediaType.TRACK, provider="spotify--test", item_id=item_id)
2704    media_item.provider_mappings = []
2705    return MagicMock(
2706        media_item=media_item,
2707        queue_item_id=queue_item_id or f"qi-{item_id}",
2708        streamdetails=streamdetails,
2709    )
2710
2711
2712async def _wait_for(predicate: Callable[[], bool], timeout: float = 2.0) -> None:
2713    """Wait until the predicate holds, so a background task can get there."""
2714    loop = asyncio.get_running_loop()
2715    deadline = loop.time() + timeout
2716    while loop.time() < deadline:
2717        if predicate():
2718            return
2719        await asyncio.sleep(0.01)
2720    raise AssertionError("condition not met within timeout")
2721
2722
2723def _client_of(session: _SoloistSession) -> AsyncMock:
2724    """Return the session's mocked WebSocket client."""
2725    return cast("AsyncMock", session._client)
2726
2727
2728def _current_of(session: _SoloistSession) -> _ItemAudio:
2729    """Return the channel the session is playing, which the caller knows exists."""
2730    item = session._current
2731    assert item is not None
2732    return item
2733
2734
2735def _sink_of(session: _SoloistSession) -> AsyncMock:
2736    """Return the session's mocked capture sink."""
2737    return cast("AsyncMock", session._sink)
2738
2739
2740def _queues_of(session: _SoloistSession) -> MagicMock:
2741    """Return the mocked player_queues controller the session consults."""
2742    return cast("MagicMock", session.mass.player_queues)
2743
2744
2745def _auth_event(*, logged_in: bool, is_active: bool = True) -> SoloistEvent:
2746    """Return an auth_state event with the given login and active-device state."""
2747    return SoloistEvent(
2748        type="auth_state",
2749        data=SoloistAuthState(logged_in=logged_in, is_active=is_active),
2750        raw={},
2751    )
2752
2753
2754def _device_event(*, is_active: bool) -> SoloistEvent:
2755    """Return a device_changed event with the given active-device state."""
2756    return SoloistEvent(
2757        type="device_changed", data=SoloistDeviceChanged(is_active=is_active), raw={}
2758    )
2759
2760
2761def _playback_event(status: str, position_ms: int = 0) -> SoloistEvent:
2762    """Return a playback_state event for the current item with the given status."""
2763    return SoloistEvent(
2764        type="playback_state",
2765        data=SoloistPlaybackState(
2766            status=status,
2767            item=SoloistEntity(uri=TRACK_A, entity_type="track"),
2768            position=SoloistPosition(position_ms=position_ms, timestamp_ms=0),
2769        ),
2770        raw={},
2771    )
2772
2773
2774def _position_event(position_ms: int) -> SoloistEvent:
2775    """Return a position_sync event reporting the given playback position."""
2776    return SoloistEvent(
2777        type="position_sync",
2778        data=SoloistPositionSync(position=SoloistPosition(position_ms=position_ms, timestamp_ms=0)),
2779        raw={},
2780    )
2781
2782
2783def _current_item_event(event_type: str, uri: str, duration_ms: int | None = None) -> SoloistEvent:
2784    """
2785    Return an event reporting the given item as the one the engine is on.
2786
2787    :param event_type: ``track_changed``, ``playback_state`` or ``playback_changed``.
2788    :param uri: The Spotify URI the event names.
2789    :param duration_ms: The duration to decorate the item with, when it has one.
2790    """
2791    item = SoloistEntity(
2792        uri=uri,
2793        entity_type="track",
2794        decorations={"playback": {"duration_ms": duration_ms}} if duration_ms else {},
2795    )
2796    if event_type == "track_changed":
2797        return SoloistEvent(type=event_type, data=SoloistTrackChanged(item=item), raw={})
2798    return SoloistEvent(
2799        type=event_type, data=SoloistPlaybackState(status="playing", item=item), raw={}
2800    )
2801
2802
2803def _install_fake_binary_manager(monkeypatch: pytest.MonkeyPatch) -> None:
2804    """Replace the shared binary manager so no download or exec is attempted."""
2805    manager = MagicMock()
2806    manager.ensure_fresh = AsyncMock(return_value=Path("/nonexistent/soloist"))
2807    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", MagicMock(return_value=manager))
2808
2809
2810async def test_seeking_the_playing_item_keeps_the_session(tmp_path: Path) -> None:
2811    """The engine is moved where it stands rather than the session being respawned."""
2812    session = _make_session(tmp_path)
2813    playing = session._current = session._open_channel(TRACK_A)
2814    playing.started.set()
2815    playing.claim()
2816    playing.duration_ms = 260_000
2817    playing.playing_seen = True
2818    playing.observe_position(30_000)
2819    # the pre-seek audio nobody may hear again
2820    playing.write(b"\x01" * 32)
2821
2822    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
2823        _current_of(session).observe_position(position_ms)
2824
2825    _client_of(session).seek.side_effect = _engine_seeks
2826    item = await session.seek_current(TRACK_A, 120_000)
2827
2828    assert item is not playing
2829    assert session.current is item
2830    assert item.claimed
2831    # what the track is stays with it; where it was does not
2832    assert item.duration_ms == 260_000
2833    assert item.playing_seen
2834    assert item.started_at_ms == 120_000
2835    # the outgoing channel is closed, which is what ends the stream reading it
2836    assert playing._closed
2837    _client_of(session).seek.assert_awaited_once_with(120_000, await_result=True)
2838
2839
2840async def test_a_seek_leaves_the_outgoing_stream_nothing_to_report(tmp_path: Path) -> None:
2841    """
2842    The superseded channel validates clean, so an ordinary seek stays out of the log.
2843
2844    Its stream is still attached and validates the channel when it ends, and the
2845    engine is nowhere near the end of an item being seeked away from.
2846    """
2847    session = _make_session(tmp_path)
2848    playing = _streamed(session)
2849    playing.duration_ms = 260_000
2850    playing.playing_seen = True
2851    playing.observe_position(30_000)
2852
2853    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
2854        _current_of(session).observe_position(position_ms)
2855
2856    _client_of(session).seek.side_effect = _engine_seeks
2857    await session.seek_current(TRACK_A, 120_000)
2858    playing.release()
2859
2860    assert playing.superseded
2861    await session.validate_item(playing)
2862
2863
2864async def test_a_seek_of_the_playing_item_is_sent_only_once(tmp_path: Path) -> None:
2865    """
2866    A landed seek is never repeated: a repeat restarts the item, audibly.
2867
2868    The engine answers late on purpose, so a re-send loop around the wait would
2869    have fired several times over before the confirmation arrives.
2870    """
2871    session = _make_session(tmp_path)
2872    playing = session._current = session._open_channel(TRACK_A)
2873    playing.started.set()
2874    playing.observe_position(30_000)
2875    pending: list[asyncio.Task[None]] = []
2876
2877    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
2878        async def _confirm_late() -> None:
2879            await asyncio.sleep(0.05)
2880            _current_of(session).observe_position(position_ms)
2881
2882        pending.append(asyncio.create_task(_confirm_late()))
2883
2884    _client_of(session).seek.side_effect = _engine_seeks
2885    with patch.object(soloist_backend, "_SEEK_RETRY_INTERVAL_S", 0.01):
2886        await session.seek_current(TRACK_A, 120_000)
2887    await asyncio.gather(*pending)
2888    assert _client_of(session).seek.await_count == 1
2889
2890
2891async def test_seeking_back_is_not_confirmed_by_the_position_seeked_away_from(
2892    tmp_path: Path,
2893) -> None:
2894    """A report still describing the pre-seek position cannot land a backward seek."""
2895    session = _make_session(tmp_path)
2896    playing = session._current = session._open_channel(TRACK_A)
2897    playing.started.set()
2898    playing.duration_ms = 260_000
2899    playing.observe_position(200_000)
2900
2901    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
2902        item = _current_of(session)
2903        # a report from before the seek is still in flight; it sits above the
2904        # target's tolerance window and must not pass for the landing
2905        item.observe_position(200_000)
2906        assert not item.seek_confirmed.is_set()
2907        item.observe_position(position_ms)
2908        item.observe_position(position_ms + 2)
2909
2910    _client_of(session).seek.side_effect = _engine_seeks
2911    item = await session.seek_current(TRACK_A, 60_000)
2912    assert item.started_at_ms == 60_002
2913    # and the pre-seek report is not left standing in for progress this item
2914    # never made, which at_own_end and the completeness check would believe
2915    assert item.last_position_ms == 60_002
2916
2917
2918async def test_audio_in_flight_across_an_in_place_seek_is_dropped(tmp_path: Path) -> None:
2919    """Only audio from past the seek reaches the fresh channel."""
2920    session = _make_session(tmp_path)
2921    playing = session._current = session._open_channel(TRACK_A)
2922    playing.started.set()
2923    playing.observe_position(30_000)
2924
2925    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
2926        # still rendering the position being left behind
2927        session._write_if_wanted(b"\x01" * 32)
2928        _current_of(session).observe_position(position_ms)
2929
2930    _client_of(session).seek.side_effect = _engine_seeks
2931    with _capture_holding(session, fifo_bytes=2 * _FRAME_BYTES, reader_bytes=_FRAME_BYTES):
2932        item = await session.seek_current(TRACK_A, 120_000)
2933    # nothing rendered while the engine was being moved reached the channel
2934    assert not item._chunks
2935    # and what the pipeline still held at the confirmation is dropped after it
2936    assert session._stale_budget == 3 * _FRAME_BYTES
2937    session._write_if_wanted(b"\x02" * (3 * _FRAME_BYTES))
2938    session._write_if_wanted(b"\x03" * 16)
2939    assert b"".join(item._chunks) == b"\x03" * 16
2940
2941
2942async def test_the_sink_is_suspended_while_a_seek_is_in_flight(tmp_path: Path) -> None:
2943    """No pre-seek audio enters the capture while the engine is being moved."""
2944    session = _make_session(tmp_path)
2945    session._demand_started = True
2946    session._engine_playing = True
2947    session._sink_running = True
2948    playing = session._current = session._open_channel(TRACK_A)
2949    playing.started.set()
2950    playing.observe_position(30_000)
2951    suspended_during_seek = False
2952
2953    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
2954        nonlocal suspended_during_seek
2955        suspended_during_seek = not session._sink_running
2956        _current_of(session).observe_position(position_ms)
2957
2958    _client_of(session).seek.side_effect = _engine_seeks
2959    await session.seek_current(TRACK_A, 120_000)
2960    assert suspended_during_seek
2961    _sink_of(session).suspend.assert_awaited()
2962
2963
2964async def test_a_seek_the_engine_never_confirms_fails_the_item(tmp_path: Path) -> None:
2965    """An unconfirmed seek is reported rather than served from the wrong position."""
2966    session = _make_session(tmp_path)
2967    playing = session._current = session._open_channel(TRACK_A)
2968    playing.started.set()
2969    playing.observe_position(30_000)
2970    with (
2971        patch.object(soloist_backend, "_SEEK_CONFIRM_TIMEOUT_S", 0.01),
2972        pytest.raises(AudioError, match="did not confirm"),
2973    ):
2974        await session.seek_current(TRACK_A, 120_000)
2975
2976
2977async def test_a_refused_seek_command_reports_soloist(tmp_path: Path) -> None:
2978    """A rejected seek names the engine, so the caller can fall back."""
2979    session = _make_session(tmp_path)
2980    playing = session._current = session._open_channel(TRACK_A)
2981    playing.started.set()
2982    _client_of(session).seek.side_effect = SoloistError("nope")
2983    with pytest.raises(AudioError, match="would not seek"):
2984        await session.seek_current(TRACK_A, 120_000)
2985
2986
2987async def test_seeking_an_item_the_engine_is_not_on_is_refused(tmp_path: Path) -> None:
2988    """Only the item the session is actually playing can be seeked in place."""
2989    session = _make_session(tmp_path)
2990    playing = session._current = session._open_channel(TRACK_A)
2991    playing.started.set()
2992    with pytest.raises(AudioError, match="is not playing"):
2993        await session.seek_current(TRACK_B, 120_000)
2994
2995
2996async def test_a_seek_of_the_playing_item_is_served_by_the_running_session(
2997    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
2998) -> None:
2999    """The session is seeked where it stands instead of being replaced."""
3000    backend = _make_backend(tmp_path)
3001    backend._server = MagicMock()
3002    backend._binary = Path("/nonexistent/soloist")
3003    session = _make_session(tmp_path)
3004    backend._session = session
3005    item = session._current = session._open_channel(TRACK_A)
3006    item.started.set()
3007    # its own stream is still attached when the seek re-opens it
3008    item.claim()
3009    item.observe_position(30_000)
3010    stopped = AsyncMock()
3011    monkeypatch.setattr(session, "stop", stopped)
3012
3013    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
3014        _current_of(session).observe_position(position_ms)
3015
3016    _client_of(session).seek.side_effect = _engine_seeks
3017    got_session, got_item = await backend._acquire(TRACK_A, 90, "player1")
3018
3019    assert got_session is session
3020    assert got_item is not item
3021    assert got_item.claimed
3022    stopped.assert_not_awaited()
3023    _client_of(session).seek.assert_awaited_once_with(90_000, await_result=True)
3024
3025
3026async def test_a_cancelled_seek_does_not_leave_the_session_wedged(tmp_path: Path) -> None:
3027    """
3028    A superseded seek ends the session instead of holding it claimed for good.
3029
3030    A second seek cancels the stream the first one is being made for, and the
3031    channel it had already claimed would otherwise keep the session in use:
3032    unable to expire, and refusing every later item as busy.
3033    """
3034    session = _make_session(tmp_path)
3035    playing = session._current = session._open_channel(TRACK_A)
3036    playing.started.set()
3037    playing.observe_position(30_000)
3038    seeking = asyncio.create_task(session.seek_current(TRACK_A, 120_000))
3039    # let it get as far as waiting for the engine to confirm
3040    while not _client_of(session).seek.await_count:
3041        await asyncio.sleep(0)
3042    seeking.cancel()
3043    with suppress(asyncio.CancelledError):
3044        await seeking
3045    assert not session.usable
3046    assert not session._seeking
3047
3048
3049async def test_a_seek_cancelled_before_the_channel_is_swapped_keeps_the_session(
3050    tmp_path: Path,
3051) -> None:
3052    """Nothing has been given up yet while the sink is still being held."""
3053    session = _make_session(tmp_path)
3054    playing = session._current = session._open_channel(TRACK_A)
3055    playing.started.set()
3056    held = asyncio.Event()
3057
3058    async def _slow_suspend(**_kwargs: Any) -> None:
3059        held.set()
3060        await asyncio.sleep(60)
3061
3062    with patch.object(session, "_apply_sink_state", _slow_suspend):
3063        seeking = asyncio.create_task(session.seek_current(TRACK_A, 120_000))
3064        await held.wait()
3065        seeking.cancel()
3066        with suppress(asyncio.CancelledError):
3067            await seeking
3068    # the session is untouched and, crucially, not left dropping every chunk
3069    assert session.usable
3070    assert not session._seeking
3071    assert session.current is playing
3072
3073
3074async def test_a_seek_is_refused_once_the_engine_has_moved_on(tmp_path: Path) -> None:
3075    """The item seeked must still be the one the engine is on when the sink settles."""
3076    session = _make_session(tmp_path)
3077    playing = session._current = session._open_channel(TRACK_A)
3078    playing.started.set()
3079    follower = session._open_channel(TRACK_B)
3080
3081    async def _boundary_lands(**_kwargs: Any) -> None:
3082        session._current = follower
3083
3084    with (
3085        patch.object(session, "_apply_sink_state", _boundary_lands),
3086        pytest.raises(AudioError, match="moved on from"),
3087    ):
3088        await session.seek_current(TRACK_A, 120_000)
3089    # the follower the engine actually reached keeps its own channel
3090    assert session.current is follower
3091    # ... and the refused seek opened no channel of its own
3092    assert [item for item in session._channels if item.uri == TRACK_A] == [playing]
3093
3094
3095async def test_a_seek_that_fails_part_way_restarts_the_session(
3096    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
3097) -> None:
3098    """A seek the engine refuses after the channel was swapped still gets its audio."""
3099    backend = _make_backend(tmp_path)
3100    backend._server = MagicMock()
3101    backend._binary = Path("/nonexistent/soloist")
3102    session = _make_session(tmp_path)
3103    backend._session = session
3104    item = session._current = session._open_channel(TRACK_A)
3105    item.started.set()
3106    item.claim()
3107    _client_of(session).seek.side_effect = SoloistError("refused")
3108    stopped = AsyncMock()
3109    monkeypatch.setattr(session, "stop", stopped)
3110    _install_fake_binary_manager(monkeypatch)
3111    monkeypatch.setattr(
3112        soloist_backend._SoloistSession, "start", AsyncMock(side_effect=AudioError("spawn"))
3113    )
3114    with pytest.raises(AudioError, match="spawn"):
3115        await backend._acquire(TRACK_A, 90, "player1")
3116    stopped.assert_awaited_once()
3117
3118
3119async def test_a_seek_refused_because_the_app_took_over_does_not_respawn(
3120    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
3121) -> None:
3122    """A replacement would claim the Connect device back off the Spotify app."""
3123    backend = _make_backend(tmp_path)
3124    backend._server = MagicMock()
3125    backend._binary = Path("/nonexistent/soloist")
3126    session = _make_session(tmp_path)
3127    backend._session = session
3128    item = session._current = session._open_channel(TRACK_A)
3129    item.started.set()
3130    item.claim()
3131    item.observe_position(30_000)
3132    started = AsyncMock()
3133    monkeypatch.setattr(soloist_backend._SoloistSession, "start", started)
3134
3135    async def _app_takes_over(_position_ms: int, **_kwargs: Any) -> None:
3136        # the user moved playback elsewhere from their Spotify app while the
3137        # seek was in flight; the wait is released by the session ending
3138        session._end_on_app_control(SoloistAppControl.TOOK_OVER)
3139
3140    _client_of(session).seek.side_effect = _app_takes_over
3141    with pytest.raises(SoloistAppControlError):
3142        await backend._acquire(TRACK_A, 120, "player1")
3143    started.assert_not_awaited()
3144
3145
3146async def test_a_seek_is_abandoned_when_the_app_took_over_during_the_suspend(
3147    tmp_path: Path,
3148) -> None:
3149    """Nothing is seeked on a session the Spotify app has already taken over."""
3150    session = _make_session(tmp_path)
3151    playing = session._current = session._open_channel(TRACK_A)
3152    playing.started.set()
3153
3154    async def _app_takes_over(**_kwargs: Any) -> None:
3155        session._end_on_app_control(SoloistAppControl.TOOK_OVER)
3156
3157    with (
3158        patch.object(session, "_apply_sink_state", _app_takes_over),
3159        pytest.raises(SoloistAppControlError),
3160    ):
3161        await session.seek_current(TRACK_A, 120_000)
3162    # the engine was never asked to move
3163    _client_of(session).seek.assert_not_awaited()
3164
3165
3166async def test_a_cancelled_sink_transition_is_re_issued(tmp_path: Path) -> None:
3167    """
3168    A suspend that may or may not have landed is never taken as done.
3169
3170    A sink that did suspend would otherwise still read as running, and the
3171    resume that should follow would be skipped as a no-op: silence for good.
3172    """
3173    session = _make_session(tmp_path)
3174    session._demand_started = True
3175    session._engine_playing = True
3176    session._sink_running = True
3177    session._seeking = True
3178
3179    async def _cancelled_suspend() -> None:
3180        raise asyncio.CancelledError
3181
3182    _sink_of(session).suspend.side_effect = _cancelled_suspend
3183    with suppress(asyncio.CancelledError):
3184        await session._apply_sink_state()
3185    # the engine plays on and the sink is wanted running again
3186    session._seeking = False
3187    await session._apply_sink_state()
3188    _sink_of(session).resume.assert_awaited_once()
3189
3190
3191async def test_a_seek_cancelled_at_the_final_resume_does_not_leave_the_session_usable(
3192    tmp_path: Path,
3193) -> None:
3194    """The channel is claimed by then, so an abandoned seek must still end the session."""
3195    session = _make_session(tmp_path)
3196    playing = session._current = session._open_channel(TRACK_A)
3197    playing.started.set()
3198    playing.observe_position(30_000)
3199    calls = 0
3200    real_apply = session._apply_sink_state
3201
3202    async def _cancel_on_the_way_out(**kwargs: Any) -> None:
3203        nonlocal calls
3204        calls += 1
3205        if calls > 1:
3206            raise asyncio.CancelledError
3207        await real_apply(**kwargs)
3208
3209    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
3210        _current_of(session).observe_position(position_ms)
3211
3212    _client_of(session).seek.side_effect = _engine_seeks
3213    with (
3214        patch.object(session, "_apply_sink_state", _cancel_on_the_way_out),
3215        suppress(asyncio.CancelledError),
3216    ):
3217        await session.seek_current(TRACK_A, 120_000)
3218    assert not session.usable
3219
3220
3221async def test_a_cold_seek_does_not_read_a_failed_session_as_landed(tmp_path: Path) -> None:
3222    """The wake-up a fatal failure gives every channel is not a confirmed seek."""
3223    session = _make_session(tmp_path)
3224    item = session._current = session._open_channel(TRACK_A)
3225
3226    async def _engine_dies(_position_ms: int, **_kwargs: Any) -> None:
3227        session._fail("the session exited")
3228
3229    _client_of(session).seek.side_effect = _engine_dies
3230    with pytest.raises(AudioError, match="the session exited"):
3231        await session._cold_seek(_client_of(session), item, 60_000)
3232
3233
3234async def test_seeking_the_playing_item_back_to_its_start_keeps_the_session(
3235    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
3236) -> None:
3237    """
3238    A seek to zero is a seek: reachable once an earlier one moved the buffer.
3239
3240    The buffer only hands a position to the provider when it cannot serve it
3241    itself, so seeking back before an earlier seek's target arrives here with a
3242    target of zero.
3243    """
3244    backend = _make_backend(tmp_path)
3245    backend._server = MagicMock()
3246    backend._binary = Path("/nonexistent/soloist")
3247    session = _make_session(tmp_path)
3248    backend._session = session
3249    item = session._current = session._open_channel(TRACK_A)
3250    item.started.set()
3251    item.claim()
3252    item.observe_position(200_000)
3253    stopped = AsyncMock()
3254    monkeypatch.setattr(session, "stop", stopped)
3255
3256    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
3257        current = _current_of(session)
3258        current.observe_position(position_ms)
3259        current.observe_position(position_ms + 2)
3260
3261    _client_of(session).seek.side_effect = _engine_seeks
3262    got_session, got_item = await backend._acquire(TRACK_A, 0, "player1")
3263
3264    assert got_session is session
3265    assert got_item is not item
3266    stopped.assert_not_awaited()
3267    _client_of(session).seek.assert_awaited_once_with(0, await_result=True)
3268    assert got_item.started_at_ms == 2
3269
3270
3271async def test_a_short_forward_seek_still_confirms(tmp_path: Path) -> None:
3272    """
3273    A seek only a little past where the engine is must not wait itself out.
3274
3275    Reachable because the engine runs ahead of what has been delivered - up to
3276    the retained cushion - so a target the buffer will not serve can still be
3277    inside the tolerance window of the engine's own position. Demanding the
3278    engine drop below that mark would never be satisfied by a seek forward.
3279    """
3280    session = _make_session(tmp_path)
3281    playing = session._current = session._open_channel(TRACK_A)
3282    playing.started.set()
3283    playing.duration_ms = 260_000
3284    # the engine is at 49s while only ~30s has been delivered
3285    playing.observe_position(49_000)
3286
3287    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
3288        _current_of(session).observe_position(position_ms)
3289
3290    _client_of(session).seek.side_effect = _engine_seeks
3291    item = await session.seek_current(TRACK_A, 50_500)
3292    assert item.seek_confirmed.is_set()
3293    assert item.started_at_ms == 50_500
3294
3295
3296async def test_a_seek_does_not_arm_the_last_items_drain(tmp_path: Path) -> None:
3297    """
3298    The channel opened for a seek has no position yet, which is not an ended item.
3299
3300    Holding the sink can make the engine report a state that is not playing, and
3301    the run's last item would then be drained out from under the seek.
3302    """
3303    session = _make_session(tmp_path)
3304    session._demand_started = True
3305    playing = session._current = session._open_channel(TRACK_A)
3306    playing.started.set()
3307    playing.duration_ms = 260_000
3308    playing.observe_position(30_000)
3309    armed: list[bool] = []
3310
3311    async def _engine_seeks(position_ms: int, **_kwargs: Any) -> None:
3312        # nothing to judge the fresh channel by yet, and no follower queued
3313        await session._handle_playback_state(
3314            SoloistPlaybackState(status="buffering", item=None, position=None)
3315        )
3316        armed.append(_current_of(session).draining)
3317        _current_of(session).observe_position(position_ms)
3318
3319    _client_of(session).seek.side_effect = _engine_seeks
3320    item = await session.seek_current(TRACK_A, 120_000)
3321    assert armed == [False]
3322    assert not item.draining
3323
3324
3325async def test_the_item_a_finished_run_stopped_on_is_not_seeked_in_place(
3326    tmp_path: Path,
3327) -> None:
3328    """
3329    A matching uri is not proof the engine is still playing it.
3330
3331    The channel stays current through the idle grace after the run ended, and a
3332    seek would wait out its confirmation on an engine that has stopped.
3333    """
3334    session = _make_session(tmp_path)
3335    ended = session._current = session._open_channel(TRACK_A)
3336    ended.started.set()
3337    ended.close()
3338    with pytest.raises(AudioError, match="is not playing"):
3339        await session.seek_current(TRACK_A, 0)
3340    _client_of(session).seek.assert_not_awaited()
3341