music-assistant-server

1"""Tests for the Spotify Soloist backend (all fakes: no processes, network or pulse)."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import stat
8from contextlib import suppress
9from functools import partial
10from pathlib import Path
11from types import SimpleNamespace
12from typing import TYPE_CHECKING, Any
13from unittest.mock import AsyncMock, MagicMock
14
15import pytest
16from music_assistant_models.enums import ContentType, RepeatMode, StreamType
17from music_assistant_models.errors import AudioError
18
19from music_assistant.providers.spotify_connect.base import (
20    AUDIO_QUALITY_HIGH,
21    AUDIO_QUALITY_LOSSLESS,
22    AUDIO_QUALITY_NORMAL,
23    AUDIO_QUALITY_VERY_HIGH,
24)
25from music_assistant.providers.spotify_connect.models import (
26    BackendEvent,
27    BackendEventType,
28    QueueEntrySource,
29)
30from music_assistant.providers.spotify_connect.soloist import backend as soloist_backend
31from music_assistant.providers.spotify_connect.soloist.backend import (
32    CACHE_SIZE_MB,
33    VOLUME_MODE_PLAYER_ONLY,
34    VOLUME_MODE_SYNC_SPOTIFY,
35    SoloistBackend,
36)
37from music_assistant.providers.spotify_connect.soloist.runtime import (
38    BuildExpiredError,
39    ConsentRequiredError,
40    SoloistAuthState,
41    SoloistCommandResult,
42    SoloistDeviceChanged,
43    SoloistEntity,
44    SoloistErrorMessage,
45    SoloistEvent,
46    SoloistOptionsChanged,
47    SoloistPlaybackOptions,
48    SoloistPlaybackState,
49    SoloistPosition,
50    SoloistPositionSync,
51    SoloistQueueChanged,
52    SoloistQueueEntry,
53    SoloistTrackChanged,
54    SoloistVolumeChanged,
55)
56
57if TYPE_CHECKING:
58    from collections.abc import AsyncGenerator, Callable
59
60_API_KEY = "sk-super-secret-key-123"
61_IDENTITY_KEY = "spotify_connect_player1"
62
63
64class _FakeServer:
65    """Fake PulseCaptureServer with a settable generation."""
66
67    def __init__(self) -> None:
68        self.generation = 1
69        self.released = 0
70
71    async def acquire(self) -> _FakeServer:
72        """Return self, like the real refcounted acquire."""
73        return self
74
75    async def release(self) -> None:
76        """Record the release."""
77        self.released += 1
78
79    def child_env(self, sink_name: str) -> dict[str, str]:
80        """Return a minimal audio-client environment."""
81        return {"PULSE_SERVER": "unix:/fake/native", "PULSE_SINK": sink_name}
82
83
84class _FakeSink:
85    """Fake PipeSink recording volume changes and unloads."""
86
87    def __init__(self, name: str = "sink1") -> None:
88        self.sink_name = name
89        self.fifo_path = Path(f"/fake/{name}.pcm")
90        self.volumes: list[float] = []
91        self.unloaded = 0
92        # when set (and not yet signalled), set_volume blocks on this gate
93        self.gate: asyncio.Event | None = None
94        # when set, set_volume raises this instead of recording the change
95        self.set_volume_error: Exception | None = None
96
97    async def set_volume(self, volume_pct: float) -> None:
98        """Record a volume change, optionally blocking on the gate first."""
99        if self.gate is not None:
100            await self.gate.wait()
101        if self.set_volume_error is not None:
102            raise self.set_volume_error
103        self.volumes.append(volume_pct)
104
105    async def unload(self) -> None:
106        """Record the unload."""
107        self.unloaded += 1
108
109
110class _FakeProc:
111    """Stand-in for AsyncProcess recording its lifecycle."""
112
113    def __init__(
114        self,
115        exit_code: int = 0,
116        start_error: Exception | None = None,
117        *,
118        stdout_lines: list[str] | None = None,
119        block_stdout: bool = False,
120        on_close: Callable[[], None] | None = None,
121    ) -> None:
122        self.returncode: int | None = None
123        self.closed = 0
124        self._exit_code = exit_code
125        self._start_error = start_error
126        self._stdout_lines = stdout_lines or []
127        self._block_stdout = block_stdout
128        self._on_close = on_close
129        self._closed_event = asyncio.Event()
130        # a real daemon's stdout reaches EOF exactly when the process exits,
131        # so the fake ties its wait() to its output ending
132        self._output_done = asyncio.Event()
133
134    async def start(self) -> None:
135        """Start the fake process, failing when configured to."""
136        if self._start_error is not None:
137            raise self._start_error
138
139    async def close(self) -> None:
140        """Mark the process closed and expose its exit code."""
141        self.closed += 1
142        self.returncode = self._exit_code
143        self._closed_event.set()
144        if self._on_close is not None:
145            self._on_close()
146
147    async def iter_stdout(self) -> AsyncGenerator[str]:
148        """Yield the configured stdout lines, optionally blocking until closed."""
149        try:
150            for line in self._stdout_lines:
151                yield line
152            if self._block_stdout:
153                await self._closed_event.wait()
154        finally:
155            self._output_done.set()
156
157    async def wait(self) -> int:
158        """Return the exit code once the fake daemon's output ended."""
159        await self._output_done.wait()
160        return self._exit_code
161
162
163def _make_backend(
164    *,
165    volume_mode: str = VOLUME_MODE_PLAYER_ONLY,
166    identity_key: str = _IDENTITY_KEY,
167    base_dir: Path | None = None,
168    consent: bool = True,
169    audio_quality: str = AUDIO_QUALITY_LOSSLESS,
170) -> tuple[SoloistBackend, list[BackendEvent]]:
171    """Build a backend on a mocked mass, capturing every emitted BackendEvent."""
172    mass = MagicMock()
173    mass.storage_path = str(base_dir / "storage") if base_dir else "/fake/storage"
174    mass.cache_path = str(base_dir / "cache") if base_dir else "/fake/cache"
175    events: list[BackendEvent] = []
176
177    async def _capture(event: BackendEvent) -> None:
178        events.append(event)
179
180    backend = SoloistBackend(
181        mass,
182        identity_key=identity_key,
183        publish_name="Test Device",
184        name="Spotify Test",
185        logger=logging.getLogger("test.soloist_backend"),
186        event_callback=_capture,
187        api_key=_API_KEY,
188        consent=consent,
189        volume_mode=volume_mode,
190        audio_quality=audio_quality,
191    )
192    return backend, events
193
194
195def _runner_backend(
196    *, volume_mode: str = VOLUME_MODE_PLAYER_ONLY
197) -> tuple[SoloistBackend, list[BackendEvent]]:
198    """Build a backend primed to run its daemon supervisor with fakes."""
199    backend, events = _make_backend(volume_mode=volume_mode)
200    backend._binary = Path("/fake/bin/soloist")
201    server: Any = _FakeServer()
202    sink: Any = _FakeSink()
203    backend._server = server
204    backend._sink = sink
205    backend._sink_generation = server.generation
206    return backend, events
207
208
209def _patch_spawn(
210    monkeypatch: pytest.MonkeyPatch, procs: list[_FakeProc]
211) -> list[tuple[list[str], dict[str, Any]]]:
212    """Replace AsyncProcess with a factory serving the given fakes, recording spawns."""
213    spawned: list[tuple[list[str], dict[str, Any]]] = []
214
215    def _spawn(args: list[str], **kwargs: Any) -> _FakeProc:
216        spawned.append((args, kwargs))
217        return procs[len(spawned) - 1]
218
219    monkeypatch.setattr(soloist_backend, "AsyncProcess", _spawn)
220    return spawned
221
222
223def _event(event_type: str, data: Any) -> SoloistEvent:
224    """Wrap a decoded payload in a SoloistEvent."""
225    return SoloistEvent(type=event_type, data=data, raw={"type": event_type})
226
227
228def _volume_event(volume: int) -> SoloistEvent:
229    """Wrap a volume value in a decoded volume_changed event."""
230    return _event("volume_changed", SoloistVolumeChanged(volume=volume))
231
232
233async def test_start_wires_binary_capture_and_supervisors(
234    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
235) -> None:
236    """start() installs the binary, acquires capture and launches both supervisors."""
237    backend, _events = _make_backend(base_dir=tmp_path)
238    tasks: list[str] = []
239
240    def _fake_create_task(coro: Any) -> MagicMock:
241        tasks.append(coro.__name__)
242        coro.close()
243        return MagicMock()
244
245    mass_mock: Any = backend.mass
246    mass_mock.create_task.side_effect = _fake_create_task
247    consents: list[bool] = []
248
249    class _FakeManager:
250        """Fake binary manager recording the consent flag."""
251
252        def __init__(self, mass: Any) -> None:
253            """Accept the mass argument like the real manager."""
254
255        def diagnostics(self) -> dict[str, Any]:
256            """Report the installed build's digest."""
257            return {"installed": True, "sha256": "sha-1"}
258
259        async def ensure_fresh(self, consent: bool) -> Path:
260            """Record the consent flag and hand out a fake binary path."""
261            consents.append(consent)
262            return Path("/fake/bin/soloist")
263
264    server: Any = _FakeServer()
265    sink: Any = _FakeSink()
266    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", _FakeManager)
267    monkeypatch.setattr(soloist_backend, "get_pulse_capture_server", lambda _mass: server)
268    monkeypatch.setattr(
269        soloist_backend, "PipeSink", SimpleNamespace(create=AsyncMock(return_value=sink))
270    )
271
272    await backend.start()
273
274    assert consents == [True]
275    assert backend._binary == Path("/fake/bin/soloist")
276    assert backend._sink_generation == server.generation
277    assert backend._server is server
278    assert backend._sink is sink
279    assert backend._client is not None
280    assert backend._client.data_dir == backend._data_dir
281    assert backend._data_dir.is_dir()
282    # the data dir holds the Spotify device identity/session: owner-only
283    assert stat.S_IMODE(backend._data_dir.stat().st_mode) == 0o700
284    assert backend._cache_dir.is_dir()
285    assert tasks == [
286        "_daemon_runner",
287        "_events_runner",
288        "_binary_refresh_loop",
289        "_generation_watcher",
290    ]
291
292
293async def test_start_setup_errors_propagate(monkeypatch: pytest.MonkeyPatch) -> None:
294    """A binary setup error fails start() before any capture resource is acquired."""
295    backend, _events = _make_backend(consent=False)
296
297    class _RefusingManager:
298        """Fake binary manager that refuses without consent."""
299
300        def __init__(self, mass: Any) -> None:
301            """Accept the mass argument like the real manager."""
302
303        async def ensure_fresh(self, consent: bool) -> Path:
304            """Refuse the download."""
305            raise ConsentRequiredError("consent required")
306
307    capture = MagicMock()
308    capture.acquire = AsyncMock()
309    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", _RefusingManager)
310    monkeypatch.setattr(soloist_backend, "get_pulse_capture_server", lambda _mass: capture)
311
312    with pytest.raises(ConsentRequiredError):
313        await backend.start()
314    capture.acquire.assert_not_awaited()
315
316
317async def test_daemon_argv_and_key_never_logged(
318    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
319) -> None:
320    """The daemon argv carries the exact documented flags; the api key never hits a log."""
321    backend, _events = _runner_backend()
322    # a single supervisor iteration: stop once the spawned process is closed
323    proc = _FakeProc(on_close=lambda: setattr(backend, "_stop_called", True))
324    spawned = _patch_spawn(monkeypatch, [proc])
325
326    with caplog.at_level(logging.DEBUG):
327        await backend._daemon_runner()
328
329    args, kwargs = spawned[0]
330    assert args == [
331        "/fake/bin/soloist",
332        "--device-name",
333        "Test Device",
334        "--api-key",
335        _API_KEY,
336        "--data-dir",
337        f"/fake/storage/spotify_connect/{_IDENTITY_KEY}/soloist-data",
338        "--cache-dir",
339        f"/fake/cache/{_IDENTITY_KEY}/soloist-cache",
340        "--cache-size",
341        str(CACHE_SIZE_MB),
342        "--initial-volume",
343        "100",
344        "--ws",
345        "127.0.0.1:0",
346    ]
347    assert kwargs["name"] == "soloist[Spotify Test]"
348    # the daemon logs to stdout; stderr is merged in so nothing is left uncaptured
349    assert kwargs["stdout"] is True
350    assert kwargs["stderr"] is asyncio.subprocess.STDOUT
351    assert kwargs["env"] == {"PULSE_SERVER": "unix:/fake/native", "PULSE_SINK": "sink1"}
352    assert all(_API_KEY not in record.getMessage() for record in caplog.records)
353
354
355async def test_exit_code_10_refreshes_binary_before_restart(
356    monkeypatch: pytest.MonkeyPatch,
357) -> None:
358    """A build-expired exit (code 10) refreshes the binary and restarts with it."""
359    backend, _events = _runner_backend()
360    monkeypatch.setattr(soloist_backend, "RESTART_DELAY_S", 0)
361    refreshed: list[bool] = []
362
363    class _FakeManager:
364        """Fake binary manager serving a replacement build."""
365
366        def __init__(self, mass: Any) -> None:
367            """Accept the mass argument like the real manager."""
368
369        def diagnostics(self) -> dict[str, Any]:
370            """Report the replacement build's digest."""
371            return {"installed": True, "sha256": "sha-v2"}
372
373        async def ensure_fresh(self, consent: bool, *, force: bool = False) -> Path:
374            """Serve the replacement build."""
375            refreshed.append(force)
376            return Path("/fake/bin/soloist-v2")
377
378    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", _FakeManager)
379    # stop the supervisor once the restarted (second) daemon has run
380    spawned = _patch_spawn(
381        monkeypatch,
382        [
383            _FakeProc(exit_code=10),
384            _FakeProc(exit_code=0, on_close=lambda: setattr(backend, "_stop_called", True)),
385        ],
386    )
387
388    await backend._daemon_runner()
389
390    assert refreshed == [True]
391    assert len(spawned) == 2
392    # the restarted daemon runs the freshly installed binary
393    assert spawned[1][0][0] == "/fake/bin/soloist-v2"
394
395
396async def test_exit_code_10_with_failed_refresh_is_fatal(
397    monkeypatch: pytest.MonkeyPatch,
398) -> None:
399    """When no replacement build exists for an expired one, the backend fails fatally."""
400    backend, events = _runner_backend()
401    monkeypatch.setattr(soloist_backend, "RESTART_DELAY_S", 0)
402
403    class _ExpiredManager:
404        """Fake binary manager that cannot replace the expired build."""
405
406        def __init__(self, mass: Any) -> None:
407            """Accept the mass argument like the real manager."""
408
409        async def ensure_fresh(self, consent: bool, *, force: bool = False) -> Path:
410            """Fail the refresh."""
411            raise BuildExpiredError("expired")
412
413    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", _ExpiredManager)
414    spawned = _patch_spawn(monkeypatch, [_FakeProc(exit_code=10)])
415
416    await backend._daemon_runner()
417
418    assert len(spawned) == 1  # the expired build is never restarted
419    assert events[-1].type is BackendEventType.FATAL_ERROR
420    assert "expired" in (events[-1].error or "")
421    # the shared binary expiring hits every daemon alike
422    assert events[-1].provider_wide is True
423
424
425async def test_five_daemon_failures_report_fatal_error(
426    monkeypatch: pytest.MonkeyPatch,
427) -> None:
428    """After five consecutive daemon failures the backend reports a fatal error."""
429    backend, events = _runner_backend()
430    monkeypatch.setattr(soloist_backend, "RESTART_DELAY_S", 0)
431    procs = [_FakeProc(exit_code=1, start_error=RuntimeError("spawn failed")) for _ in range(5)]
432    spawned = _patch_spawn(monkeypatch, procs)
433
434    await backend._daemon_runner()
435
436    assert len(spawned) == 5
437    assert sum(1 for e in events if e.type is BackendEventType.CONNECTION_LOST) == 5
438    assert events[-1].type is BackendEventType.FATAL_ERROR
439    # soloist fatals take the whole provider down (engine-level failure)
440    assert events[-1].provider_wide is True
441
442
443@pytest.mark.parametrize(
444    ("data", "expected_type"),
445    [
446        pytest.param(
447            SoloistAuthState(logged_in=False, is_active=False),
448            BackendEventType.SESSION_INACTIVE,
449            id="auth_state-logged_out",
450        ),
451        pytest.param(
452            SoloistAuthState(logged_in=True, is_active=True),
453            BackendEventType.SESSION_ACTIVE,
454            id="auth_state-active",
455        ),
456        pytest.param(
457            SoloistAuthState(logged_in=True, is_active=False),
458            BackendEventType.SESSION_INACTIVE,
459            id="auth_state-inactive",
460        ),
461        pytest.param(
462            SoloistDeviceChanged(is_active=True),
463            BackendEventType.SESSION_ACTIVE,
464            id="device_changed-active",
465        ),
466        pytest.param(
467            SoloistDeviceChanged(is_active=False),
468            BackendEventType.SESSION_INACTIVE,
469            id="device_changed-inactive",
470        ),
471        pytest.param(
472            SoloistPlaybackState(status="playing"), BackendEventType.PLAYING, id="status-playing"
473        ),
474        pytest.param(
475            SoloistPlaybackState(status="paused"), BackendEventType.PAUSED, id="status-paused"
476        ),
477        pytest.param(
478            SoloistPlaybackState(status="buffering"),
479            BackendEventType.BUFFERING,
480            id="status-buffering",
481        ),
482        pytest.param(
483            SoloistPlaybackState(status="idle"), BackendEventType.STOPPED, id="status-idle"
484        ),
485        pytest.param(
486            SoloistPlaybackState(status="stopped"), BackendEventType.STOPPED, id="status-stopped"
487        ),
488        pytest.param(
489            SoloistPlaybackState(status="warping"), BackendEventType.OTHER, id="status-unknown"
490        ),
491        pytest.param(
492            SoloistErrorMessage(message="boom"), BackendEventType.ERROR, id="error-message"
493        ),
494        pytest.param(
495            SoloistOptionsChanged(options=SoloistPlaybackOptions()),
496            BackendEventType.OPTIONS_CHANGED,
497            id="options_changed",
498        ),
499        pytest.param(SoloistQueueChanged(), BackendEventType.QUEUE_CHANGED, id="queue_changed"),
500        pytest.param(
501            SoloistCommandResult(command="pause"), BackendEventType.OTHER, id="command_result"
502        ),
503        pytest.param(None, BackendEventType.OTHER, id="unknown-event"),
504    ],
505)
506async def test_event_adaptation(data: Any, expected_type: BackendEventType) -> None:
507    """Every documented soloist event maps onto its normalized counterpart."""
508    backend, events = _make_backend()
509
510    await backend._handle_event(_event("test_event", data))
511
512    assert [event.type for event in events] == [expected_type]
513
514
515async def test_account_takeover_is_a_session_change_not_an_auth_loss() -> None:
516    """Another account claiming the device reports sessions ending and starting, no error."""
517    backend, events = _make_backend()
518
519    # a fresh daemon advertising for pairing, then one account, then a takeover
520    # by another: the daemon signs the first one out and the second one in
521    for logged_in, is_active in ((False, False), (True, True), (False, False), (True, True)):
522        await backend._handle_event(
523            _event("auth_state", SoloistAuthState(logged_in=logged_in, is_active=is_active))
524        )
525
526    assert [event.type for event in events] == [
527        BackendEventType.SESSION_INACTIVE,
528        BackendEventType.SESSION_ACTIVE,
529        BackendEventType.SESSION_INACTIVE,
530        BackendEventType.SESSION_ACTIVE,
531    ]
532
533
534async def test_error_event_carries_message() -> None:
535    """An error event forwards the daemon's message on the normalized event."""
536    backend, events = _make_backend()
537
538    await backend._handle_event(
539        _event("error", SoloistErrorMessage(message="command requires authentication"))
540    )
541
542    assert events[0].type is BackendEventType.ERROR
543    assert events[0].error == "command requires authentication"
544
545
546async def test_track_changed_maps_decorations() -> None:
547    """A track_changed event maps the entity decorations onto normalized metadata."""
548    backend, events = _make_backend()
549    item = SoloistEntity(
550        uri="spotify:track:t1",
551        entity_type="track",
552        # shape captured from a real soloist 1.3.7 playback_state payload
553        decorations={
554            "identity": {"name": "My Song"},
555            "playback": {"duration_ms": 210999, "content_ratings": []},
556            "creators": [
557                {
558                    "entity": {
559                        "uri": "spotify:artist:a1",
560                        "entity_type": "artist",
561                        "decorations": {"identity": {"name": "Main Artist"}},
562                    }
563                },
564                {
565                    "entity": {
566                        "uri": "spotify:artist:a2",
567                        "entity_type": "artist",
568                        "decorations": {"identity": {"name": "Feat Artist"}},
569                    }
570                },
571            ],
572            "parent": {
573                "entity": {
574                    "uri": "spotify:album:al1",
575                    "entity_type": "album",
576                    "decorations": {"identity": {"name": "The Album"}},
577                }
578            },
579            "visual_identity": {
580                "cover": [
581                    {"url": "http://img.invalid/small.jpg", "size": "small"},
582                    {"url": "http://img.invalid/c.jpg", "size": "large"},
583                    {"url": "http://img.invalid/xl.jpg", "size": "xlarge"},
584                ]
585            },
586        },
587    )
588
589    await backend._handle_event(_event("track_changed", SoloistTrackChanged(item=item)))
590
591    event = events[0]
592    assert event.type is BackendEventType.METADATA
593    assert event.track_uri == "spotify:track:t1"
594    metadata = event.metadata
595    assert metadata is not None
596    assert metadata.track_uri == "spotify:track:t1"
597    assert metadata.title == "My Song"
598    assert metadata.artist == "Main Artist"
599    assert metadata.album == "The Album"
600    assert metadata.image_url == "http://img.invalid/c.jpg"
601    assert metadata.duration == 210
602    assert metadata.position == 0
603
604
605async def test_track_changed_with_sparse_decorations() -> None:
606    """Undecorated entities still produce a METADATA event with only the uri set."""
607    backend, events = _make_backend()
608    item = SoloistEntity(uri="spotify:track:t2", entity_type="track")
609
610    await backend._handle_event(_event("track_changed", SoloistTrackChanged(item=item)))
611
612    metadata = events[0].metadata
613    assert metadata is not None
614    assert metadata.track_uri == "spotify:track:t2"
615    assert metadata.title is None
616    assert metadata.artist is None
617    assert metadata.album is None
618    assert metadata.image_url is None
619    assert metadata.duration is None
620
621
622async def test_track_changed_without_item_is_other() -> None:
623    """A track_changed event without an item degrades to OTHER."""
624    backend, events = _make_backend()
625
626    await backend._handle_event(_event("track_changed", SoloistTrackChanged(item=None)))
627
628    assert events[0].type is BackendEventType.OTHER
629    assert events[0].metadata is None
630
631
632async def test_position_sync_maps_to_seconds() -> None:
633    """A position_sync event carries the position in whole seconds."""
634    backend, events = _make_backend()
635
636    await backend._handle_event(
637        _event(
638            "position_sync",
639            SoloistPositionSync(
640                position=SoloistPosition(position_ms=45999, timestamp_ms=1, speed=1.0)
641            ),
642        )
643    )
644
645    assert events[0].type is BackendEventType.POSITION
646    assert events[0].position == 45
647
648
649async def test_queue_changed_maps_entries() -> None:
650    """A queue_changed event maps its entries to normalized uid/uri/source/name entries."""
651    backend, events = _make_backend()
652    previous = [
653        SoloistQueueEntry(
654            uid="p1",
655            source="context",
656            item=SoloistEntity(
657                uri="spotify:track:t0",
658                entity_type="track",
659                decorations={"identity": {"name": "Played Song"}},
660            ),
661        ),
662    ]
663    upcoming = [
664        SoloistQueueEntry(
665            uid="u1",
666            source="queue",
667            item=SoloistEntity(
668                uri="spotify:track:t1",
669                entity_type="track",
670                decorations={"identity": {"name": "Queued Song"}},
671            ),
672        ),
673        # a name-less entry is tolerated (decorations is an extensible bag)
674        SoloistQueueEntry(
675            uid="u2",
676            source="autoplay",
677            item=SoloistEntity(uri="spotify:track:t2", entity_type="track"),
678        ),
679        # the title fallback used for track metadata applies to queue entries too
680        SoloistQueueEntry(
681            uid="u3",
682            source="new_source_kind",
683            item=SoloistEntity(
684                uri="spotify:track:t3",
685                entity_type="track",
686                decorations={"identity": {"title": "Titled Song"}},
687            ),
688        ),
689        # entries without a resolvable uri are skipped
690        SoloistQueueEntry(uid="u4", source="autoplay", item=None),
691        SoloistQueueEntry(
692            uid="u5", source="autoplay", item=SoloistEntity(uri="", entity_type="track")
693        ),
694    ]
695
696    await backend._handle_event(
697        _event("queue_changed", SoloistQueueChanged(previous=previous, upcoming=upcoming))
698    )
699
700    event = events[0]
701    assert event.type is BackendEventType.QUEUE_CHANGED
702    queue = event.queue
703    assert queue is not None
704    assert [(e.uid, e.uri, e.source, e.name) for e in queue.previous] == [
705        ("p1", "spotify:track:t0", QueueEntrySource.CONTEXT, "Played Song"),
706    ]
707    assert [(e.uid, e.uri, e.source, e.name) for e in queue.upcoming] == [
708        ("u1", "spotify:track:t1", QueueEntrySource.QUEUE, "Queued Song"),
709        ("u2", "spotify:track:t2", QueueEntrySource.AUTOPLAY, None),
710        # an unrecognized source value degrades to UNKNOWN instead of raising
711        ("u3", "spotify:track:t3", QueueEntrySource.UNKNOWN, "Titled Song"),
712    ]
713
714
715async def test_options_changed_maps_shuffle_and_repeat() -> None:
716    """An options_changed event carries the session's shuffle and repeat state."""
717    backend, events = _make_backend()
718
719    await backend._handle_event(
720        _event(
721            "options_changed",
722            SoloistOptionsChanged(options=SoloistPlaybackOptions(shuffle=True, repeat="context")),
723        )
724    )
725
726    event = events[0]
727    assert event.type is BackendEventType.OPTIONS_CHANGED
728    assert event.options is not None
729    assert event.options.shuffle is True
730    assert event.options.repeat is RepeatMode.ALL
731
732
733async def test_playback_state_options_emit_options_changed_precursor() -> None:
734    """A playback_state carrying options emits OPTIONS_CHANGED before the state event."""
735    backend, events = _make_backend()
736
737    await backend._handle_event(
738        _event(
739            "playback_state",
740            SoloistPlaybackState(
741                status="playing", options=SoloistPlaybackOptions(shuffle=True, repeat="track")
742            ),
743        )
744    )
745
746    assert [e.type for e in events] == [
747        BackendEventType.OPTIONS_CHANGED,
748        BackendEventType.PLAYING,
749    ]
750    options = events[0].options
751    assert options is not None
752    assert options.shuffle is True
753    assert options.repeat is RepeatMode.ONE
754    # the state event itself carries no options; OPTIONS_CHANGED is the one channel
755    assert events[1].options is None
756
757
758async def test_unknown_repeat_vocabulary_degrades_to_unknown() -> None:
759    """An unrecognized repeat value from the wire maps to RepeatMode.UNKNOWN."""
760    backend, events = _make_backend()
761
762    await backend._handle_event(
763        _event(
764            "options_changed",
765            SoloistOptionsChanged(options=SoloistPlaybackOptions(repeat="context_repeat")),
766        )
767    )
768
769    assert events[0].options is not None
770    assert events[0].options.repeat is RepeatMode.UNKNOWN
771
772
773async def test_uri_cache_feeds_all_events() -> None:
774    """Context/track uris from state events feed every later normalized event."""
775    backend, events = _make_backend()
776    state = SoloistPlaybackState(
777        status="playing",
778        item=SoloistEntity(uri="spotify:track:t1", entity_type="track"),
779        context=SoloistEntity(uri="spotify:playlist:ctx", entity_type="playlist"),
780    )
781
782    await backend._handle_event(_event("playback_state", state))
783    await backend._handle_event(
784        _event(
785            "position_sync",
786            SoloistPositionSync(
787                position=SoloistPosition(position_ms=1000, timestamp_ms=1, speed=1.0)
788            ),
789        )
790    )
791
792    # the unseen track first yields its metadata, then the playback event
793    assert events[0].type is BackendEventType.METADATA
794    assert events[1].type is BackendEventType.PLAYING
795    assert events[1].context_uri == "spotify:playlist:ctx"
796    assert events[1].track_uri == "spotify:track:t1"
797    # the position event does not carry uris itself; the cache fills them in
798    assert events[2].type is BackendEventType.POSITION
799    assert events[2].context_uri == "spotify:playlist:ctx"
800    assert events[2].track_uri == "spotify:track:t1"
801
802
803async def test_event_resets_restart_counter() -> None:
804    """A delivered event proves the daemon is healthy and resets the failure counter."""
805    backend, _events = _make_backend()
806    backend._restart_error_count = 3
807
808    await backend._handle_event(_event("command_result", SoloistCommandResult(command="pause")))
809
810    assert backend._restart_error_count == 0
811
812
813async def test_get_stream_source_named_pipe_with_readrate_pacing() -> None:
814    """The stream source is the sink FIFO as a named pipe, paced by ffmpeg readrate."""
815    backend, _events = _make_backend()
816    server: Any = _FakeServer()
817    sink: Any = _FakeSink()
818    backend._server = server
819    backend._sink = sink
820    backend._sink_generation = server.generation
821
822    source = await backend.get_stream_source()
823
824    assert source.stream_type is StreamType.NAMED_PIPE
825    assert source.path == str(sink.fifo_path)
826    assert source.extra_input_args == ["-readrate", "1", "-readrate_initial_burst", "0.5"]
827
828
829async def test_get_stream_source_stale_generation_raises_without_recovery() -> None:
830    """A stale sink fails the (side-effect-free) stream request; recovery is not run."""
831    backend, _events = _make_backend()
832    server: Any = _FakeServer()
833    server.generation = 3
834    sink: Any = _FakeSink("old")
835    proc: Any = _FakeProc()
836    backend._server = server
837    backend._sink = sink
838    backend._sink_generation = 2  # the pulse daemon restarted since sink creation
839    backend._proc = proc
840
841    with pytest.raises(AudioError, match="not available"):
842        await backend.get_stream_source()
843
844    # pure read: nothing was unloaded, closed or flagged for respawn
845    assert sink.unloaded == 0
846    assert backend._sink is sink
847    assert proc.closed == 0
848    assert backend._respawn_requested is False
849
850
851async def test_generation_watcher_recovers_stale_sink(monkeypatch: pytest.MonkeyPatch) -> None:
852    """The watcher notices a pulse daemon restart and drops sink + daemon for rebuild."""
853    backend, _events = _runner_backend()
854    monkeypatch.setattr(soloist_backend, "GENERATION_WATCH_INTERVAL_S", 0)
855    server: Any = backend._server
856    sink: Any = backend._sink
857    proc: Any = _FakeProc()
858    backend._proc = proc
859    watcher = asyncio.get_running_loop().create_task(backend._generation_watcher())
860    # a fresh generation passes several watch cycles untouched
861    for _ in range(5):
862        await asyncio.sleep(0)
863    assert sink.unloaded == 0
864
865    server.generation += 1
866    async with asyncio.timeout(1.0):
867        while proc.closed == 0:
868            await asyncio.sleep(0)
869
870    # the sink is dropped; the daemon supervisor recreates it before the respawn
871    assert sink.unloaded == 1
872    assert backend._sink is None
873    assert backend._respawn_requested is True
874    watcher.cancel()
875    with suppress(asyncio.CancelledError):
876        await watcher
877
878
879async def test_concurrent_ensure_fresh_sink_creates_single_sink(
880    monkeypatch: pytest.MonkeyPatch,
881) -> None:
882    """Concurrent supervisor calls replace a stale sink exactly once."""
883    backend, _events = _make_backend()
884    server: Any = _FakeServer()
885    server.generation = 5
886    old_sink: Any = _FakeSink("old")
887    backend._server = server
888    backend._sink = old_sink
889    backend._sink_generation = 4  # stale: the pulse daemon restarted
890    created: list[Any] = []
891    gate = asyncio.Event()
892
893    async def _create(_server: Any, _prefix: str) -> Any:
894        await gate.wait()
895        sink = _FakeSink("new")
896        created.append(sink)
897        return sink
898
899    monkeypatch.setattr(soloist_backend, "PipeSink", SimpleNamespace(create=_create))
900    loop = asyncio.get_running_loop()
901    task1 = loop.create_task(backend._ensure_fresh_sink())
902    task2 = loop.create_task(backend._ensure_fresh_sink())
903    for _ in range(5):
904        await asyncio.sleep(0)
905    gate.set()
906    sink1, sink2 = await asyncio.gather(task1, task2)
907
908    assert len(created) == 1
909    assert sink1 is sink2 is created[0]
910    assert old_sink.unloaded == 1
911
912
913async def test_get_stream_source_after_stop_raises_clean_error() -> None:
914    """get_stream_source on a stopped backend raises AudioError, not AssertionError."""
915    backend, _events = _make_backend()
916    server: Any = _FakeServer()
917    backend._server = server
918    await backend.stop()
919
920    with pytest.raises(AudioError, match="not available"):
921        await backend.get_stream_source()
922
923
924async def test_spawn_resets_stale_volume_state(monkeypatch: pytest.MonkeyPatch) -> None:
925    """A freshly spawned daemon starts at 100%: stale volume state and sink gain are reset."""
926    backend, _events = _runner_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
927    backend._spotify_volume = 25  # stale from before a crash (sink compensating at 400%)
928    sink: Any = backend._sink
929    proc = _FakeProc(on_close=lambda: setattr(backend, "_stop_called", True))
930    _patch_spawn(monkeypatch, [proc])
931
932    await backend._daemon_runner()
933
934    assert backend._spotify_volume == 100
935    assert sink.volumes == [100]
936
937
938async def test_failed_unity_reset_fails_closed() -> None:
939    """A failed unity reset drops sink and daemon: a stale gain must never clip audio."""
940    backend, _events = _runner_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
941    sink: Any = backend._sink
942    sink.set_volume_error = RuntimeError("pulse gone")
943    proc: Any = _FakeProc()
944    backend._proc = proc
945
946    await backend._reset_volume_state(sink)
947
948    assert sink.unloaded == 1
949    assert backend._sink is None
950    assert backend._respawn_requested is True
951    assert proc.closed == 1
952
953
954async def test_failed_compensation_fails_closed_and_suppresses_volume_event() -> None:
955    """A failed compensation set recovers sink + daemon and never forwards the VOLUME event."""
956    backend, events = _make_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
957    sink: Any = _FakeSink()
958    sink.set_volume_error = RuntimeError("pulse gone")
959    proc: Any = _FakeProc()
960    backend._sink = sink
961    backend._proc = proc
962
963    await backend._handle_event(_volume_event(50))
964
965    # the player must not adopt a volume whose compensation is unknown
966    assert events == []
967    assert sink.unloaded == 1
968    assert backend._sink is None
969    assert backend._respawn_requested is True
970    assert proc.closed == 1
971
972
973async def test_failed_compensation_drops_the_playback_snapshot() -> None:
974    """A snapshot whose volume resync triggered recovery is not forwarded as playback state."""
975    backend, events = _make_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
976    sink: Any = _FakeSink()
977    sink.set_volume_error = RuntimeError("pulse gone")
978    proc: Any = _FakeProc()
979    backend._sink = sink
980    backend._proc = proc
981
982    await backend._handle_event(
983        _event("playback_state", SoloistPlaybackState(status="playing", volume=50))
984    )
985
986    # no PLAYING against the torn-down sink; the respawned daemon reports fresh state
987    assert events == []
988    assert backend._sink is None
989    assert proc.closed == 1
990
991
992async def test_binary_refresh_loop_respawns_on_new_build(
993    monkeypatch: pytest.MonkeyPatch,
994) -> None:
995    """The daily refresh survives failures and restarts the daemon once a new build lands."""
996    backend, _events = _runner_backend()
997    monkeypatch.setattr(soloist_backend, "BINARY_REFRESH_INTERVAL_S", 0)
998    proc: Any = _FakeProc()
999    backend._proc = proc
1000    backend._build_sha = "sha-old"
1001    checks: list[bool] = []
1002    sha = {"value": "sha-old"}
1003
1004    class _FakeManager:
1005        """Fake binary manager: fails once, idles once, then installs a new build."""
1006
1007        def __init__(self, mass: Any) -> None:
1008            """Accept the mass argument like the real manager."""
1009
1010        def diagnostics(self) -> dict[str, Any]:
1011            """Report the currently installed build's digest."""
1012            return {"installed": True, "sha256": sha["value"]}
1013
1014        async def ensure_fresh(self, consent: bool, *, force: bool = False) -> Path:
1015            """Fail the first check, keep the build on the second, replace it on the third."""
1016            checks.append(consent)
1017            if len(checks) == 1:
1018                raise OSError("cdn offline")
1019            if len(checks) >= 3:
1020                # a replacement build installs onto the SAME path; only the
1021                # install metadata's digest changes
1022                sha["value"] = "sha-new"
1023            return Path("/fake/bin/soloist")
1024
1025    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", _FakeManager)
1026    loop_task = asyncio.get_running_loop().create_task(backend._binary_refresh_loop())
1027
1028    async with asyncio.timeout(1.0):
1029        while proc.closed == 0:
1030            await asyncio.sleep(0)
1031    loop_task.cancel()
1032    with suppress(asyncio.CancelledError):
1033        await loop_task
1034
1035    # failed check + unchanged check passed without a respawn; the changed
1036    # digest triggered exactly one intentional daemon restart
1037    assert len(checks) >= 3
1038    assert all(checks)  # ensure_fresh is always called with the consent flag
1039
1040
1041async def test_binary_refresh_loop_picks_up_sibling_install(
1042    monkeypatch: pytest.MonkeyPatch,
1043) -> None:
1044    """A build installed by a sibling instance onto the shared path still triggers a respawn."""
1045    backend, _events = _runner_backend()
1046    monkeypatch.setattr(soloist_backend, "BINARY_REFRESH_INTERVAL_S", 0)
1047    proc: Any = _FakeProc()
1048    backend._proc = proc
1049    # this instance spawned its daemon from the old build; a sibling instance
1050    # already replaced the shared install before this loop's first check
1051    backend._build_sha = "sha-old"
1052
1053    class _FakeManager:
1054        """Fake binary manager whose shared install was updated by a sibling."""
1055
1056        def __init__(self, mass: Any) -> None:
1057            """Accept the mass argument like the real manager."""
1058
1059        def diagnostics(self) -> dict[str, Any]:
1060            """Report the sibling-installed build's digest."""
1061            return {"installed": True, "sha256": "sha-new"}
1062
1063        async def ensure_fresh(self, consent: bool, *, force: bool = False) -> Path:
1064            """Return the (already fresh) shared install path."""
1065            return Path("/fake/bin/soloist")
1066
1067    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", _FakeManager)
1068    loop_task = asyncio.get_running_loop().create_task(backend._binary_refresh_loop())
1069
1070    async with asyncio.timeout(1.0):
1071        while proc.closed == 0:
1072            await asyncio.sleep(0)
1073    loop_task.cancel()
1074    with suppress(asyncio.CancelledError):
1075        await loop_task
1076
1077    assert backend._build_sha == "sha-new"
1078    assert proc.closed == 1
1079    assert backend._respawn_requested is True
1080    assert backend._binary == Path("/fake/bin/soloist")
1081
1082
1083async def test_stdout_redacts_api_key(
1084    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
1085) -> None:
1086    """The api key is redacted from daemon stdout lines before they are logged."""
1087    backend, _events = _runner_backend()
1088    proc = _FakeProc(
1089        stdout_lines=[f"argv: --api-key {_API_KEY}"],
1090        on_close=lambda: setattr(backend, "_stop_called", True),
1091    )
1092    _patch_spawn(monkeypatch, [proc])
1093
1094    with caplog.at_level(logging.DEBUG):
1095        await backend._daemon_runner()
1096
1097    assert all(_API_KEY not in record.getMessage() for record in caplog.records)
1098    assert any("<redacted>" in record.getMessage() for record in caplog.records)
1099
1100
1101async def test_daemon_runner_does_not_wait_for_the_log_reader(
1102    monkeypatch: pytest.MonkeyPatch,
1103) -> None:
1104    """
1105    A log reader that never ends must not hold up the daemon supervisor.
1106
1107    AsyncProcess.close() takes the stream lock and keeps it, so a reader parked
1108    mid-line when another supervisor closes the daemon (sink replacement, binary
1109    refresh) never reaches EOF. The runner therefore has to key off the process
1110    exit, not off its own reader.
1111    """
1112
1113    class _StuckReaderProc(_FakeProc):
1114        """A daemon whose output reader never ends, but which does exit."""
1115
1116        async def iter_stdout(self) -> AsyncGenerator[str]:
1117            await asyncio.Event().wait()  # never returns, never yields
1118            yield ""  # pragma: no cover
1119
1120        async def wait(self) -> int:
1121            return self._exit_code
1122
1123    backend, _events = _runner_backend()
1124    proc = _StuckReaderProc(on_close=lambda: setattr(backend, "_stop_called", True))
1125    _patch_spawn(monkeypatch, [proc])
1126    # a short drain leaves the outer deadline real headroom, so the assertion is
1127    # about the runner returning rather than about which timeout fires first
1128    monkeypatch.setattr(soloist_backend, "DAEMON_LOG_DRAIN_TIMEOUT_S", 0.1)
1129
1130    # the supervisor must return on its own; a hang here is the regression
1131    async with asyncio.timeout(5):
1132        await backend._daemon_runner()
1133
1134    assert proc.closed == 1
1135
1136
1137async def test_daemon_runner_cancellation_stops_the_supervisor(
1138    monkeypatch: pytest.MonkeyPatch,
1139) -> None:
1140    """
1141    Cancelling the supervisor stops it instead of respawning the daemon.
1142
1143    The log reader is cleaned up on the way out, so the cancellation must not
1144    be consumed by that cleanup and leave the loop running.
1145    """
1146    backend, _events = _runner_backend()
1147    proc = _FakeProc(block_stdout=True)
1148    _patch_spawn(monkeypatch, [proc])
1149
1150    task = asyncio.create_task(backend._daemon_runner())
1151    # let the runner reach its wait on the (blocked) daemon before cancelling
1152    for _ in range(20):
1153        await asyncio.sleep(0)
1154    task.cancel()
1155
1156    with pytest.raises(asyncio.CancelledError):
1157        await task
1158    assert task.cancelled()
1159
1160
1161async def test_daemon_runner_restarts_when_the_log_reader_dies(
1162    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
1163) -> None:
1164    """
1165    A reader that fails closes the daemon instead of leaving it wedged.
1166
1167    Nothing else drains the daemon's stdout, so once the pipe fills a daemon
1168    with a dead reader can never make progress and the supervisor would wait
1169    on it forever.
1170    """
1171
1172    class _FailingReaderProc(_FakeProc):
1173        """A daemon whose log reader raises while the process is still alive."""
1174
1175        async def iter_stdout(self) -> AsyncGenerator[str]:
1176            for line in self._stdout_lines:
1177                yield line
1178            raise RuntimeError("reader blew up")
1179
1180        async def wait(self) -> int:
1181            # only ever returns once something closes the daemon
1182            await self._closed_event.wait()
1183            return self._exit_code
1184
1185    backend, _events = _runner_backend()
1186    proc = _FailingReaderProc(on_close=lambda: setattr(backend, "_stop_called", True))
1187    _patch_spawn(monkeypatch, [proc])
1188
1189    with caplog.at_level(logging.ERROR):
1190        async with asyncio.timeout(5):
1191            await backend._daemon_runner()
1192
1193    assert proc.closed >= 1
1194    assert any("log reader failed" in record.getMessage() for record in caplog.records)
1195
1196
1197async def test_daemon_runner_drains_buffered_log_after_exit(
1198    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
1199) -> None:
1200    """
1201    Output still buffered when the daemon exits is logged, not dropped.
1202
1203    A daemon that fails at startup writes its reason and exits within
1204    milliseconds, so dropping the reader the moment the process ends throws
1205    away exactly the output that explains the failure.
1206    """
1207
1208    class _BufferedProc(_FakeProc):
1209        """A daemon that has already exited with its output still queued."""
1210
1211        async def wait(self) -> int:
1212            return self._exit_code
1213
1214        async def iter_stdout(self) -> AsyncGenerator[str]:
1215            for line in self._stdout_lines:
1216                await asyncio.sleep(0)  # the reader cannot drain it all in one step
1217                yield line
1218
1219    backend, _events = _runner_backend()
1220    proc = _BufferedProc(
1221        stdout_lines=[f"buffered line {index}" for index in range(20)],
1222        on_close=lambda: setattr(backend, "_stop_called", True),
1223    )
1224    _patch_spawn(monkeypatch, [proc])
1225
1226    with caplog.at_level(logging.DEBUG):
1227        await backend._daemon_runner()
1228
1229    logged = [record.getMessage() for record in caplog.records]
1230    assert sum("buffered line" in message for message in logged) == 20
1231
1232
1233async def test_intentional_respawn_skips_failure_accounting(
1234    monkeypatch: pytest.MonkeyPatch,
1235) -> None:
1236    """A sink recovery respawn restarts immediately and never counts as a failure."""
1237    backend, _events = _runner_backend()
1238    server: Any = backend._server
1239    proc1 = _FakeProc(block_stdout=True)
1240    proc2 = _FakeProc(block_stdout=True)
1241    spawned = _patch_spawn(monkeypatch, [proc1, proc2])
1242    new_sink: Any = _FakeSink("new")
1243    monkeypatch.setattr(
1244        soloist_backend, "PipeSink", SimpleNamespace(create=AsyncMock(return_value=new_sink))
1245    )
1246    runner = asyncio.get_running_loop().create_task(backend._daemon_runner())
1247    async with asyncio.timeout(1.0):
1248        while backend._proc is None:
1249            await asyncio.sleep(0)
1250
1251    # the pulse daemon restarted: the recovery routine (as run by the watcher)
1252    # drops the sink and intentionally closes the running daemon
1253    server.generation += 1
1254    await backend._recover_sink()
1255    assert proc1.closed >= 1
1256    # the supervisor recreates the sink and respawns without the restart delay
1257    # (no RESTART_DELAY patch: a counted failure would make this wait time out)
1258    async with asyncio.timeout(1.0):
1259        while len(spawned) < 2:
1260            await asyncio.sleep(0)
1261
1262    source = await backend.get_stream_source()
1263    assert source.path == str(new_sink.fifo_path)
1264    assert backend._sink is new_sink
1265    assert backend._sink_generation == server.generation
1266    assert backend._restart_error_count == 0
1267    assert backend._respawn_requested is False
1268    backend._stop_called = True
1269    await proc2.close()
1270    await runner
1271
1272
1273async def test_playback_state_volume_resyncs_compensation() -> None:
1274    """A playback_state carrying a volume resyncs the sink before the playback event."""
1275    backend, events = _make_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
1276    sink: Any = _FakeSink()
1277    backend._sink = sink
1278
1279    await backend._handle_event(
1280        _event("playback_state", SoloistPlaybackState(status="playing", volume=50))
1281    )
1282
1283    assert sink.volumes == [200.0]
1284    assert [(event.type, event.volume) for event in events] == [
1285        (BackendEventType.VOLUME, 50),
1286        (BackendEventType.PLAYING, None),
1287    ]
1288
1289    # an unchanged volume on the next snapshot is not re-applied
1290    await backend._handle_event(
1291        _event("playback_state", SoloistPlaybackState(status="paused", volume=50))
1292    )
1293    assert sink.volumes == [200.0]
1294    assert events[-1].type is BackendEventType.PAUSED
1295
1296
1297async def test_playback_state_volume_pins_in_player_only() -> None:
1298    """player_only: an off-100 playback_state volume re-pins the daemon, no VOLUME event."""
1299    backend, events = _make_backend(volume_mode=VOLUME_MODE_PLAYER_ONLY)
1300    client = AsyncMock()
1301    backend._client = client
1302    sink: Any = _FakeSink()
1303    backend._sink = sink
1304
1305    await backend._handle_event(
1306        _event("playback_state", SoloistPlaybackState(status="playing", volume=80))
1307    )
1308
1309    client.set_volume.assert_awaited_once_with(100)
1310    assert [event.type for event in events] == [BackendEventType.PLAYING]
1311
1312
1313async def test_failed_pin_marks_volume_unknown_and_retries() -> None:
1314    """player_only: a failed 100% pin is retried by a snapshot reporting the same volume."""
1315    backend, _events = _make_backend(volume_mode=VOLUME_MODE_PLAYER_ONLY)
1316    client = AsyncMock()
1317    client.set_volume.side_effect = [OSError("ws down"), None]
1318    backend._client = client
1319
1320    await backend._handle_event(
1321        _event("playback_state", SoloistPlaybackState(status="playing", volume=80))
1322    )
1323    assert backend._spotify_volume is None
1324    # the reconnect snapshot reports the unchanged volume; the pin is retried
1325    await backend._handle_event(
1326        _event("playback_state", SoloistPlaybackState(status="paused", volume=80))
1327    )
1328    assert client.set_volume.await_count == 2
1329
1330
1331async def test_playback_state_snapshot_emits_metadata_for_unseen_track() -> None:
1332    """A snapshot carrying an unseen track emits its metadata; a repeat does not."""
1333    backend, events = _make_backend()
1334    state = SoloistPlaybackState(
1335        status="playing",
1336        item=SoloistEntity(uri="spotify:track:t1", entity_type="track"),
1337    )
1338
1339    await backend._handle_event(_event("playback_state", state))
1340    await backend._handle_event(_event("playback_state", state))
1341
1342    assert [event.type for event in events] == [
1343        BackendEventType.METADATA,
1344        BackendEventType.PLAYING,
1345        BackendEventType.PLAYING,
1346    ]
1347    assert events[0].metadata is not None
1348    assert events[0].metadata.track_uri == "spotify:track:t1"
1349
1350
1351def test_sink_prefix_is_sanitized() -> None:
1352    """Characters unsafe for PA sink names are stripped from the identity key."""
1353    backend, _events = _make_backend(identity_key="weird id!*")
1354
1355    assert backend._sink_prefix == "weird_id__"
1356
1357
1358def test_decoded_format_reports_the_capture_pcm() -> None:
1359    """The decoded format is the fixed capture PCM (s32le/44.1/2) the pipe delivers."""
1360    backend, _events = _make_backend()
1361
1362    decoded = backend.decoded_audio_format
1363    assert decoded.content_type is ContentType.PCM_S32LE
1364    assert decoded.sample_rate == 44100
1365    assert decoded.bit_depth == 32
1366    assert decoded.channels == 2
1367    assert backend.get_audio_reader() is None
1368
1369
1370def test_each_stream_gets_its_own_copy_of_the_capture_format() -> None:
1371    """
1372    Every stream is handed its own decoded format, not one shared object.
1373
1374    ffmpeg writes what it probes onto the format it is given, so a shared instance
1375    would carry one stream's probe over into the next.
1376    """
1377    backend, _events = _make_backend()
1378
1379    first = backend.decoded_audio_format
1380    second = backend.decoded_audio_format
1381
1382    assert first is not second
1383    first.bit_rate = 12345
1384    assert second.bit_rate != 12345
1385    assert backend.decoded_audio_format.bit_rate != 12345
1386
1387
1388@pytest.mark.parametrize(
1389    ("audio_quality", "content_type", "bit_depth"),
1390    [
1391        (AUDIO_QUALITY_NORMAL, ContentType.OGG, 16),
1392        (AUDIO_QUALITY_HIGH, ContentType.OGG, 16),
1393        (AUDIO_QUALITY_VERY_HIGH, ContentType.OGG, 16),
1394        (AUDIO_QUALITY_LOSSLESS, ContentType.FLAC, 24),
1395    ],
1396)
1397def test_advertised_format_follows_the_configured_tier(
1398    audio_quality: str, content_type: ContentType, bit_depth: int
1399) -> None:
1400    """The display format is the tier Spotify was asked for, not the capture PCM."""
1401    backend, _events = _make_backend(audio_quality=audio_quality)
1402
1403    assert backend.audio_format.content_type is content_type
1404    assert backend.audio_format.bit_depth == bit_depth
1405
1406
1407@pytest.mark.parametrize(
1408    ("entity_type", "content_type"),
1409    [
1410        ("track", ContentType.FLAC),
1411        ("episode", ContentType.OGG),
1412        ("chapter", ContentType.OGG),
1413    ],
1414)
1415async def test_spoken_content_is_never_advertised_as_lossless(
1416    entity_type: str, content_type: ContentType
1417) -> None:
1418    """Spotify serves lossless for music only, whatever the tier is set to."""
1419    backend, _events = _make_backend(audio_quality=AUDIO_QUALITY_LOSSLESS)
1420    item = SoloistEntity(uri=f"spotify:{entity_type}:x1", entity_type=entity_type)
1421
1422    await backend._handle_event(_event("track_changed", SoloistTrackChanged(item=item)))
1423
1424    assert backend.audio_format.content_type is content_type
1425
1426
1427async def test_transport_commands_map_to_client() -> None:
1428    """Transport commands map 1:1 onto the SoloistClient methods."""
1429    backend, _events = _make_backend()
1430    client = AsyncMock()
1431    backend._client = client
1432
1433    await backend.play("spotify:album:x", skip_to_uri="spotify:track:y")
1434    # play claims active device status first (Connect transfer), then plays
1435    client.activate.assert_awaited_once_with(await_result=True)
1436    client.play.assert_awaited_once_with("spotify:album:x")
1437    call_names = [name for name, _args, _kwargs in client.mock_calls]
1438    assert call_names.index("activate") < call_names.index("play")
1439
1440    # resume also re-claims active device status first
1441    client.reset_mock()
1442    await backend.resume()
1443    client.activate.assert_awaited_once_with(await_result=True)
1444    client.resume.assert_awaited_once_with()
1445    await backend.pause()
1446    client.pause.assert_awaited_once_with()
1447    await backend.next()
1448    client.skip_next.assert_awaited_once_with()
1449    await backend.previous()
1450    client.skip_prev.assert_awaited_once_with()
1451    await backend.seek(30000)
1452    client.seek.assert_awaited_once_with(30000)
1453
1454    # deactivate pauses first (position preserved), then gives up the device
1455    client.reset_mock()
1456    await backend.deactivate()
1457    client.pause.assert_awaited_once_with(await_result=True)
1458    client.deactivate.assert_awaited_once_with()
1459    call_names = [name for name, _args, _kwargs in client.mock_calls]
1460    assert call_names.index("pause") < call_names.index("deactivate")
1461
1462
1463async def test_queue_commands_map_to_client() -> None:
1464    """The queue-session verbs pass through to the SoloistClient."""
1465    backend, _events = _make_backend()
1466    client = AsyncMock()
1467    backend._client = client
1468
1469    assert backend.supports_queue_control is True
1470
1471    await backend.add_to_queue("spotify:track:t1")
1472    client.add_to_queue.assert_awaited_once_with("spotify:track:t1")
1473
1474    await backend.set_shuffle(True)
1475    client.set_shuffle.assert_awaited_once_with(True)
1476
1477    # the queue snapshot arrives as a queue_changed event, no ack is awaited
1478    await backend.request_queue(limit=25)
1479    client.get_queue.assert_awaited_once_with(25)
1480
1481
1482@pytest.mark.parametrize(
1483    ("repeat", "expected_calls"),
1484    [
1485        (RepeatMode.OFF, [("set_repeat_track", False), ("set_repeat_context", False)]),
1486        (RepeatMode.ALL, [("set_repeat_track", False), ("set_repeat_context", True)]),
1487        (RepeatMode.ONE, [("set_repeat_context", False), ("set_repeat_track", True)]),
1488    ],
1489)
1490async def test_set_repeat_sequences_the_two_flags(
1491    repeat: RepeatMode, expected_calls: list[tuple[str, bool]]
1492) -> None:
1493    """set_repeat disables one repeat flag before enabling the other, awaiting each ack."""
1494    backend, _events = _make_backend()
1495    client = AsyncMock()
1496    backend._client = client
1497
1498    await backend.set_repeat(repeat)
1499
1500    assert [(name, args[0]) for name, args, _kwargs in client.mock_calls] == expected_calls
1501    # each command waits for its ack so the pair cannot race
1502    assert all(kwargs == {"await_result": True} for _name, _args, kwargs in client.mock_calls)
1503
1504
1505async def test_set_repeat_rejects_unknown() -> None:
1506    """set_repeat refuses RepeatMode.UNKNOWN instead of silently disabling repeat."""
1507    backend, _events = _make_backend()
1508    client = AsyncMock()
1509    backend._client = client
1510
1511    with pytest.raises(ValueError, match="unknown repeat mode"):
1512        await backend.set_repeat(RepeatMode.UNKNOWN)
1513    assert client.mock_calls == []
1514
1515
1516async def test_set_repeat_serializes_concurrent_calls() -> None:
1517    """Concurrent set_repeat calls cannot interleave their two-command sequences."""
1518    backend, _events = _make_backend()
1519    call_order: list[tuple[str, bool]] = []
1520
1521    async def record(name: str, enabled: bool, **_kwargs: Any) -> None:
1522        call_order.append((name, enabled))
1523        await asyncio.sleep(0)  # yield so an unserialized second call could interleave
1524
1525    client = AsyncMock()
1526    client.set_repeat_track.side_effect = partial(record, "set_repeat_track")
1527    client.set_repeat_context.side_effect = partial(record, "set_repeat_context")
1528    backend._client = client
1529
1530    await asyncio.gather(backend.set_repeat(RepeatMode.ALL), backend.set_repeat(RepeatMode.ONE))
1531
1532    assert call_order == [
1533        ("set_repeat_track", False),
1534        ("set_repeat_context", True),
1535        ("set_repeat_context", False),
1536        ("set_repeat_track", True),
1537    ]
1538
1539
1540async def test_player_only_pins_spotify_volume_and_suppresses_events() -> None:
1541    """player_only: off-100 volume events reset the daemon to 100 and are suppressed."""
1542    backend, events = _make_backend(volume_mode=VOLUME_MODE_PLAYER_ONLY)
1543    client = AsyncMock()
1544    backend._client = client
1545
1546    await backend._handle_event(_volume_event(80))
1547    client.set_volume.assert_awaited_once_with(100)
1548    assert events == []  # never forwarded: it would fight the MA player volume
1549
1550    client.set_volume.reset_mock()
1551    await backend._handle_event(_volume_event(100))
1552    client.set_volume.assert_not_awaited()
1553    assert events == []
1554
1555
1556async def test_player_only_set_volume_pins_100_once() -> None:
1557    """player_only: MA volume pushes pin the daemon at 100 and dedupe afterwards."""
1558    backend, _events = _make_backend(volume_mode=VOLUME_MODE_PLAYER_ONLY)
1559    client = AsyncMock()
1560    backend._client = client
1561
1562    await backend.set_volume(55)
1563    client.set_volume.assert_awaited_once_with(100)
1564
1565    # once the daemon confirmed 100 (via its volume event) the pin is deduped
1566    await backend._handle_event(_volume_event(100))
1567    client.set_volume.reset_mock()
1568    await backend.set_volume(70)
1569    client.set_volume.assert_not_awaited()
1570
1571
1572@pytest.mark.parametrize(
1573    ("volume", "sink_pct"),
1574    [(80, 125.0), (50, 200.0), (25, 400.0), (10, 1000.0), (1, 10000.0)],
1575)
1576async def test_sync_spotify_reciprocal_sink_compensation(volume: int, sink_pct: float) -> None:
1577    """sync_spotify: the sink gain is the reciprocal of the Spotify volume percentage."""
1578    backend, events = _make_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
1579    sink: Any = _FakeSink()
1580    backend._sink = sink
1581
1582    await backend._handle_event(_volume_event(volume))
1583
1584    assert sink.volumes == [sink_pct]
1585    assert [(event.type, event.volume) for event in events] == [(BackendEventType.VOLUME, volume)]
1586
1587
1588async def test_sync_spotify_zero_volume_silences_sink() -> None:
1589    """sync_spotify: volume 0 silences the sink (no reciprocal exists) and forwards 0."""
1590    backend, events = _make_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
1591    sink: Any = _FakeSink()
1592    backend._sink = sink
1593
1594    await backend._handle_event(_volume_event(0))
1595
1596    assert sink.volumes == [0.0]
1597    assert [(event.type, event.volume) for event in events] == [(BackendEventType.VOLUME, 0)]
1598
1599
1600async def test_sync_spotify_set_volume_passes_through() -> None:
1601    """sync_spotify: MA volume changes go straight to the daemon, not the sink."""
1602    backend, _events = _make_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
1603    client = AsyncMock()
1604    sink: Any = _FakeSink()
1605    backend._client = client
1606    backend._sink = sink
1607
1608    await backend.set_volume(42)
1609
1610    client.set_volume.assert_awaited_once_with(42)
1611    assert sink.volumes == []
1612
1613
1614async def test_sync_spotify_volume_ops_serialized() -> None:
1615    """sync_spotify: concurrent volume events apply their sink/forward ops in order."""
1616    backend, events = _make_backend(volume_mode=VOLUME_MODE_SYNC_SPOTIFY)
1617    sink: Any = _FakeSink()
1618    sink.gate = asyncio.Event()
1619    backend._sink = sink
1620
1621    task1 = asyncio.get_running_loop().create_task(backend._handle_event(_volume_event(50)))
1622    task2 = asyncio.get_running_loop().create_task(backend._handle_event(_volume_event(80)))
1623    for _ in range(5):
1624        await asyncio.sleep(0)
1625    # first op parked on the sink gate, second queued on the volume lock
1626    assert sink.volumes == []
1627    assert events == []
1628
1629    sink.gate.set()
1630    await asyncio.gather(task1, task2)
1631
1632    assert sink.volumes == [200.0, 125.0]
1633    assert [(event.type, event.volume) for event in events] == [
1634        (BackendEventType.VOLUME, 50),
1635        (BackendEventType.VOLUME, 80),
1636    ]
1637
1638
1639async def test_stop_teardown_order_and_idempotency() -> None:
1640    """stop() tears down events task, daemon task, process, sink, server — exactly once."""
1641    backend, _events = _make_backend()
1642    order: list[str] = []
1643
1644    async def _supervisor(tag: str) -> None:
1645        try:
1646            await asyncio.sleep(3600)
1647        except asyncio.CancelledError:
1648            order.append(tag)
1649            raise
1650
1651    class _Proc:
1652        """Minimal process stub recording its close."""
1653
1654        returncode: int | None = None
1655
1656        async def close(self) -> None:
1657            """Record the close."""
1658            order.append("proc")
1659
1660    class _Sink:
1661        """Minimal sink stub recording its unload."""
1662
1663        async def unload(self) -> None:
1664            """Record the unload."""
1665            order.append("sink")
1666
1667    class _Server:
1668        """Minimal capture server stub recording its release."""
1669
1670        generation = 1
1671
1672        async def release(self) -> None:
1673            """Record the release."""
1674            order.append("server")
1675
1676    loop = asyncio.get_running_loop()
1677    backend._events_task = loop.create_task(_supervisor("events"))
1678    backend._daemon_task = loop.create_task(_supervisor("daemon"))
1679    await asyncio.sleep(0)  # let the supervisors enter their sleep
1680    proc: Any = _Proc()
1681    sink: Any = _Sink()
1682    server: Any = _Server()
1683    backend._proc = proc
1684    backend._sink = sink
1685    backend._server = server
1686
1687    await backend.stop()
1688
1689    assert order == ["events", "daemon", "proc", "sink", "server"]
1690    assert backend._stop_called is True
1691
1692    await backend.stop()  # second call must be a no-op
1693    assert order == ["events", "daemon", "proc", "sink", "server"]
1694
1695
1696async def test_start_failure_releases_capture_server(
1697    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
1698) -> None:
1699    """A startup failure after acquiring the capture server releases it again."""
1700    backend, _events = _make_backend(base_dir=tmp_path)
1701
1702    class _FakeManager:
1703        def __init__(self, mass: Any) -> None:
1704            """Accept the mass argument like the real manager."""
1705
1706        def diagnostics(self) -> dict[str, Any]:
1707            """Report the installed build's digest."""
1708            return {"installed": True, "sha256": "sha-1"}
1709
1710        async def ensure_fresh(self, consent: bool, *, force: bool = False) -> Path:
1711            """Hand out a fake binary path."""
1712            return Path("/fake/bin/soloist")
1713
1714    server: Any = _FakeServer()
1715    monkeypatch.setattr(soloist_backend, "SoloistBinaryManager", _FakeManager)
1716    monkeypatch.setattr(soloist_backend, "get_pulse_capture_server", lambda _mass: server)
1717    monkeypatch.setattr(
1718        soloist_backend,
1719        "PipeSink",
1720        SimpleNamespace(create=AsyncMock(side_effect=RuntimeError("sink creation failed"))),
1721    )
1722
1723    with pytest.raises(RuntimeError, match="sink creation failed"):
1724        await backend.start()
1725
1726    # the acquire must be paired with a release despite the aborted startup
1727    assert server.released
1728
1729
1730def _prefs_backend(
1731    tmp_path: Path,
1732    *,
1733    crossfade_ms: int,
1734    normalization: bool,
1735    audio_quality: str = AUDIO_QUALITY_LOSSLESS,
1736) -> SoloistBackend:
1737    """Build a backend with the given audio behavior, rooted in a real tmp data dir."""
1738    backend, _ = _make_backend(base_dir=tmp_path)
1739    backend._crossfade_ms = crossfade_ms
1740    backend._loudness_normalization = normalization
1741    backend._audio_quality = audio_quality
1742    backend._data_dir = tmp_path / "soloist-data"
1743    return backend
1744
1745
1746def test_audio_prefs_written_to_global_and_per_user(tmp_path: Path) -> None:
1747    """Managed keys are replaced in the global and every per-user prefs store."""
1748    backend = _prefs_backend(tmp_path, crossfade_ms=8000, normalization=False)
1749    settings = backend._data_dir / "settings"
1750    (settings / "Users" / "alice-user").mkdir(parents=True)
1751    (settings / "prefs").write_text("core.clock_delta=0\naudio.crossfade_v2=false\n")
1752    (settings / "Users" / "alice-user" / "prefs").write_text(
1753        "storage.size=512\naudio.crossfade.time_v2=99\n"
1754    )
1755
1756    backend._write_audio_prefs()
1757
1758    global_prefs = (settings / "prefs").read_text().splitlines()
1759    user_prefs = (settings / "Users" / "alice-user" / "prefs").read_text().splitlines()
1760    for prefs in (global_prefs, user_prefs):
1761        assert "audio.crossfade_v2=true" in prefs
1762        assert "audio.crossfade.time_v2=8000" in prefs
1763        assert "audio.normalize_v2=false" in prefs
1764    # foreign keys survive, replaced stale values do not
1765    assert "core.clock_delta=0" in global_prefs
1766    assert "storage.size=512" in user_prefs
1767    assert "audio.crossfade_v2=false" not in global_prefs
1768    assert "audio.crossfade.time_v2=99" not in user_prefs
1769
1770
1771@pytest.mark.parametrize(
1772    ("tier", "expected"),
1773    [
1774        (AUDIO_QUALITY_NORMAL, 2),
1775        (AUDIO_QUALITY_HIGH, 3),
1776        (AUDIO_QUALITY_VERY_HIGH, 4),
1777        (AUDIO_QUALITY_LOSSLESS, 5),
1778        # an unknown tier must never reach the prefs file: the engine rejects
1779        # anything outside 1-5 and silently drops back to ~160 kbps
1780        ("nonsense", 5),
1781    ],
1782)
1783def test_audio_prefs_quality_tier_mapping(tmp_path: Path, tier: str, expected: int) -> None:
1784    """Each quality tier writes its bitrate enumeration to both quality keys."""
1785    backend = _prefs_backend(tmp_path, crossfade_ms=0, normalization=True, audio_quality=tier)
1786
1787    backend._write_audio_prefs()
1788
1789    prefs = (backend._data_dir / "settings" / "prefs").read_text().splitlines()
1790    assert f"audio.play_bitrate_enumeration={expected}" in prefs
1791    assert f"audio.play_bitrate_non_metered_enumeration={expected}" in prefs
1792    # without the migration marker the engine derives the non-metered value itself
1793    assert "audio.play_bitrate_non_metered_migrated=true" in prefs
1794
1795
1796def test_audio_prefs_replace_a_stale_quality_tier(tmp_path: Path) -> None:
1797    """A quality value left by a previous run is replaced, not appended to."""
1798    backend = _prefs_backend(
1799        tmp_path, crossfade_ms=0, normalization=True, audio_quality=AUDIO_QUALITY_NORMAL
1800    )
1801    settings = backend._data_dir / "settings"
1802    settings.mkdir(parents=True)
1803    (settings / "prefs").write_text("audio.play_bitrate_non_metered_enumeration=5\n")
1804
1805    backend._write_audio_prefs()
1806
1807    prefs = (settings / "prefs").read_text().splitlines()
1808    assert "audio.play_bitrate_non_metered_enumeration=5" not in prefs
1809    assert "audio.play_bitrate_non_metered_enumeration=2" in prefs
1810
1811
1812def test_audio_prefs_crossfade_off_omits_the_time_key(tmp_path: Path) -> None:
1813    """
1814    Crossfade off writes crossfade_v2=false and no time key.
1815
1816    Sub-second time values silently disable crossfade, so the time key may only
1817    exist while crossfade is enabled.
1818    """
1819    backend = _prefs_backend(tmp_path, crossfade_ms=0, normalization=True)
1820
1821    backend._write_audio_prefs()
1822
1823    global_prefs = (backend._data_dir / "settings" / "prefs").read_text()
1824    assert "audio.crossfade_v2=false" in global_prefs
1825    assert "audio.crossfade.time_v2" not in global_prefs
1826    assert "audio.normalize_v2=true" in global_prefs
1827
1828
1829def test_audio_prefs_write_failure_is_non_fatal(tmp_path: Path) -> None:
1830    """A failing prefs write logs a warning instead of blocking the daemon spawn."""
1831    backend = _prefs_backend(tmp_path, crossfade_ms=8000, normalization=True)
1832    backend._data_dir = Path("/proc/no-such-place")
1833
1834    backend._write_audio_prefs()  # must not raise
1835
1836
1837def test_audio_prefs_corrupt_file_skips_only_that_store(tmp_path: Path) -> None:
1838    """
1839    A prefs file with invalid UTF-8 (truncated write) does not block the spawn.
1840
1841    Only the corrupt store is skipped; the remaining stores are still updated.
1842    """
1843    backend = _prefs_backend(tmp_path, crossfade_ms=8000, normalization=True)
1844    settings = backend._data_dir / "settings"
1845    (settings / "Users" / "alice-user").mkdir(parents=True)
1846    corrupt = b"core.clock_delta=0\naudio.play_bitrate\xc3"
1847    (settings / "prefs").write_bytes(corrupt)
1848
1849    backend._write_audio_prefs()  # must not raise
1850
1851    # the corrupt global store is left untouched, the per-user store is written
1852    assert (settings / "prefs").read_bytes() == corrupt
1853    user_prefs = (settings / "Users" / "alice-user" / "prefs").read_text()
1854    assert "audio.crossfade.time_v2=8000" in user_prefs
1855