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