music-assistant-server

1"""
2Unit tests for the Spotify Soloist playback backend.
3
4The backend spawns one engine run per Spotify URI (single-track mode) and
5streams the captured PCM as that item's audio. These tests lock down the pure
6logic around that: lead-silence trimming, the run's cushion and its sink
7backpressure, tail-padding suppression, startup and event handling, delivery
8validation, run acquisition (busy/replaced/superseded), paired-session
9adoption and setup. No real process or PulseAudio is involved.
10"""
11
12from __future__ import annotations
13
14import asyncio
15from collections.abc import AsyncGenerator
16from pathlib import Path
17from typing import Any, cast
18from unittest.mock import AsyncMock, MagicMock, patch
19
20import pytest
21from music_assistant_models.enums import ContentType, MediaType
22from music_assistant_models.errors import AudioError, LoginFailed
23from music_assistant_models.media_items import AudioFormat
24from music_assistant_models.streamdetails import StreamDetails
25
26from music_assistant.models.music_provider import ProviderStreamLimitError
27from music_assistant.providers.spotify.backends import StreamSupersededError
28from music_assistant.providers.spotify.backends import soloist as soloist_backend
29from music_assistant.providers.spotify.backends.soloist import (
30    _BYTES_PER_SECOND,
31    _FRAME_BYTES,
32    _MAX_LEAD_TRIM_S,
33    _TAIL_PAD_GRACE_S,
34    _TAIL_PAD_ZONE_S,
35    SoloistBackend,
36    _SingleTrackRun,
37    _trim_lead_silence,
38)
39from music_assistant.providers.spotify.constants import (
40    CONF_SOLOIST_API_KEY,
41    CONF_SOLOIST_CONSENT,
42)
43from music_assistant.providers.spotify.helpers import soloist_session_present
44from music_assistant.providers.spotify.provider import SpotifyProvider
45from music_assistant.providers.spotify_connect.soloist.runtime import (
46    WS_ADDR_FILE,
47    WS_PORT_FILE,
48)
49
50TRACK_A = "spotify:track:aaa"
51TRACK_B = "spotify:track:bbb"
52# an audiobook is one item whose chapters are separate Spotify URIs
53AUDIOBOOK = "spotify:show:book"
54CHAPTER_A = "spotify:episode:ch1"
55CHAPTER_B = "spotify:episode:ch2"
56CHAPTER_C = "spotify:episode:ch3"
57
58
59def test_trim_drops_an_all_zero_chunk_within_the_bound() -> None:
60    """A pure-silence chunk inside the trim budget is dropped entirely."""
61    chunk = b"\x00" * 1024
62    trimmed, skipped = _trim_lead_silence(chunk, 0)
63    assert trimmed == b""
64    assert skipped == 1024
65
66
67def test_trim_keeps_frame_alignment_when_audio_starts_mid_chunk() -> None:
68    """Audio starting mid-chunk is cut on a sample-frame boundary."""
69    # audio starts one byte into the third frame: the trim must keep that frame whole
70    chunk = b"\x00" * (_FRAME_BYTES * 2 + 1) + b"\x01" * 64
71    trimmed, skipped = _trim_lead_silence(chunk, 0)
72    assert skipped == _FRAME_BYTES * 2
73    assert len(trimmed) % _FRAME_BYTES == 1  # the partial frame's remainder is preserved
74    assert trimmed.endswith(b"\x01" * 64)
75
76
77def test_trim_passes_silence_through_once_the_bound_is_exceeded() -> None:
78    """Beyond the trim budget, silence is genuine content and is delivered."""
79    chunk = b"\x00" * 1024
80    trimmed, skipped = _trim_lead_silence(chunk, int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND))
81    assert trimmed == chunk
82    assert skipped == 0
83
84
85def test_the_lead_trim_never_exceeds_its_budget() -> None:
86    """Silence beyond the budget is content, including where audio starts mid-chunk."""
87    budget = int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND)
88    # already at the budget, with a chunk whose silence runs well past it
89    chunk = b"\x00" * 4096 + b"\x01" * 64
90    trimmed, skipped = _trim_lead_silence(chunk, budget - _FRAME_BYTES)
91    assert skipped == _FRAME_BYTES
92    assert len(trimmed) == len(chunk) - _FRAME_BYTES
93
94
95async def test_a_superseded_audiobook_stream_stops_instead_of_stitching_on(
96    tmp_path: Path,
97) -> None:
98    """The chapters after a seek belong to the stream that took over, not to this one."""
99    provider = _make_provider(tmp_path)
100    calls: list[str] = []
101
102    async def _cut(uri: str, *_args: Any, **_kwargs: Any) -> AsyncGenerator[bytes]:
103        calls.append(uri)
104        yield b"audio"
105        raise StreamSupersededError("replaced")
106
107    provider.backend = MagicMock(stream_spotify_uri=_cut)
108    streamdetails = MagicMock(
109        media_type=MediaType.AUDIOBOOK,
110        data={"chapters": [CHAPTER_A, CHAPTER_B], "chapters_data": []},
111    )
112    chunks = [chunk async for chunk in provider.get_audio_stream(streamdetails)]
113    assert chunks == [b"audio"]
114    assert calls == [CHAPTER_A]
115
116
117async def test_only_the_chapter_a_stream_starts_on_may_take_the_session(
118    tmp_path: Path,
119) -> None:
120    """The chapter a seek lands on starts the stream; the ones after it continue it."""
121    provider = _make_provider(tmp_path)
122    calls: list[tuple[str, bool]] = []
123
124    async def _stream(
125        uri: str, _seek: int = 0, *, continuation: bool = False, **_kwargs: Any
126    ) -> AsyncGenerator[bytes]:
127        calls.append((uri, continuation))
128        yield b"audio"
129
130    provider.backend = MagicMock(stream_spotify_uri=_stream)
131    streamdetails = MagicMock(
132        media_type=MediaType.AUDIOBOOK,
133        data={
134            "chapters": [CHAPTER_A, CHAPTER_B, CHAPTER_C],
135            "chapters_data": [{"duration_ms": 60_000}] * 3,
136        },
137    )
138    async for _ in provider.get_audio_stream(streamdetails, seek_position=70):
139        pass
140    assert calls == [(CHAPTER_B, False), (CHAPTER_C, True)]
141
142
143async def test_a_superseded_track_stream_ends_without_an_error(tmp_path: Path) -> None:
144    """A replaced stream is no failure: the item plays on the stream that took over."""
145    provider = _make_provider(tmp_path)
146
147    async def _cut(_uri: str, *_args: Any, **_kwargs: Any) -> AsyncGenerator[bytes]:
148        yield b"audio"
149        raise StreamSupersededError("replaced")
150
151    provider.backend = MagicMock(stream_spotify_uri=_cut)
152    streamdetails = MagicMock(media_type=MediaType.TRACK, item_id="aaa", data=None)
153    chunks = [chunk async for chunk in provider.get_audio_stream(streamdetails)]
154    assert chunks == [b"audio"]
155
156
157async def test_an_audiobook_gives_up_on_capacity_instead_of_burning_chapters(
158    tmp_path: Path,
159) -> None:
160    """Skipping ahead would cost the audiobook its availability and the caller its retry."""
161    provider = _make_provider(tmp_path)
162    calls: list[str] = []
163
164    async def _refuse(uri: str, *_args: Any, **_kwargs: Any) -> AsyncGenerator[bytes]:
165        calls.append(uri)
166        for _ in ():  # never yields; only makes this an async generator
167            yield b""
168        raise soloist_backend.SoloistSessionBusyError(provider)
169
170    provider.backend = MagicMock(stream_spotify_uri=_refuse)
171    streamdetails = MagicMock(
172        media_type=MediaType.AUDIOBOOK,
173        data={"chapters": [TRACK_A, TRACK_B, "spotify:track:ccc"], "chapters_data": []},
174    )
175
176    with pytest.raises(ProviderStreamLimitError):
177        async for _ in provider.get_audio_stream(streamdetails):
178            pass
179    # the first chapter's refusal ends it: no chapter is skipped over
180    assert calls == [TRACK_A]
181
182
183def test_the_shaper_only_emits_whole_frames() -> None:
184    """A read that ends mid-frame must never split a frame across two items."""
185    shaper = soloist_backend._CaptureShaper()
186    # the session's first bytes are infrastructure silence, and are dropped
187    assert shaper.shape(b"\x00" * 4096) == b""
188    # a mis-aligned read emits whole frames and carries the remainder
189    first = shaper.shape(b"\x01" * (_FRAME_BYTES + 3))
190    assert len(first) == _FRAME_BYTES
191    # which is then completed by the next read, losing nothing
192    second = shaper.shape(b"\x02" * (_FRAME_BYTES - 3))
193    assert len(second) == _FRAME_BYTES
194    assert second[:3] == b"\x01" * 3
195    # an aligned read passes straight through
196    assert shaper.shape(b"\x03" * _FRAME_BYTES) == b"\x03" * _FRAME_BYTES
197
198
199def test_the_shaper_trims_lead_silence_only_once() -> None:
200    """Silence after the audio has started is content, not pre-roll."""
201    shaper = soloist_backend._CaptureShaper()
202    assert shaper.shape(b"\x01" * _FRAME_BYTES) == b"\x01" * _FRAME_BYTES
203    silence = b"\x00" * _FRAME_BYTES
204    assert shaper.shape(silence) == silence
205
206
207def test_the_engine_is_told_not_to_normalize(tmp_path: Path) -> None:
208    """MA normalizes this audio itself, so the engine's own normalization is switched off."""
209    backend = _make_backend(tmp_path)
210    prefs = backend._data_dir / "settings" / "Users" / "alice-user" / "prefs"
211    prefs.parent.mkdir(parents=True)
212    prefs.write_text("some.engine.key=1\n", encoding="utf-8")
213    backend._prepare_data_dir(normalize=False)
214    content = prefs.read_text(encoding="utf-8").splitlines()
215    assert "some.engine.key=1" in content
216    assert "audio.normalize_v2=false" in content
217    # MA mixes the queue's crossfade itself, so the engine's own is always off
218    assert "audio.crossfade_v2=false" in content
219    # the ceiling is stated rather than left to the engine's own default
220    assert "audio.play_bitrate_enumeration=5" in content
221    assert "audio.play_bitrate_non_metered_enumeration=5" in content
222    assert "audio.play_bitrate_non_metered_migrated=true" in content
223
224
225def test_disabling_crossfade_writes_the_boolean(tmp_path: Path) -> None:
226    """Crossfade off is written explicitly, so a stale 'on' cannot survive."""
227    backend = _make_backend(tmp_path)
228    prefs = backend._data_dir / "settings" / "prefs"
229    prefs.parent.mkdir(parents=True)
230    prefs.write_text("audio.crossfade_v2=true\naudio.crossfade.time_v2=8000\n", encoding="utf-8")
231    backend._prepare_data_dir(normalize=False)
232    content = prefs.read_text(encoding="utf-8").splitlines()
233    assert "audio.crossfade_v2=false" in content
234    assert not any(line.startswith("audio.crossfade.time_v2") for line in content)
235
236
237async def test_setup_requires_an_api_key(tmp_path: Path) -> None:
238    """Without a stored API key the user must be sent back through the setup flow."""
239    backend = _make_backend(tmp_path)
240    with pytest.raises(LoginFailed) as err:
241        await backend.setup()
242    assert err.value.translation_key == "soloist_pairing_required"
243
244
245async def test_setup_requires_a_paired_session(
246    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
247) -> None:
248    """An API key without a paired session also routes back to the setup flow."""
249    backend = _make_backend(tmp_path, {CONF_SOLOIST_API_KEY: "k" * 20, CONF_SOLOIST_CONSENT: True})
250    _install_fake_binary_manager(monkeypatch)
251    with pytest.raises(LoginFailed) as err:
252        await backend.setup()
253    assert err.value.translation_key == "soloist_pairing_required"
254
255
256async def test_streaming_without_setup_is_refused(tmp_path: Path) -> None:
257    """A backend whose setup never ran refuses to stream instead of half-starting."""
258    backend = _make_backend(tmp_path)
259    with pytest.raises(AudioError, match="not started"):
260        async for _ in backend.stream_spotify_uri(TRACK_A):
261            pass
262
263
264def test_session_present_detection(tmp_path: Path) -> None:
265    """Only the engine's per-account state counts as paired."""
266    data_dir = tmp_path / "soloist-data"
267    assert soloist_session_present(data_dir) is False
268    data_dir.mkdir()
269    (data_dir / WS_ADDR_FILE).write_text("127.0.0.1", encoding="utf-8")
270    (data_dir / WS_PORT_FILE).write_text("1234", encoding="utf-8")
271    assert soloist_session_present(data_dir) is False
272    # everything a spawn leaves behind outlives the pairing it ran on: the engine
273    # keeps its identity, lock, cache and crash handler in the data dir even
274    # though it is given a cache dir of its own, and Music Assistant writes the
275    # prefs there before every spawn
276    (data_dir / "settings").mkdir()
277    (data_dir / "settings" / "prefs").write_text("audio.normalize_v2=false\n", encoding="utf-8")
278    (data_dir / ".device_id").write_text("6b6c2a07", encoding="utf-8")
279    (data_dir / ".lock").write_bytes(b"")
280    (data_dir / "cache" / "Users" / "spotify-user-user").mkdir(parents=True)
281    (data_dir / "crashpad").mkdir()
282    assert soloist_session_present(data_dir) is False
283    (data_dir / "settings" / "Users" / "spotify-user-user").mkdir(parents=True)
284    assert soloist_session_present(data_dir) is True
285
286
287def _make_provider(tmp_path: Path, setup_data: dict[str, Any] | None = None) -> SpotifyProvider:
288    """Return a SpotifyProvider (bypassing __init__) with the given setup_data."""
289    prov = object.__new__(SpotifyProvider)
290    config = MagicMock(instance_id="spotify--test")
291    config.get_value = MagicMock(return_value=None)
292    config.values = {}
293    prov.config = config
294    prov.manifest = MagicMock(domain="spotify")
295    prov.logger = MagicMock()
296    prov.available = True
297    mass = MagicMock()
298    mass.storage_path = str(tmp_path / "storage")
299    mass.cache_path = str(tmp_path / "cache")
300    # get_setup_value reads the live setup_data blob from the store
301    mass.config.get = MagicMock(return_value=setup_data or {})
302    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
303    # the store keeps values encrypted; decrypt is an identity map for the test
304    mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
305    prov.mass = mass
306    return prov
307
308
309def _make_backend(tmp_path: Path, setup_data: dict[str, Any] | None = None) -> SoloistBackend:
310    """Return a SoloistBackend on a mocked provider."""
311    return SoloistBackend(_make_provider(tmp_path, setup_data))
312
313
314def _streamdetails_for(
315    *,
316    queue_id: str | None = "player1",
317    uri: str = TRACK_A,
318    media_type: MediaType = MediaType.TRACK,
319) -> StreamDetails:
320    """Return stream details for a Spotify item served by the test instance."""
321    return StreamDetails(
322        provider="spotify--test",
323        item_id=uri.rsplit(":", 1)[1],
324        audio_format=AudioFormat(content_type=ContentType.PCM_S16LE),
325        media_type=media_type,
326        queue_id=queue_id,
327    )
328
329
330def _install_fake_binary_manager(monkeypatch: pytest.MonkeyPatch) -> None:
331    """Replace the shared binary manager so no download or exec is attempted."""
332    manager = MagicMock()
333    manager.ensure_fresh = AsyncMock(return_value=Path("/nonexistent/soloist"))
334    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", MagicMock(return_value=manager))
335
336
337def _make_run(
338    tmp_path: Path,
339    uri: str = TRACK_A,
340    seek_ms: int = 0,
341    duration: int | None = 180,
342    media_key: str | None = None,
343) -> _SingleTrackRun:
344    """Return a run with its process/sink/client replaced by mocks."""
345    streamdetails = _streamdetails_for(uri=uri)
346    if duration is not None:
347        streamdetails.duration = duration
348    run = _SingleTrackRun(_make_backend(tmp_path), uri, seek_ms, streamdetails)
349    if media_key is not None:
350        run.media_key = media_key
351    run._sink = AsyncMock()
352    run._client = AsyncMock()
353    run._proc = MagicMock(returncode=None)
354    run._logged_in = True
355    run._sink_running = True
356    return run
357
358
359async def _collect(run: _SingleTrackRun) -> bytes:
360    """Return everything the run streams, up to where it ended the item."""
361    collected = bytearray()
362    async for chunk in run.stream():
363        collected.extend(chunk)
364    return bytes(collected)
365
366
367def test_scrub_leaves_mid_track_silence_alone(tmp_path: Path) -> None:
368    """A quiet passage outside the tail zone is content (the shaper owns the lead)."""
369    run = _make_run(tmp_path)
370    run._read_bytes = 10 * _BYTES_PER_SECOND
371    assert run._scrub(b"\x00" * 1024) == b"\x00" * 1024
372    assert run._scrub(b"\x01" * 64) == b"\x01" * 64
373
374
375def test_scrub_refuses_padding_in_the_items_tail_zone(tmp_path: Path) -> None:
376    """Zeros inside the tail zone are the sink idling out the engine's end."""
377    run = _make_run(tmp_path, duration=60)
378    second = _BYTES_PER_SECOND
379    run._read_bytes = 55 * second
380    grace = int(_TAIL_PAD_GRACE_S * second)
381    # the first moment of padding is kept, the rest refused
382    assert run._scrub(b"\x00" * grace) == b"\x00" * grace
383    run._tail_zeros = grace
384    assert run._scrub(b"\x00" * second) == b""
385    # real audio resets the run: the zeros were a quiet passage after all
386    assert run._scrub(b"\x01" * 64) == b"\x01" * 64
387    assert run._tail_zeros == 0
388
389
390def test_scrub_leaves_a_short_items_silence_alone(tmp_path: Path) -> None:
391    """An item no longer than the zone has no distinguishable tail."""
392    run = _make_run(tmp_path, duration=int(_TAIL_PAD_ZONE_S))
393    run._read_bytes = int(_TAIL_PAD_ZONE_S - 1) * _BYTES_PER_SECOND
394    chunk = b"\x00" * (2 * int(_TAIL_PAD_GRACE_S * _BYTES_PER_SECOND))
395    assert run._scrub(chunk) == chunk
396
397
398async def test_a_full_cushion_pauses_the_engine(tmp_path: Path) -> None:
399    """When the consumer stops taking audio, the sink is suspended, not overflowed."""
400    run = _make_run(tmp_path)
401    sink = cast("AsyncMock", run._sink)
402    while not run._chunks.full():
403        run._chunks.put_nowait(b"\x01")
404
405    blocked = asyncio.ensure_future(run._hand_over(b"\x02"))
406    await asyncio.sleep(0.01)
407    assert not blocked.done()
408    sink.suspend.assert_awaited_once()
409
410    # the consumer takes a chunk: the write lands and the engine resumes
411    run._engine_playing = True
412    assert run._chunks.get_nowait() == b"\x01"
413    assert await blocked is True
414    sink.resume.assert_awaited()
415
416
417async def test_a_full_cushion_still_ends_the_stream(tmp_path: Path) -> None:
418    """The end of delivery survives a cushion with no room left for the sentinel."""
419    # no duration: this covers the cushion, not how much audio arrived
420    run = _make_run(tmp_path, duration=None)
421    while not run._chunks.full():
422        run._chunks.put_nowait(b"\x01" * 64)
423    run._finish_delivery()
424    # nothing may re-signal the end once the consumer drains: the flag carries it
425    assert len(await _collect(run)) == run._chunks.maxsize * 64
426
427
428async def test_a_failed_run_surfaces_its_error_to_the_stream(tmp_path: Path) -> None:
429    """The consumer sees the run's real failure, not a clean end."""
430    run = _make_run(tmp_path)
431    run._chunks.put_nowait(b"\x01" * 64)
432    run._fail("the engine broke")
433    with pytest.raises(AudioError, match="the engine broke"):
434        await _collect(run)
435
436
437async def test_a_run_that_delivered_nothing_is_rejected(tmp_path: Path) -> None:
438    """A run still going that rendered nothing must not read as a completed stream."""
439    run = _make_run(tmp_path, duration=152)
440    run._chunks.put_nowait(b"\x01" * _FRAME_BYTES)
441    run._finish_delivery()
442    with pytest.raises(AudioError, match="stopped before it played"):
443        await _collect(run)
444
445
446async def test_a_refused_item_is_reported_as_a_refusal(tmp_path: Path) -> None:
447    """An engine that plays nothing and ends the run cleanly was refused the item."""
448    run = _make_run(tmp_path, duration=152)
449    run._engine_exited = True
450    run._proc = MagicMock(returncode=0)
451    run._chunks.put_nowait(b"\x01" * _FRAME_BYTES)
452    run._finish_delivery()
453    # the message travels on the error itself; the queue reports it where it skips
454    with pytest.raises(AudioError, match="would not play this track"):
455        await _collect(run)
456
457
458async def test_a_crashed_run_is_not_reported_as_a_refusal(tmp_path: Path) -> None:
459    """An engine that died on its own item is a fault, whatever it managed to play."""
460    run = _make_run(tmp_path, duration=152)
461    run._engine_exited = True
462    run._proc = MagicMock(returncode=1)
463    run._chunks.put_nowait(b"\x01" * _FRAME_BYTES)
464    run._finish_delivery()
465    # names the code, so a refusal that turns out to exit non-zero is recognisable
466    with pytest.raises(AudioError, match="engine exit code 1"):
467        await _collect(run)
468
469
470async def test_audio_that_arrived_and_stopped_is_not_a_failure(tmp_path: Path) -> None:
471    """Something ended the item early - a skip, a seek, the account playing elsewhere."""
472    run = _make_run(tmp_path, duration=152)
473    run._engine_exited = True
474    run._proc = MagicMock(returncode=0)
475    run._chunks.put_nowait(b"\x01" * (100 * _BYTES_PER_SECOND))
476    run._finish_delivery()
477    await _collect(run)
478
479
480async def test_a_crash_part_way_through_is_reported(tmp_path: Path) -> None:
481    """Audio arrived, but the engine died on the item rather than ending it."""
482    run = _make_run(tmp_path, duration=152)
483    run._engine_exited = True
484    run._proc = MagicMock(returncode=1)
485    run._chunks.put_nowait(b"\x01" * (30 * _BYTES_PER_SECOND))
486    run._finish_delivery()
487    with pytest.raises(AudioError, match="stopped unexpectedly"):
488        await _collect(run)
489
490
491async def test_a_brief_item_that_played_nothing_is_still_reported(tmp_path: Path) -> None:
492    """Accepting short items wholesale would let a refused one through unnoticed."""
493    run = _make_run(tmp_path, duration=2)
494    run._duration_ms = 1200
495    run._engine_exited = True
496    run._proc = MagicMock(returncode=0)
497    run._chunks.put_nowait(b"\x01" * _FRAME_BYTES)
498    run._finish_delivery()
499    with pytest.raises(AudioError, match="would not play this track"):
500        await _collect(run)
501
502
503async def test_a_brief_item_played_in_full_is_accepted(tmp_path: Path) -> None:
504    """The lead trim takes its cut off a complete delivery of a very short item."""
505    run = _make_run(tmp_path, duration=2)
506    run._duration_ms = 1200
507    run._engine_exited = True
508    run._proc = MagicMock(returncode=0)
509    # what a complete 1.2s item arrives as once the trim has taken its full budget
510    run._chunks.put_nowait(b"\x01" * (7 * _BYTES_PER_SECOND // 10))
511    run._finish_delivery()
512    await _collect(run)
513
514
515async def test_a_short_item_that_played_nothing_is_still_reported(tmp_path: Path) -> None:
516    """A short item is judged the same way: nothing arrived, so nothing played."""
517    run = _make_run(tmp_path, duration=8)
518    run._engine_exited = True
519    run._proc = MagicMock(returncode=0)
520    run._chunks.put_nowait(b"\x01" * _FRAME_BYTES)
521    run._finish_delivery()
522    with pytest.raises(AudioError, match="would not play this track"):
523        await _collect(run)
524
525
526async def test_a_short_item_played_in_full_is_accepted(tmp_path: Path) -> None:
527    """A complete short item must not be mistaken for one that never played."""
528    run = _make_run(tmp_path, duration=8)
529    run._engine_exited = True
530    run._proc = MagicMock(returncode=0)
531    run._chunks.put_nowait(b"\x01" * (8 * _BYTES_PER_SECOND))
532    run._finish_delivery()
533    await _collect(run)
534
535
536async def test_a_seek_to_the_items_end_is_not_read_as_a_refusal(tmp_path: Path) -> None:
537    """Audio the seek skipped counts, so a near-complete delivery ends cleanly."""
538    run = _make_run(tmp_path, duration=152, seek_ms=151_000)
539    run._engine_exited = True
540    run._proc = MagicMock(returncode=0)
541    run._chunks.put_nowait(b"\x01" * _FRAME_BYTES)
542    run._finish_delivery()
543    await _collect(run)
544
545
546async def test_a_stopped_run_is_not_judged_incomplete(tmp_path: Path) -> None:
547    """A consumer that left early is the normal end of an aborted stream."""
548    run = _make_run(tmp_path, duration=152)
549    run._chunks.put_nowait(b"\x01" * _FRAME_BYTES)
550    run._stopped = True
551    run._finish_delivery()
552    assert await _collect(run) == b"\x01" * _FRAME_BYTES
553
554
555async def test_a_seek_counts_towards_the_delivery(tmp_path: Path) -> None:
556    """Audio skipped by the seek is not audio the engine failed to deliver."""
557    run = _make_run(tmp_path, duration=60, seek_ms=55_000)
558    run._chunks.put_nowait(b"\x01" * (6 * _BYTES_PER_SECOND))
559    run._finish_delivery()
560    await _collect(run)
561
562
563def test_the_own_item_report_starts_the_run_and_refines_the_duration(tmp_path: Path) -> None:
564    """The engine reaching the item is what playback start means."""
565    run = _make_run(tmp_path, duration=180)
566    run._observe_item(TRACK_A, 179_000)
567    assert run._started.is_set()
568    assert run._duration_ms == 179_000
569
570
571def test_a_seek_is_confirmed_only_near_its_target(tmp_path: Path) -> None:
572    """A pre-seek position report cannot confirm the seek."""
573    run = _make_run(tmp_path, seek_ms=60_000)
574    run._observe_position(0)
575    assert not run._seek_confirmed.is_set()
576    run._observe_position(58_000)
577    assert run._seek_confirmed.is_set()
578
579
580def test_single_track_args_carry_the_uri(tmp_path: Path) -> None:
581    """The engine is spawned on exactly one URI, in single-track mode."""
582    backend = _make_backend(tmp_path, {CONF_SOLOIST_API_KEY: "k" * 20})
583    backend._binary = tmp_path / "soloist-bin"
584    args = backend._session_args(TRACK_A)
585    assert "--single-track" in args
586    assert args[args.index("--single-track") + 1] == TRACK_A
587    # the binary refuses to start without a device name, even though
588    # single-track mode never advertises one
589    assert "--device-name" in args
590
591
592async def test_a_run_for_another_item_reports_capacity(tmp_path: Path) -> None:
593    """A live run is one stream slot: anything else waits or resolves elsewhere."""
594    backend = _make_backend(tmp_path)
595    backend._run = _make_run(tmp_path, uri=TRACK_A, media_key=TRACK_A)
596    with pytest.raises(ProviderStreamLimitError):
597        await backend._acquire_run(TRACK_B, 0, _streamdetails_for(uri=TRACK_B), continuation=False)
598
599
600async def test_a_replaced_streams_continuation_is_superseded(tmp_path: Path) -> None:
601    """A continuation must not take the run back from the stream that replaced it."""
602    backend = _make_backend(tmp_path)
603    streamdetails = _streamdetails_for(uri=AUDIOBOOK, media_type=MediaType.AUDIOBOOK)
604    run = _make_run(tmp_path, uri=CHAPTER_B, media_key=streamdetails.uri)
605    backend._run = run
606    with pytest.raises(StreamSupersededError):
607        await backend._acquire_run(CHAPTER_A, 0, streamdetails, continuation=True)
608
609
610def test_session_normalizes_answers_only_for_the_items_own_run(tmp_path: Path) -> None:
611    """Another item's run says nothing about this one."""
612    backend = _make_backend(tmp_path)
613    run = _make_run(tmp_path, uri=TRACK_A, media_key=_streamdetails_for(uri=TRACK_A).uri)
614    run.engine_normalizes = True
615    backend._run = run
616    assert backend.session_normalizes(_streamdetails_for(uri=TRACK_A)) is True
617    assert backend.session_normalizes(_streamdetails_for(uri=TRACK_B)) is None
618
619
620async def test_the_engine_wandering_on_after_delivery_ends_the_run_cleanly(
621    tmp_path: Path,
622) -> None:
623    """Autoplay reaching the next track right before exit is this item's natural end."""
624    run = _make_run(tmp_path, duration=60)
625    run._observe_item(TRACK_A, 60_000)
626    run.mass.create_task = MagicMock()  # type: ignore[method-assign]
627    run._chunks.put_nowait(b"\x01" * (60 * _BYTES_PER_SECOND))
628    run._observe_item(TRACK_B, 100_000)
629    assert run._error is None
630    assert run._item_over is True
631    assert await _collect(run) == b"\x01" * (60 * _BYTES_PER_SECOND)
632    run.mass.create_task.assert_called_once()
633
634
635def test_losing_the_device_mid_item_is_reported_as_a_takeover(tmp_path: Path) -> None:
636    """Another player taking the account is what ends this item, and it must say so."""
637    run = _make_run(tmp_path)
638    run.mass.create_task = MagicMock()  # type: ignore[method-assign]
639    run._started.set()
640    run._observe_device_active(active=False)
641    assert run._error is not None
642    assert "already streaming somewhere else" in run._error
643
644
645def test_the_device_report_at_startup_is_not_a_takeover(tmp_path: Path) -> None:
646    """The engine reports itself inactive while a run is still starting up."""
647    run = _make_run(tmp_path)
648    run.mass.create_task = MagicMock()  # type: ignore[method-assign]
649    run._observe_device_active(active=False)
650    assert run._error is None
651
652
653def test_losing_the_device_once_the_daemon_has_gone_is_not_a_takeover(tmp_path: Path) -> None:
654    """A run that ended on its own sheds the device as it goes; the item still played."""
655    run = _make_run(tmp_path)
656    run.mass.create_task = MagicMock()  # type: ignore[method-assign]
657    run._started.set()
658    run._engine_exited = True
659    run._observe_device_active(active=False)
660    assert run._error is None
661
662
663def test_losing_the_device_after_the_item_is_over_is_not_a_takeover(tmp_path: Path) -> None:
664    """The daemon drops the device on its way out; the item already played."""
665    run = _make_run(tmp_path)
666    run.mass.create_task = MagicMock()  # type: ignore[method-assign]
667    run._started.set()
668    run._item_over = True
669    run._observe_device_active(active=False)
670    assert run._error is None
671
672
673def test_the_engine_starting_on_the_wrong_item_fails_the_run(tmp_path: Path) -> None:
674    """Before this run's item ever played, a foreign report is not an ending."""
675    run = _make_run(tmp_path)
676    run.mass.create_task = MagicMock()  # type: ignore[method-assign]
677    run._observe_item(TRACK_B, 100_000)
678    assert run._error is not None
679
680
681async def test_a_seek_replaces_the_held_run(tmp_path: Path) -> None:
682    """A positive seek restarts the item's run; only a prefetch must never steal it."""
683    backend = _make_backend(tmp_path, {CONF_SOLOIST_API_KEY: "k" * 20})
684    backend._server = MagicMock()
685    backend._binary = tmp_path / "soloist-bin"
686    held = _make_run(tmp_path, uri=TRACK_A, media_key=_streamdetails_for(uri=TRACK_A).uri)
687    held.stop = AsyncMock()  # type: ignore[method-assign]
688    backend._run = held
689
690    with (
691        patch(
692            "music_assistant.providers.spotify.backends.soloist.SoloistBinaryManager.ensure_fresh",
693            AsyncMock(),
694        ),
695        patch.object(soloist_backend._SingleTrackRun, "start", AsyncMock()),
696    ):
697        run = await backend._acquire_run(
698            TRACK_A, 30, _streamdetails_for(uri=TRACK_A), continuation=False
699        )
700    held.stop.assert_awaited_once()
701    assert run is not held
702    assert backend._run is run
703