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