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