music-assistant-server

1"""Tests for the AsyncProcess helper."""
2
3from __future__ import annotations
4
5import asyncio
6import os
7import sys
8import time
9from collections.abc import AsyncGenerator
10from unittest.mock import MagicMock
11
12import pytest
13
14from music_assistant.helpers import process as process_module
15from music_assistant.helpers.process import AsyncProcess
16
17# Comfortably beyond the OS pipe capacity plus asyncio's default high-water mark,
18# so the bytes are guaranteed to still be queued in our own write buffer.
19_MORE_THAN_THE_PIPE_HOLDS = b"\x00" * (4 * 1024 * 1024)
20
21# Ignores SIGINT and keeps stdout open, so nothing but SIGKILL ends it and its
22# pipe never reaches EOF. The marker tells the test the handler is installed.
23_WEDGED_CHILD = (
24    "import signal, sys, time; signal.signal(signal.SIGINT, signal.SIG_IGN); "
25    "sys.stdout.write('ready\\n'); sys.stdout.flush(); time.sleep(30)"
26)
27
28# Writes more than a pipe buffer holds and never exits, so its output is still
29# undelivered when the process is killed.
30_NOISY_CHILD = (
31    "import sys, time; sys.stdout.write('x' * 300000); sys.stdout.flush(); time.sleep(30)"
32)
33
34
35@pytest.fixture(name="piped_process")
36async def piped_process_fixture() -> AsyncGenerator[tuple[AsyncProcess, int]]:
37    """
38    Yield an AsyncProcess writing to a real pipe, plus its unread read end.
39
40    The write side is a genuine asyncio pipe transport, so the flow control
41    stdin_quiesced relies on behaves exactly as it does against a real process,
42    while nothing consumes the read end until a test chooses to.
43    """
44    read_fd, write_fd = os.pipe()
45    loop = asyncio.get_running_loop()
46    transport, protocol = await loop.connect_write_pipe(
47        asyncio.streams.FlowControlMixin, os.fdopen(write_fd, "wb", 0)
48    )
49    writer = asyncio.StreamWriter(transport, protocol, None, loop)
50    proc = AsyncProcess(["cat"], stdin=True)
51    proc.proc = MagicMock(stdin=writer)
52    try:
53        yield proc, read_fd
54    finally:
55        transport.abort()
56        # abort() only schedules the callback that closes the write side, so let
57        # it run before the loop goes away rather than relying on the ordering.
58        await asyncio.sleep(0)
59        os.close(read_fd)
60
61
62def _queue_without_waiting(proc: AsyncProcess, data: bytes) -> None:
63    """
64    Hand data to stdin without awaiting it, leaving it in the write buffer.
65
66    ``write`` would block on its own drain while the pipe is unread, which is the
67    state these tests need to set up rather than something to wait through.
68    """
69    assert proc.proc is not None
70    assert proc.proc.stdin is not None
71    proc.proc.stdin.write(data)
72
73
74async def _consume(read_fd: int, total: int) -> None:
75    """Read `total` bytes off the pipe so the write buffer can empty."""
76    remaining = total
77    while remaining > 0:
78        remaining -= len(await asyncio.to_thread(os.read, read_fd, 65536))
79
80
81@pytest.mark.asyncio
82async def test_stdin_quiesced_waits_for_the_write_buffer_to_empty(
83    piped_process: tuple[AsyncProcess, int],
84) -> None:
85    """The block is entered only once every queued byte has reached the reader."""
86    proc, read_fd = piped_process
87    _queue_without_waiting(proc, _MORE_THAN_THE_PIPE_HOLDS)
88    assert proc.proc is not None
89    assert proc.proc.stdin is not None
90    assert proc.proc.stdin.transport.get_write_buffer_size() > 0
91    reader = asyncio.create_task(_consume(read_fd, len(_MORE_THAN_THE_PIPE_HOLDS)))
92
93    async with proc.stdin_quiesced() as quiesced:
94        assert quiesced is True
95        assert proc.proc.stdin.transport.get_write_buffer_size() == 0
96
97    await reader
98
99
100@pytest.mark.asyncio
101async def test_stdin_quiesced_reports_a_reader_that_never_catches_up(
102    piped_process: tuple[AsyncProcess, int],
103) -> None:
104    """A reader that never consumes the pipe reports failure instead of hanging."""
105    proc, _ = piped_process
106    _queue_without_waiting(proc, _MORE_THAN_THE_PIPE_HOLDS)
107
108    async with proc.stdin_quiesced(timeout=0.2) as quiesced:
109        assert quiesced is False
110
111
112@pytest.mark.asyncio
113async def test_stdin_quiesced_keeps_writes_out_of_the_block(
114    piped_process: tuple[AsyncProcess, int],
115) -> None:
116    """
117    No write can land while the block runs, so nothing queues up behind a drain.
118
119    This is the guarantee the block exists for: a caller telling the process
120    something about the bytes it has been handed needs stdin to stay as it left it
121    until the process has answered.
122    """
123    proc, read_fd = piped_process
124    reader = asyncio.create_task(_consume(read_fd, len(b"late")))
125
126    async with proc.stdin_quiesced() as quiesced:
127        assert quiesced is True
128        writing = asyncio.create_task(proc.write(b"late"))
129        await asyncio.sleep(0)
130
131        assert not writing.done()
132        assert proc.proc is not None
133        assert proc.proc.stdin is not None
134        assert proc.proc.stdin.transport.get_write_buffer_size() == 0
135
136    await writing
137    await reader
138
139
140@pytest.mark.asyncio
141async def test_stdin_quiesced_restores_normal_writing(
142    piped_process: tuple[AsyncProcess, int],
143) -> None:
144    """Writing carries on unaffected afterwards, on the restored buffer limits."""
145    proc, read_fd = piped_process
146    assert proc.proc is not None
147    assert proc.proc.stdin is not None
148    limits = proc.proc.stdin.transport.get_write_buffer_limits()
149    reader = asyncio.create_task(_consume(read_fd, len(b"first") + len(b"second")))
150    await proc.write(b"first")
151
152    async with proc.stdin_quiesced() as quiesced:
153        assert quiesced is True
154
155    assert proc.proc.stdin.transport.get_write_buffer_limits() == limits
156    await proc.write(b"second")
157    await reader
158
159
160@pytest.mark.asyncio
161async def test_stdin_quiesced_is_a_noop_without_a_process() -> None:
162    """A process that was never started quiesces trivially rather than raising."""
163    proc = AsyncProcess(["cat"], stdin=True)
164
165    async with proc.stdin_quiesced() as quiesced:
166        assert quiesced is True
167
168
169@pytest.mark.asyncio
170async def test_iter_stdout_drains_lines_buffered_after_exit() -> None:
171    """
172    Output written just before the process exits is still delivered.
173
174    A short-lived process can write everything and be reaped before the reader
175    runs, so keying the stdout reader off the returncode would drop exactly the
176    output that explains why it exited.
177    """
178    proc = AsyncProcess(
179        ["sh", "-c", "for i in $(seq 1 50); do echo line$i; done"],
180        stdout=True,
181        stderr=asyncio.subprocess.STDOUT,
182    )
183    await proc.start()
184    await proc.wait()
185
186    lines = [line async for line in proc.iter_stdout()]
187
188    assert lines == [f"line{index}" for index in range(1, 51)]
189    await proc.close()
190
191
192@pytest.mark.asyncio
193async def test_write_eof_marks_stdin_closed_while_the_process_lives() -> None:
194    """
195    A process that was sent EOF keeps running but can no longer be fed.
196
197    Writing EOF closes the pipe for good, so a caller that hands the write end to
198    someone else (or means to keep feeding it) has to be able to tell.
199    """
200    proc = AsyncProcess(["cat"], stdin=True, stdout=True)
201    await proc.start()
202    await proc.write(b"hello\n")
203
204    await proc.write_eof()
205
206    assert proc.stdin_closed
207    assert not proc.closed
208    assert proc.returncode is None
209    # neither a further write nor a second EOF may reach the closed pipe
210    await proc.write(b"more\n")
211    await proc.write_eof()
212    assert await proc.read_stdout() == b"hello\n"
213    await proc.close()
214
215
216@pytest.mark.asyncio
217async def test_read_stdout_stops_once_the_process_is_closed() -> None:
218    """A closed process reports EOF instead of waiting on a stream it no longer owns."""
219    proc = AsyncProcess(["sh", "-c", "sleep 30"], stdout=True, stderr=asyncio.subprocess.STDOUT)
220    await proc.start()
221    await proc.close()
222
223    assert await proc.read_stdout() == b""
224
225
226@pytest.mark.asyncio
227async def test_second_close_returns_without_waiting_out_the_stream_locks() -> None:
228    """
229    Closing an already-closed process is cheap.
230
231    close() keeps the stdin/stdout locks it takes, so a second call used to sit
232    through both 5s acquire timeouts - a delay paid on every supervised restart
233    that closes the process before its own cleanup runs.
234    """
235    proc = AsyncProcess(["sh", "-c", "sleep 30"], stdout=True, stderr=asyncio.subprocess.STDOUT)
236    await proc.start()
237    await proc.close()
238
239    started = time.monotonic()
240    await proc.close()
241
242    assert time.monotonic() - started < 1
243
244
245@pytest.mark.asyncio
246async def test_kill_retrieves_the_exception_of_a_finished_stdin_feeder(
247    caplog: pytest.LogCaptureFixture,
248) -> None:
249    """
250    A stdin feeder that already failed is awaited and its failure logged.
251
252    Awaiting only a still-pending task leaves the exception of one that already
253    ended unretrieved, which asyncio reports as unhandled when it is collected;
254    retrieving it without logging would drop the only trace of the failure.
255    """
256
257    async def _failing_feeder() -> None:
258        raise RuntimeError("feeder blew up")
259
260    proc = AsyncProcess(["sh", "-c", "sleep 30"], stdout=True, stderr=asyncio.subprocess.STDOUT)
261    await proc.start()
262    feeder = asyncio.create_task(_failing_feeder())
263    await asyncio.wait([feeder])  # let it fail without retrieving the exception
264    proc._stdin_feeder_task = feeder
265
266    await proc.kill()
267
268    # asyncio clears this flag once the exception has been retrieved; while it is
269    # set the task is the one that triggers "Task exception was never retrieved"
270    assert feeder._log_traceback is False
271    assert "feeder blew up" in caplog.text
272
273
274@pytest.mark.asyncio
275async def test_kill_returns_promptly_with_output_left_in_the_pipes() -> None:
276    """
277    Killing a process whose pipes still hold output returns without delay.
278
279    The reap only completes once every pipe has disconnected, and nothing reads
280    them after a kill, so the pipes have to be drained for it to finish.
281    """
282    proc = AsyncProcess([sys.executable, "-c", _NOISY_CHILD], stdout=True, stderr=True)
283    await proc.start()
284    await asyncio.sleep(0.5)
285
286    started = time.monotonic()
287    await proc.kill()
288
289    assert proc.returncode is not None
290    assert time.monotonic() - started < 1
291
292
293@pytest.mark.asyncio
294async def test_close_reaps_a_child_that_never_closes_its_pipes(
295    monkeypatch: pytest.MonkeyPatch,
296) -> None:
297    """
298    A child holding its pipes open must not keep close() from reaping it.
299
300    Draining stdout is what lets a healthy process flush before it is reaped, so
301    an unbounded drain waits out a wedged child forever and the terminate/SIGKILL
302    escalation is never reached.
303    """
304    monkeypatch.setattr(process_module, "PIPE_DRAIN_TIMEOUT", 0.2)
305    proc = AsyncProcess(
306        [sys.executable, "-c", _WEDGED_CHILD], stdout=True, stderr=asyncio.subprocess.STDOUT
307    )
308    await proc.start()
309    assert await proc.read_stdout() == b"ready\n"
310
311    async with asyncio.timeout(20):
312        await proc.close()
313
314    assert proc.returncode is not None
315
316
317@pytest.mark.asyncio
318async def test_close_reaps_a_child_when_cancelled_mid_drain() -> None:
319    """
320    Cancellation landing while a pipe is draining must not leave the child running.
321
322    Walking away there skips the terminate/SIGKILL escalation, and nothing else
323    ever comes back for the process.
324    """
325    proc = AsyncProcess(
326        [sys.executable, "-c", _WEDGED_CHILD], stdout=True, stderr=asyncio.subprocess.STDOUT
327    )
328    await proc.start()
329    assert await proc.read_stdout() == b"ready\n"
330
331    # well inside PIPE_DRAIN_TIMEOUT, so the cancellation lands on the stdout drain
332    with pytest.raises(TimeoutError):
333        async with asyncio.timeout(0.5):
334            await proc.close()
335
336    assert proc.returncode is not None
337
338
339@pytest.mark.asyncio
340async def test_close_reaps_a_child_when_cancelled_while_waiting_for_exit(
341    monkeypatch: pytest.MonkeyPatch,
342) -> None:
343    """
344    Cancellation landing while waiting for the process to exit must still reap it.
345
346    That wait is where close() spends most of its time, so it is the likeliest place
347    for a cancellation to land, and giving up there skips the SIGKILL escalation.
348    """
349    # short enough that the drain is over well before the cancellation below
350    monkeypatch.setattr(process_module, "PIPE_DRAIN_TIMEOUT", 0.2)
351    proc = AsyncProcess(
352        [sys.executable, "-c", _WEDGED_CHILD], stdout=True, stderr=asyncio.subprocess.STDOUT
353    )
354    await proc.start()
355    assert await proc.read_stdout() == b"ready\n"
356
357    with pytest.raises(TimeoutError):
358        async with asyncio.timeout(0.5):
359            await proc.close()
360
361    assert proc.returncode is not None
362