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