/
/
/
1"""Tests for the WebDAV helper functions."""
2
3from __future__ import annotations
4
5from typing import Any, Self, cast
6
7import aiohttp
8import pytest
9
10from music_assistant.providers.webdav.helpers import build_webdav_url, webdav_propfind
11
12BASE_URL = "https://host.example/remote.php/dav/files/user/Music"
13
14EMPTY_MULTISTATUS = (
15 '<?xml version="1.0" encoding="utf-8"?>\n<d:multistatus xmlns:d="DAV:"></d:multistatus>'
16)
17
18
19class _FakeResponse:
20 """Minimal async-context-manager stand-in for an aiohttp response."""
21
22 def __init__(self, status: int, body: str) -> None:
23 self.status = status
24 self._body = body
25
26 async def __aenter__(self) -> Self:
27 return self
28
29 async def __aexit__(self, *_exc: object) -> bool:
30 return False
31
32 async def text(self) -> str:
33 return self._body
34
35
36class _FakeSession:
37 """Capture the headers passed to session.request for assertion."""
38
39 def __init__(self, response: _FakeResponse) -> None:
40 self._response = response
41 self.last_headers: dict[str, str] | None = None
42
43 def request(self, _method: str, _url: str, *, headers: dict[str, str], **_kwargs: Any) -> Any:
44 self.last_headers = headers
45 return self._response
46
47
48@pytest.mark.parametrize(
49 ("path", "expected"),
50 [
51 ("Artist/Album", f"{BASE_URL}/Artist/Album"),
52 ("/Artist/Album", f"{BASE_URL}/Artist/Album"),
53 # reserved characters must be percent-encoded, not interpreted as
54 # params/query/fragment/scheme by the URL machinery
55 ("Live; Unplugged", f"{BASE_URL}/Live%3B%20Unplugged"),
56 ("Die drei ???", f"{BASE_URL}/Die%20drei%20%3F%3F%3F"),
57 ("Rock #1", f"{BASE_URL}/Rock%20%231"),
58 ("Live: In Concert", f"{BASE_URL}/Live%3A%20In%20Concert"),
59 # the path separator and umlauts are handled as expected
60 ("Sigur Rós/Ãgætis", f"{BASE_URL}/Sigur%20R%C3%B3s/%C3%81g%C3%A6tis"),
61 ],
62)
63def test_build_webdav_url_encodes_reserved_characters(path: str, expected: str) -> None:
64 """Reserved characters in resource paths must be percent-encoded."""
65 assert build_webdav_url(BASE_URL, path) == expected
66
67
68def test_build_webdav_url_passes_through_absolute_urls() -> None:
69 """An absolute URL (e.g. from a playlist line) must be returned unchanged."""
70 absolute = "http://other.example/song.mp3"
71 assert build_webdav_url(BASE_URL, absolute) == absolute
72
73
74async def test_webdav_propfind_sends_authorization_header() -> None:
75 """A provided auth_header must be sent as the Authorization request header."""
76 session = _FakeSession(_FakeResponse(207, EMPTY_MULTISTATUS))
77 auth_header = aiohttp.encode_basic_auth("user", "pass")
78
79 await webdav_propfind(
80 cast("aiohttp.ClientSession", session),
81 BASE_URL,
82 depth=0,
83 auth_header=auth_header,
84 )
85
86 assert session.last_headers is not None
87 assert session.last_headers["Authorization"] == auth_header
88
89
90async def test_webdav_propfind_omits_authorization_header_when_unset() -> None:
91 """Without credentials no Authorization header must be sent."""
92 session = _FakeSession(_FakeResponse(207, EMPTY_MULTISTATUS))
93
94 await webdav_propfind(
95 cast("aiohttp.ClientSession", session),
96 BASE_URL,
97 depth=0,
98 auth_header=None,
99 )
100
101 assert session.last_headers is not None
102 assert "Authorization" not in session.last_headers
103
104
105async def test_webdav_propfind_parses_etag_from_single_propstat() -> None:
106 """A normal single-propstat response yields a normalized (unquoted, non-weak) ETag."""
107 body = f"""<?xml version="1.0" encoding="utf-8"?>
108<d:multistatus xmlns:d="DAV:">
109 <d:response>
110 <d:href>{BASE_URL}/track.mp3</d:href>
111 <d:propstat>
112 <d:prop>
113 <d:resourcetype/>
114 <d:getcontentlength>1234</d:getcontentlength>
115 <d:getlastmodified>Mon, 01 Jan 2024 00:00:00 GMT</d:getlastmodified>
116 <d:getetag>W/"abc123"</d:getetag>
117 </d:prop>
118 <d:status>HTTP/1.1 200 OK</d:status>
119 </d:propstat>
120 </d:response>
121</d:multistatus>"""
122 session = _FakeSession(_FakeResponse(207, body))
123
124 items = await webdav_propfind(cast("aiohttp.ClientSession", session), BASE_URL, depth=1)
125
126 assert len(items) == 1
127 assert items[0].name == "track.mp3"
128 assert items[0].is_dir is False
129 assert items[0].size == 1234
130 assert items[0].etag == "abc123" # weak-tag prefix and quotes stripped
131
132
133async def test_webdav_propfind_merges_split_propstat_blocks() -> None:
134 """
135 A server splitting an unsupported getetag into its own failed propstat is tolerated.
136
137 A response with an earlier 404 propstat (only offering getetag) must not shadow the
138 resourcetype/getlastmodified/getcontentlength carried by a later, successful propstat.
139 """
140 body = f"""<?xml version="1.0" encoding="utf-8"?>
141<d:multistatus xmlns:d="DAV:">
142 <d:response>
143 <d:href>{BASE_URL}/Album</d:href>
144 <d:propstat>
145 <d:prop>
146 <d:getetag/>
147 </d:prop>
148 <d:status>HTTP/1.1 404 Not Found</d:status>
149 </d:propstat>
150 <d:propstat>
151 <d:prop>
152 <d:resourcetype><d:collection/></d:resourcetype>
153 <d:getlastmodified>Mon, 01 Jan 2024 00:00:00 GMT</d:getlastmodified>
154 <d:displayname>Album</d:displayname>
155 </d:prop>
156 <d:status>HTTP/1.1 200 OK</d:status>
157 </d:propstat>
158 </d:response>
159</d:multistatus>"""
160 session = _FakeSession(_FakeResponse(207, body))
161
162 items = await webdav_propfind(cast("aiohttp.ClientSession", session), BASE_URL, depth=1)
163
164 assert len(items) == 1
165 assert items[0].name == "Album"
166 assert items[0].is_dir is True # resourcetype must not be lost to the failed propstat
167 assert items[0].last_modified == "Mon, 01 Jan 2024 00:00:00 GMT"
168 assert items[0].etag is None # genuinely unsupported by this server
169