/
/
/
1"""Helpers for replaying DACP requests through the AirPlay handler."""
2
3from __future__ import annotations
4
5import asyncio
6import base64
7import json
8from unittest.mock import AsyncMock, MagicMock
9
10from music_assistant.providers.airplay.provider import AirPlayProvider
11
12
13def build_dacp_request(
14 path: str,
15 active_remote: str | None = "123",
16 *,
17 method: str = "POST",
18 body: str = "",
19 headers: dict[str, str] | None = None,
20) -> bytes:
21 """
22 Build raw DACP request bytes for replay.
23
24 :param path: Request path, e.g. "/ctrl-int/1/play".
25 :param active_remote: Active-Remote header value; omitted from the request when None.
26 :param method: HTTP method.
27 :param body: Request body.
28 :param headers: Extra headers to include.
29 """
30 hdrs: dict[str, str] = {}
31 if active_remote is not None:
32 hdrs["Active-Remote"] = active_remote
33 if headers:
34 hdrs.update(headers)
35 lines = [f"{method} {path} HTTP/1.1"]
36 lines += [f"{key}: {value}" for key, value in hdrs.items()]
37 request = "\r\n".join(lines) + "\r\n\r\n" + body
38 return request.encode("utf-8")
39
40
41def load_capture(path: str) -> list[bytes]:
42 """
43 Parse AIRPLAY_DACP_CAPTURE lines from a verbose log file into raw request bytes.
44
45 Parses real-device capture logs for future replay fixtures; landed ahead of its
46 first consumer (the capture-mode task).
47
48 :param path: Path to a log file containing AIRPLAY_DACP_CAPTURE lines.
49 """
50 marker = "AIRPLAY_DACP_CAPTURE "
51 requests: list[bytes] = []
52 with open(path, encoding="utf-8") as handle:
53 for line in handle:
54 if marker not in line:
55 continue
56 payload = json.loads(line.split(marker, 1)[1])
57 requests.append(base64.b64decode(payload["raw_b64"]))
58 return requests
59
60
61async def replay(provider: AirPlayProvider, raw: bytes) -> bytes:
62 """
63 Feed raw request bytes through the real DACP handler and return the response bytes.
64
65 :param provider: Real AirPlayProvider instance (built with a mock mass).
66 :param raw: Raw DACP request bytes.
67 """
68 reader = asyncio.StreamReader()
69 reader.feed_data(raw)
70 reader.feed_eof()
71 writer = MagicMock()
72 writer.drain = AsyncMock()
73 writer.wait_closed = AsyncMock()
74 await provider._handle_dacp_request(reader, writer)
75 return writer.write.call_args.args[0] if writer.write.called else b""
76