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