music-assistant-server

1"""
2Tests for the pulse_capture helper.
3
4No real PulseAudio is involved: the daemon process and the libpulse controller
5are replaced by fakes patched into the pulse_capture module namespace. The fake
6daemon creates the native socket file on start so readiness polling completes,
7and the fake controller records every load_module/unload_module/volume call.
8"""
9
10from __future__ import annotations
11
12import asyncio
13import os
14from collections.abc import AsyncGenerator, Callable
15from pathlib import Path
16from typing import ClassVar
17from unittest.mock import Mock
18
19import pytest
20
21from music_assistant.helpers import pulse_capture
22from music_assistant.helpers.pulse_capture import (
23    MAX_RAW_VOLUME_PCT,
24    PA_VOLUME_NORM,
25    PipeSink,
26    PulseCaptureServer,
27)
28
29
30class FakeAsyncProcess:
31    """Fake AsyncProcess that mimics the pulseaudio daemon."""
32
33    instances: ClassVar[list[FakeAsyncProcess]] = []
34
35    def __init__(
36        self,
37        args: list[str],
38        stdin: bool | int | None = None,
39        stdout: bool | int | None = None,
40        stderr: bool | int | None = False,
41        name: str | None = None,
42        env: dict[str, str] | None = None,
43    ) -> None:
44        """Record the launch parameters."""
45        self.args = args
46        self.name = name
47        self.env = env or {}
48        self.returncode: int | None = None
49        self.closed = False
50        self._exited = asyncio.Event()
51        FakeAsyncProcess.instances.append(self)
52
53    async def start(self) -> None:
54        """Create the native socket file, like the real daemon would."""
55        runtime_dir = Path(self.env["PULSE_RUNTIME_PATH"])
56        runtime_dir.mkdir(parents=True, exist_ok=True)
57        (runtime_dir / "native").touch()
58
59    async def iter_stderr(self) -> AsyncGenerator[str]:
60        """Yield nothing and block until the daemon exits."""
61        await self._exited.wait()
62        lines: list[str] = []
63        for line in lines:
64            yield line
65
66    async def close(self) -> None:
67        """Terminate the fake daemon."""
68        self.closed = True
69        if self.returncode is None:
70            self.returncode = 0
71        self._exited.set()
72
73    def crash(self) -> None:
74        """Simulate an unexpected daemon exit."""
75        self.returncode = 1
76        self._exited.set()
77
78
79class FakeVolumeController:
80    """Fake PAVolumeController recording all calls."""
81
82    instances: ClassVar[list[FakeVolumeController]] = []
83
84    def __init__(self, server: str | None = None) -> None:
85        """Record the server address used to connect."""
86        self.server = server
87        self.load_module_calls: list[tuple[str, str]] = []
88        self.unload_module_calls: list[int] = []
89        self.volume_calls: list[tuple[str, float]] = []
90        self.closed = False
91        self._next_module_index = 1
92        FakeVolumeController.instances.append(self)
93
94    def load_module(self, module_name: str, argument: str) -> int | None:
95        """Record the call and hand out a unique module index."""
96        self.load_module_calls.append((module_name, argument))
97        index = self._next_module_index
98        self._next_module_index += 1
99        return index
100
101    def unload_module(self, module_index: int) -> bool:
102        """Record the call."""
103        self.unload_module_calls.append(module_index)
104        return True
105
106    def set_sink_volume_raw(self, sink_name: str, volume_pct: float, channels: int = 2) -> bool:
107        """Record the call."""
108        self.volume_calls.append((sink_name, volume_pct))
109        return True
110
111    def close(self) -> None:
112        """Mark the controller closed."""
113        self.closed = True
114
115
116@pytest.fixture
117def server(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> PulseCaptureServer:
118    """Return a PulseCaptureServer wired to the fakes, on a tmp cache dir."""
119    FakeAsyncProcess.instances = []
120    FakeVolumeController.instances = []
121    monkeypatch.setattr(pulse_capture, "AsyncProcess", FakeAsyncProcess)
122    monkeypatch.setattr(pulse_capture, "PAVolumeController", FakeVolumeController)
123    monkeypatch.setattr(pulse_capture, "_READY_POLL_INTERVAL", 0.01)
124    monkeypatch.setattr(pulse_capture, "_RESTART_BACKOFF_INITIAL", 0.01)
125    mass = Mock()
126    mass.cache_path = str(tmp_path)
127    return PulseCaptureServer(mass)
128
129
130async def _wait_for(condition: Callable[[], bool], timeout: float = 2.0) -> None:
131    """Poll until condition holds or the timeout expires."""
132    async with asyncio.timeout(timeout):
133        while not condition():
134            await asyncio.sleep(0.01)
135
136
137async def test_refcounted_lifecycle(server: PulseCaptureServer) -> None:
138    """First acquire starts the daemon, last release stops it, double release is safe."""
139    assert not FakeAsyncProcess.instances
140
141    await server.acquire()
142    assert len(FakeAsyncProcess.instances) == 1
143    proc = FakeAsyncProcess.instances[0]
144    assert not proc.closed
145    assert proc.args[0] == "pulseaudio"
146    assert "--daemonize=no" in proc.args
147    assert "-n" in proc.args
148    assert "--exit-idle-time=-1" in proc.args
149    assert server.generation == 1
150    # connectivity was verified against the private socket
151    assert FakeVolumeController.instances[0].server == server.server_address
152
153    # second acquire does not start another daemon
154    await server.acquire()
155    assert len(FakeAsyncProcess.instances) == 1
156
157    # first release keeps the daemon running
158    await server.release()
159    assert not proc.closed
160
161    # last release stops the daemon and removes the private dir
162    # (read via the class list: mypy would consider a re-read of the earlier
163    # `assert not proc.closed`-narrowed expression unreachable)
164    await server.release()
165    assert FakeAsyncProcess.instances[0].closed
166    assert FakeVolumeController.instances[0].closed
167    assert not Path(server.server_address.removeprefix("unix:")).parent.exists()
168
169    # extra release is a no-op
170    await server.release()
171    assert len(FakeAsyncProcess.instances) == 1
172
173
174async def test_daemon_env_isolation(server: PulseCaptureServer, tmp_path: Path) -> None:
175    """The daemon env is private and os.environ is never mutated."""
176    environ_before = dict(os.environ)
177    await server.acquire()
178    try:
179        assert dict(os.environ) == environ_before
180
181        private_dir = str(tmp_path / "pulse_capture")
182        proc = FakeAsyncProcess.instances[0]
183        assert proc.env["XDG_RUNTIME_DIR"] == private_dir
184        assert proc.env["PULSE_RUNTIME_PATH"] == private_dir
185        assert proc.env["PULSE_STATE_PATH"] == private_dir
186        # the generated config loads only the native protocol on the private socket
187        config_text = (Path(private_dir) / "pulse_capture.pa").read_text(encoding="utf-8")
188        assert config_text.startswith("load-module module-native-protocol-unix ")
189        assert f"socket={private_dir}/native" in config_text
190        assert "auth-anonymous=1" in config_text
191
192        child_env = server.child_env("my_sink")
193        assert child_env["PULSE_SERVER"] == f"unix:{private_dir}/native"
194        assert child_env["PULSE_SINK"] == "my_sink"
195        # merged over the regular subprocess env, not replacing it
196        assert child_env["PATH"] == os.environ["PATH"]
197        assert dict(os.environ) == environ_before
198    finally:
199        await server.release()
200
201
202async def test_unique_sinks_and_module_args(server: PulseCaptureServer) -> None:
203    """Each sink gets a unique name/FIFO and correct module-pipe-sink args."""
204    await server.acquire()
205    try:
206        sink1 = await PipeSink.create(server, "soloist")
207        sink2 = await PipeSink.create(server, "soloist")
208        assert sink1.sink_name != sink2.sink_name
209        assert sink1.fifo_path != sink2.fifo_path
210        assert sink1.sink_name.startswith("soloist_")
211        assert sink1.fifo_path.parent == Path(server.server_address.removeprefix("unix:")).parent
212
213        controller = FakeVolumeController.instances[-1]
214        module_name, argument = controller.load_module_calls[0]
215        assert module_name == "module-pipe-sink"
216        assert f"sink_name={sink1.sink_name}" in argument
217        assert f"file={sink1.fifo_path}" in argument
218        assert "format=s32le" in argument
219        assert "rate=44100" in argument
220        assert "channels=2" in argument
221    finally:
222        await server.release()
223
224
225async def test_set_volume_forwards_raw_percentage(server: PulseCaptureServer) -> None:
226    """set_volume passes the raw percentage (also above 100) to the controller."""
227    await server.acquire()
228    try:
229        sink = await PipeSink.create(server, "soloist")
230        await sink.set_volume(400.0)
231        await sink.set_volume(85.5)
232        controller = FakeVolumeController.instances[-1]
233        assert controller.volume_calls == [(sink.sink_name, 400.0), (sink.sink_name, 85.5)]
234    finally:
235        await server.release()
236
237
238def test_set_sink_volume_raw_mapping(monkeypatch: pytest.MonkeyPatch) -> None:
239    """The raw setter maps percentage linearly onto PA_VOLUME_NORM, with clamping."""
240    controller = pulse_capture.PAVolumeController.__new__(pulse_capture.PAVolumeController)
241    applied: list[tuple[str, int, int]] = []
242
243    def fake_apply(sink_name: str, pa_volume: int, channels: int) -> bool:
244        applied.append((sink_name, pa_volume, channels))
245        return True
246
247    monkeypatch.setattr(controller, "_apply_sink_volume", fake_apply)
248
249    assert controller.set_sink_volume_raw("sink", 100.0) is True
250    assert applied[-1] == ("sink", PA_VOLUME_NORM, 2)
251    # linear (no taper): 50% is exactly half of PA_VOLUME_NORM
252    controller.set_sink_volume_raw("sink", 50.0)
253    assert applied[-1][1] == PA_VOLUME_NORM // 2
254    # values above 100% amplify linearly (reciprocal cubic compensation)
255    controller.set_sink_volume_raw("sink", 400.0)
256    assert applied[-1][1] == PA_VOLUME_NORM * 4
257    # clamped to the sane maximum and to zero
258    controller.set_sink_volume_raw("sink", 1_000_000.0)
259    assert applied[-1][1] == round(PA_VOLUME_NORM * MAX_RAW_VOLUME_PCT / 100.0)
260    controller.set_sink_volume_raw("sink", -5.0)
261    assert applied[-1][1] == 0
262
263
264async def test_suspend_resume_use_pactl(
265    server: PulseCaptureServer, monkeypatch: pytest.MonkeyPatch
266) -> None:
267    """suspend/resume invoke pactl against the private server."""
268    calls: list[tuple[tuple[str, ...], dict[str, str] | None, float | None]] = []
269
270    async def fake_check_output(
271        *args: str, env: dict[str, str] | None = None, timeout: float | None = None
272    ) -> tuple[int, bytes]:
273        calls.append((args, env, timeout))
274        return 0, b""
275
276    monkeypatch.setattr(pulse_capture, "check_output", fake_check_output)
277    await server.acquire()
278    try:
279        sink = await PipeSink.create(server, "soloist")
280        await sink.suspend()
281        await sink.resume()
282        assert calls[0][0] == (
283            "pactl",
284            "--server",
285            server.server_address,
286            "suspend-sink",
287            sink.sink_name,
288            "1",
289        )
290        assert calls[1][0][-1] == "0"
291        assert calls[0][1] is not None
292        assert calls[0][1]["PULSE_SERVER"] == server.server_address
293    finally:
294        await server.release()
295
296
297async def test_unload_idempotent_and_fifo_cleanup(server: PulseCaptureServer) -> None:
298    """Unload removes the module once and cleans up the FIFO file."""
299    await server.acquire()
300    try:
301        sink = await PipeSink.create(server, "soloist")
302        # the real pipe-sink module creates the FIFO; simulate that
303        sink.fifo_path.touch()
304
305        await sink.unload()
306        controller = FakeVolumeController.instances[-1]
307        assert controller.unload_module_calls == [1]
308        assert not sink.fifo_path.exists()
309
310        await sink.unload()
311        assert controller.unload_module_calls == [1]
312    finally:
313        await server.release()
314
315
316async def test_crash_restart_bumps_generation(server: PulseCaptureServer) -> None:
317    """An unexpected daemon exit triggers a restart with a new generation."""
318    await server.acquire()
319    try:
320        stale_sink = await PipeSink.create(server, "soloist")
321        assert server.generation == 1
322        old_controller = FakeVolumeController.instances[-1]
323
324        FakeAsyncProcess.instances[0].crash()
325        await _wait_for(lambda: server.generation == 2)
326
327        # a fresh daemon and controller are in place, the old controller is gone
328        assert len(FakeAsyncProcess.instances) == 2
329        assert not FakeAsyncProcess.instances[1].closed
330        assert old_controller.closed
331        new_controller = FakeVolumeController.instances[-1]
332        assert new_controller is not old_controller
333
334        # a sink from before the crash skips the module unload (module died
335        # with the old daemon; the index would target the wrong module now)
336        await stale_sink.unload()
337        assert not new_controller.unload_module_calls
338
339        # creating a sink against the restarted daemon works
340        sink = await PipeSink.create(server, "soloist")
341        assert new_controller.load_module_calls
342        assert sink.sink_name.startswith("soloist_")
343    finally:
344        await server.release()
345
346
347async def test_release_during_restart_backoff(
348    server: PulseCaptureServer, monkeypatch: pytest.MonkeyPatch
349) -> None:
350    """Releasing while the supervisor waits to restart stops cleanly."""
351    monkeypatch.setattr(pulse_capture, "_RESTART_BACKOFF_INITIAL", 60.0)
352    await server.acquire()
353    proc = FakeAsyncProcess.instances[0]
354    proc.crash()
355    # give the supervisor a chance to enter its backoff sleep
356    await asyncio.sleep(0.05)
357    await server.release()
358    assert server.generation == 1
359    assert len(FakeAsyncProcess.instances) == 1
360
361
362def test_get_pulse_capture_server_is_shared(tmp_path: Path) -> None:
363    """The accessor returns one shared instance per mass object."""
364    mass = Mock()
365    mass.cache_path = str(tmp_path)
366    first = pulse_capture.get_pulse_capture_server(mass)
367    assert pulse_capture.get_pulse_capture_server(mass) is first
368    other_mass = Mock()
369    other_mass.cache_path = str(tmp_path)
370    assert pulse_capture.get_pulse_capture_server(other_mass) is not first
371    # a second mass must not evict the first mass's server (it may hold a
372    # running daemon that would otherwise leak unreferenced)
373    assert pulse_capture.get_pulse_capture_server(mass) is first
374