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