/
/
/
1"""Tests for the shairport-sync metadata pipe reader."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import os
8import pathlib
9import time
10from collections.abc import Callable
11from typing import Any
12
13import pytest
14
15from music_assistant.providers.airplay_receiver.metadata import MetadataReader
16
17
18@pytest.fixture
19def pipe_path(tmp_path: pathlib.Path) -> str:
20 """Create a metadata FIFO like the provider does."""
21 path = str(tmp_path / "metadata_pipe")
22 os.mkfifo(path)
23 return path
24
25
26async def _write_marker(pipe_path: str, marker: str, timeout: float = 5.0) -> None:
27 """Write a marker using a short-lived FIFO writer, like a sessioncontrol hook does."""
28 # O_NONBLOCK so a regression leaving the FIFO readerless fails the test instead of
29 # hanging it: opening for write raises ENXIO while no reader is attached.
30 async with asyncio.timeout(timeout):
31 while True:
32 try:
33 fd = os.open(pipe_path, os.O_WRONLY | os.O_NONBLOCK)
34 break
35 except OSError:
36 await asyncio.sleep(0.05)
37 try:
38 os.write(fd, f"{marker}\n".encode())
39 finally:
40 os.close(fd)
41
42
43async def _wait_for(condition: Callable[[], bool], timeout: float = 2.0) -> None:
44 async with asyncio.timeout(timeout):
45 while not condition():
46 await asyncio.sleep(0.01)
47
48
49async def test_pipe_is_open_when_start_returns(pipe_path: str) -> None:
50 """start() attaches to the FIFO so hook writers never block on a missing reader."""
51 reader = MetadataReader(pipe_path, logging.getLogger("test"), None)
52 await reader.start()
53 try:
54 # Opening for write fails with ENXIO while no reader is attached.
55 fd = os.open(pipe_path, os.O_WRONLY | os.O_NONBLOCK)
56 os.close(fd)
57 finally:
58 await reader.stop()
59
60
61async def test_hook_marker_delivered(pipe_path: str) -> None:
62 """A sessioncontrol hook marker is delivered as a play_state update."""
63 updates: list[dict[str, Any]] = []
64 reader = MetadataReader(pipe_path, logging.getLogger("test"), updates.append)
65 await reader.start()
66 try:
67 await _write_marker(pipe_path, "MA_PLAY_BEGIN")
68 await _wait_for(lambda: bool(updates))
69 assert updates == [{"play_state": "playing"}]
70 finally:
71 await reader.stop()
72
73
74async def test_markers_after_writer_close(pipe_path: str) -> None:
75 """Accept markers from hook writers that connect after an earlier writer closed."""
76 updates: list[dict[str, Any]] = []
77 reader = MetadataReader(pipe_path, logging.getLogger("test"), updates.append)
78 await reader.start()
79 try:
80 await _write_marker(pipe_path, "MA_PLAY_BEGIN")
81 await _wait_for(lambda: len(updates) == 1)
82 await _write_marker(pipe_path, "MA_PLAY_END")
83 await _wait_for(lambda: len(updates) == 2)
84 assert updates == [{"play_state": "playing"}, {"play_state": "stopped"}]
85 finally:
86 await reader.stop()
87
88
89async def test_reader_stays_idle_after_writer_close(pipe_path: str) -> None:
90 """A FIFO with no writers left must not keep the event loop busy."""
91 updates: list[dict[str, Any]] = []
92 reader = MetadataReader(pipe_path, logging.getLogger("test"), updates.append)
93 await reader.start()
94 try:
95 await _write_marker(pipe_path, "MA_PLAY_BEGIN")
96 # The marker arriving proves the reader consumed it and the hook writer is gone.
97 await _wait_for(lambda: bool(updates))
98 cpu_before = time.process_time()
99 await asyncio.sleep(0.5)
100 # Only epoll reports a writerless FIFO readable forever, so this guard is
101 # meaningful on Linux and passes trivially on macOS/kqueue.
102 assert time.process_time() - cpu_before < 0.1
103 finally:
104 await reader.stop()
105
106
107async def test_reader_keeps_pipe_attached_across_restart(
108 pipe_path: str, monkeypatch: pytest.MonkeyPatch
109) -> None:
110 """An unexpected read error restarts the loop without ever detaching from the FIFO."""
111 updates: list[dict[str, Any]] = []
112 reader = MetadataReader(pipe_path, logging.getLogger("test"), updates.append)
113 real_read = os.read
114 fail_once = True
115
116 def flaky_read(fd: int, size: int) -> bytes:
117 nonlocal fail_once
118 if fd == reader._fd and fail_once:
119 fail_once = False
120 raise ValueError("unexpected error")
121 return real_read(fd, size)
122
123 monkeypatch.setattr(os, "read", flaky_read)
124 await reader.start()
125 try:
126 await _write_marker(pipe_path, "MA_PLAY_BEGIN")
127 await _wait_for(lambda: not fail_once)
128 # Well inside the restart backoff: a hook writer must still find a reader here.
129 await asyncio.sleep(0.2)
130 os.close(os.open(pipe_path, os.O_WRONLY | os.O_NONBLOCK))
131 await _write_marker(pipe_path, "MA_PLAY_END")
132 await _wait_for(lambda: updates[-1:] == [{"play_state": "stopped"}], timeout=5.0)
133 finally:
134 await reader.stop()
135