/
/
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_read_stdout_stops_once_the_process_is_closed() -> None:
194 """A closed process reports EOF instead of waiting on a stream it no longer owns."""
195 proc = AsyncProcess(["sh", "-c", "sleep 30"], stdout=True, stderr=asyncio.subprocess.STDOUT)
196 await proc.start()
197 await proc.close()
198
199 assert await proc.read_stdout() == b""
200
201
202@pytest.mark.asyncio
203async def test_second_close_returns_without_waiting_out_the_stream_locks() -> None:
204 """
205 Closing an already-closed process is cheap.
206
207 close() keeps the stdin/stdout locks it takes, so a second call used to sit
208 through both 5s acquire timeouts - a delay paid on every supervised restart
209 that closes the process before its own cleanup runs.
210 """
211 proc = AsyncProcess(["sh", "-c", "sleep 30"], stdout=True, stderr=asyncio.subprocess.STDOUT)
212 await proc.start()
213 await proc.close()
214
215 started = time.monotonic()
216 await proc.close()
217
218 assert time.monotonic() - started < 1
219
220
221@pytest.mark.asyncio
222async def test_kill_retrieves_the_exception_of_a_finished_stdin_feeder(
223 caplog: pytest.LogCaptureFixture,
224) -> None:
225 """
226 A stdin feeder that already failed is awaited and its failure logged.
227
228 Awaiting only a still-pending task leaves the exception of one that already
229 ended unretrieved, which asyncio reports as unhandled when it is collected;
230 retrieving it without logging would drop the only trace of the failure.
231 """
232
233 async def _failing_feeder() -> None:
234 raise RuntimeError("feeder blew up")
235
236 proc = AsyncProcess(["sh", "-c", "sleep 30"], stdout=True, stderr=asyncio.subprocess.STDOUT)
237 await proc.start()
238 feeder = asyncio.create_task(_failing_feeder())
239 await asyncio.wait([feeder]) # let it fail without retrieving the exception
240 proc._stdin_feeder_task = feeder
241
242 await proc.kill()
243
244 # asyncio clears this flag once the exception has been retrieved; while it is
245 # set the task is the one that triggers "Task exception was never retrieved"
246 assert feeder._log_traceback is False
247 assert "feeder blew up" in caplog.text
248
249
250@pytest.mark.asyncio
251async def test_kill_returns_promptly_with_output_left_in_the_pipes() -> None:
252 """
253 Killing a process whose pipes still hold output returns without delay.
254
255 The reap only completes once every pipe has disconnected, and nothing reads
256 them after a kill, so the pipes have to be drained for it to finish.
257 """
258 proc = AsyncProcess([sys.executable, "-c", _NOISY_CHILD], stdout=True, stderr=True)
259 await proc.start()
260 await asyncio.sleep(0.5)
261
262 started = time.monotonic()
263 await proc.kill()
264
265 assert proc.returncode is not None
266 assert time.monotonic() - started < 1
267
268
269@pytest.mark.asyncio
270async def test_close_reaps_a_child_that_never_closes_its_pipes(
271 monkeypatch: pytest.MonkeyPatch,
272) -> None:
273 """
274 A child holding its pipes open must not keep close() from reaping it.
275
276 Draining stdout is what lets a healthy process flush before it is reaped, so
277 an unbounded drain waits out a wedged child forever and the terminate/SIGKILL
278 escalation is never reached.
279 """
280 monkeypatch.setattr(process_module, "PIPE_DRAIN_TIMEOUT", 0.2)
281 proc = AsyncProcess(
282 [sys.executable, "-c", _WEDGED_CHILD], stdout=True, stderr=asyncio.subprocess.STDOUT
283 )
284 await proc.start()
285 assert await proc.read_stdout() == b"ready\n"
286
287 async with asyncio.timeout(20):
288 await proc.close()
289
290 assert proc.returncode is not None
291
292
293@pytest.mark.asyncio
294async def test_close_reaps_a_child_when_cancelled_mid_drain() -> None:
295 """
296 Cancellation landing while a pipe is draining must not leave the child running.
297
298 Walking away there skips the terminate/SIGKILL escalation, and nothing else
299 ever comes back for the process.
300 """
301 proc = AsyncProcess(
302 [sys.executable, "-c", _WEDGED_CHILD], stdout=True, stderr=asyncio.subprocess.STDOUT
303 )
304 await proc.start()
305 assert await proc.read_stdout() == b"ready\n"
306
307 # well inside PIPE_DRAIN_TIMEOUT, so the cancellation lands on the stdout drain
308 with pytest.raises(TimeoutError):
309 async with asyncio.timeout(0.5):
310 await proc.close()
311
312 assert proc.returncode is not None
313
314
315@pytest.mark.asyncio
316async def test_close_reaps_a_child_when_cancelled_while_waiting_for_exit(
317 monkeypatch: pytest.MonkeyPatch,
318) -> None:
319 """
320 Cancellation landing while waiting for the process to exit must still reap it.
321
322 That wait is where close() spends most of its time, so it is the likeliest place
323 for a cancellation to land, and giving up there skips the SIGKILL escalation.
324 """
325 # short enough that the drain is over well before the cancellation below
326 monkeypatch.setattr(process_module, "PIPE_DRAIN_TIMEOUT", 0.2)
327 proc = AsyncProcess(
328 [sys.executable, "-c", _WEDGED_CHILD], stdout=True, stderr=asyncio.subprocess.STDOUT
329 )
330 await proc.start()
331 assert await proc.read_stdout() == b"ready\n"
332
333 with pytest.raises(TimeoutError):
334 async with asyncio.timeout(0.5):
335 await proc.close()
336
337 assert proc.returncode is not None
338