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