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        # PA writes an auth cookie under HOME and refuses to start when it is
187        # unwritable, as it is when the container runs as a uid with no passwd entry
188        assert proc.env["HOME"] == private_dir
189        # the generated config loads only the native protocol on the private socket
190        config_text = (Path(private_dir) / "pulse_capture.pa").read_text(encoding="utf-8")
191        assert config_text.startswith("load-module module-native-protocol-unix ")
192        assert f"socket={private_dir}/native" in config_text
193        assert "auth-anonymous=1" in config_text
194
195        child_env = server.child_env("my_sink")
196        assert child_env["PULSE_SERVER"] == f"unix:{private_dir}/native"
197        assert child_env["PULSE_SINK"] == "my_sink"
198        # merged over the regular subprocess env, not replacing it
199        assert child_env["PATH"] == os.environ["PATH"]
200        assert dict(os.environ) == environ_before
201    finally:
202        await server.release()
203
204
205async def test_unique_sinks_and_module_args(server: PulseCaptureServer) -> None:
206    """Each sink gets a unique name/FIFO and correct module-pipe-sink args."""
207    await server.acquire()
208    try:
209        sink1 = await PipeSink.create(server, "soloist")
210        sink2 = await PipeSink.create(server, "soloist")
211        assert sink1.sink_name != sink2.sink_name
212        assert sink1.fifo_path != sink2.fifo_path
213        assert sink1.sink_name.startswith("soloist_")
214        assert sink1.fifo_path.parent == Path(server.server_address.removeprefix("unix:")).parent
215
216        controller = FakeVolumeController.instances[-1]
217        module_name, argument = controller.load_module_calls[0]
218        assert module_name == "module-pipe-sink"
219        assert f"sink_name={sink1.sink_name}" in argument
220        assert f"file={sink1.fifo_path}" in argument
221        assert "format=s32le" in argument
222        assert "rate=44100" in argument
223        assert "channels=2" in argument
224    finally:
225        await server.release()
226
227
228async def test_set_volume_forwards_raw_percentage(server: PulseCaptureServer) -> None:
229    """set_volume passes the raw percentage (also above 100) to the controller."""
230    await server.acquire()
231    try:
232        sink = await PipeSink.create(server, "soloist")
233        await sink.set_volume(400.0)
234        await sink.set_volume(85.5)
235        controller = FakeVolumeController.instances[-1]
236        assert controller.volume_calls == [(sink.sink_name, 400.0), (sink.sink_name, 85.5)]
237    finally:
238        await server.release()
239
240
241def test_set_sink_volume_raw_mapping(monkeypatch: pytest.MonkeyPatch) -> None:
242    """The raw setter maps percentage linearly onto PA_VOLUME_NORM, with clamping."""
243    controller = pulse_capture.PAVolumeController.__new__(pulse_capture.PAVolumeController)
244    applied: list[tuple[str, int, int]] = []
245
246    def fake_apply(sink_name: str, pa_volume: int, channels: int) -> bool:
247        applied.append((sink_name, pa_volume, channels))
248        return True
249
250    monkeypatch.setattr(controller, "_apply_sink_volume", fake_apply)
251
252    assert controller.set_sink_volume_raw("sink", 100.0) is True
253    assert applied[-1] == ("sink", PA_VOLUME_NORM, 2)
254    # linear (no taper): 50% is exactly half of PA_VOLUME_NORM
255    controller.set_sink_volume_raw("sink", 50.0)
256    assert applied[-1][1] == PA_VOLUME_NORM // 2
257    # values above 100% amplify linearly (reciprocal cubic compensation)
258    controller.set_sink_volume_raw("sink", 400.0)
259    assert applied[-1][1] == PA_VOLUME_NORM * 4
260    # clamped to the sane maximum and to zero
261    controller.set_sink_volume_raw("sink", 1_000_000.0)
262    assert applied[-1][1] == round(PA_VOLUME_NORM * MAX_RAW_VOLUME_PCT / 100.0)
263    controller.set_sink_volume_raw("sink", -5.0)
264    assert applied[-1][1] == 0
265
266
267async def test_suspend_resume_use_pactl(
268    server: PulseCaptureServer, monkeypatch: pytest.MonkeyPatch
269) -> None:
270    """suspend/resume invoke pactl against the private server."""
271    calls: list[tuple[tuple[str, ...], dict[str, str] | None, float | None]] = []
272
273    async def fake_check_output(
274        *args: str, env: dict[str, str] | None = None, timeout: float | None = None
275    ) -> tuple[int, bytes]:
276        calls.append((args, env, timeout))
277        return 0, b""
278
279    monkeypatch.setattr(pulse_capture, "check_output", fake_check_output)
280    await server.acquire()
281    try:
282        sink = await PipeSink.create(server, "soloist")
283        await sink.suspend()
284        await sink.resume()
285        assert calls[0][0] == (
286            "pactl",
287            "--server",
288            server.server_address,
289            "suspend-sink",
290            sink.sink_name,
291            "1",
292        )
293        assert calls[1][0][-1] == "0"
294        assert calls[0][1] is not None
295        assert calls[0][1]["PULSE_SERVER"] == server.server_address
296    finally:
297        await server.release()
298
299
300async def test_unload_idempotent_and_fifo_cleanup(server: PulseCaptureServer) -> None:
301    """Unload removes the module once and cleans up the FIFO file."""
302    await server.acquire()
303    try:
304        sink = await PipeSink.create(server, "soloist")
305        # the real pipe-sink module creates the FIFO; simulate that
306        sink.fifo_path.touch()
307
308        await sink.unload()
309        controller = FakeVolumeController.instances[-1]
310        assert controller.unload_module_calls == [1]
311        assert not sink.fifo_path.exists()
312
313        await sink.unload()
314        assert controller.unload_module_calls == [1]
315    finally:
316        await server.release()
317
318
319async def test_crash_restart_bumps_generation(server: PulseCaptureServer) -> None:
320    """An unexpected daemon exit triggers a restart with a new generation."""
321    await server.acquire()
322    try:
323        stale_sink = await PipeSink.create(server, "soloist")
324        assert server.generation == 1
325        old_controller = FakeVolumeController.instances[-1]
326
327        FakeAsyncProcess.instances[0].crash()
328        await _wait_for(lambda: server.generation == 2)
329
330        # a fresh daemon and controller are in place, the old controller is gone
331        assert len(FakeAsyncProcess.instances) == 2
332        assert not FakeAsyncProcess.instances[1].closed
333        assert old_controller.closed
334        new_controller = FakeVolumeController.instances[-1]
335        assert new_controller is not old_controller
336
337        # a sink from before the crash skips the module unload (module died
338        # with the old daemon; the index would target the wrong module now)
339        await stale_sink.unload()
340        assert not new_controller.unload_module_calls
341
342        # creating a sink against the restarted daemon works
343        sink = await PipeSink.create(server, "soloist")
344        assert new_controller.load_module_calls
345        assert sink.sink_name.startswith("soloist_")
346    finally:
347        await server.release()
348
349
350async def test_release_during_restart_backoff(
351    server: PulseCaptureServer, monkeypatch: pytest.MonkeyPatch
352) -> None:
353    """Releasing while the supervisor waits to restart stops cleanly."""
354    monkeypatch.setattr(pulse_capture, "_RESTART_BACKOFF_INITIAL", 60.0)
355    await server.acquire()
356    proc = FakeAsyncProcess.instances[0]
357    proc.crash()
358    # give the supervisor a chance to enter its backoff sleep
359    await asyncio.sleep(0.05)
360    await server.release()
361    assert server.generation == 1
362    assert len(FakeAsyncProcess.instances) == 1
363
364
365def test_get_pulse_capture_server_is_shared(tmp_path: Path) -> None:
366    """The accessor returns one shared instance per mass object."""
367    mass = Mock()
368    mass.cache_path = str(tmp_path)
369    first = pulse_capture.get_pulse_capture_server(mass)
370    assert pulse_capture.get_pulse_capture_server(mass) is first
371    other_mass = Mock()
372    other_mass.cache_path = str(tmp_path)
373    assert pulse_capture.get_pulse_capture_server(other_mass) is not first
374    # a second mass must not evict the first mass's server (it may hold a
375    # running daemon that would otherwise leak unreferenced)
376    assert pulse_capture.get_pulse_capture_server(mass) is first
377