/
/
/
1"""Tests for the WebDAV provider directory scan logic."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8from music_assistant_models.errors import ProviderUnavailableError
9
10from music_assistant.providers.filesystem_local.helpers import FileSystemItem, ScanErrors
11from music_assistant.providers.webdav.helpers import WebDAVItem
12from music_assistant.providers.webdav.provider import WebDAVFileSystemProvider
13
14BASE_URL = "https://host.example/dav"
15
16
17def _make_provider() -> WebDAVFileSystemProvider:
18 provider = WebDAVFileSystemProvider.__new__(WebDAVFileSystemProvider)
19 provider.base_url = BASE_URL
20 provider.username = None
21 provider.password = None
22 provider.logger = MagicMock()
23 provider.media_content_type = "music"
24 provider.config = MagicMock()
25 provider.config.get_value = MagicMock(return_value=False)
26 return provider
27
28
29async def _run_enumerate(
30 provider: WebDAVFileSystemProvider,
31 *,
32 file_checksums: dict[str, str] | None = None,
33 cur_filenames: set[str] | None = None,
34 scan_errors: ScanErrors | None = None,
35) -> None:
36 """Drive _enumerate_files_for_sync with empty sync buckets."""
37 await provider._enumerate_files_for_sync(
38 file_checksums=file_checksums or {},
39 cue_file_checksums={},
40 cur_filenames=cur_filenames if cur_filenames is not None else set(),
41 items_to_process=[],
42 unchanged_cue_items=[],
43 cue_stems=set(),
44 scan_errors=scan_errors if scan_errors is not None else ScanErrors(),
45 metadata_files=[],
46 )
47
48
49def test_convert_skips_scanned_directory_with_special_chars() -> None:
50 """
51 The directory being scanned is returned by a depth-1 PROPFIND and must be skipped.
52
53 A name with a URL-reserved character (here ``;``) previously slipped past the skip
54 check, making the directory list itself and recurse until the recursion limit.
55 """
56 provider = _make_provider()
57 scan_path = "Live; Unplugged"
58 webdav_items = [
59 WebDAVItem(href="/dav/Live; Unplugged", name="Live; Unplugged", is_dir=True),
60 WebDAVItem(href="/dav/Live; Unplugged/01 track.mp3", name="01 track.mp3", is_dir=False),
61 ]
62
63 result = provider._convert_webdav_items(webdav_items, scan_path)
64
65 relative_paths = [item.relative_path for item in result]
66 assert relative_paths == ["Live; Unplugged/01 track.mp3"]
67 assert scan_path not in relative_paths
68
69
70def test_convert_handles_absolute_href_with_special_chars() -> None:
71 """
72 Some servers return absolute hrefs; the path must be extracted without a URL parser.
73
74 ``urlparse`` would treat the ``;`` as params and truncate the name, mis-scanning the
75 folder; the resolved relative path must keep reserved characters intact.
76 """
77 provider = _make_provider()
78 scan_path = "Live; Unplugged"
79 webdav_items = [
80 WebDAVItem(
81 href="https://host.example/dav/Live; Unplugged",
82 name="Live; Unplugged",
83 is_dir=True,
84 ),
85 WebDAVItem(
86 href="https://host.example/dav/Live; Unplugged/01 track.mp3",
87 name="01 track.mp3",
88 is_dir=False,
89 ),
90 ]
91
92 result = provider._convert_webdav_items(webdav_items, scan_path)
93
94 assert [item.relative_path for item in result] == ["Live; Unplugged/01 track.mp3"]
95
96
97def test_convert_skips_base_directory_at_root() -> None:
98 """The base directory itself must be skipped when scanning the root."""
99 provider = _make_provider()
100 webdav_items = [
101 WebDAVItem(href="/dav", name="dav", is_dir=True),
102 WebDAVItem(href="/dav/Artist", name="Artist", is_dir=True),
103 ]
104
105 result = provider._convert_webdav_items(webdav_items, "")
106
107 assert [item.relative_path for item in result] == ["Artist"]
108
109
110async def test_enumerate_stops_on_directory_cycle() -> None:
111 """A directory cycle must not exhaust the recursion limit."""
112 provider = _make_provider()
113 # A -> A/B -> A (back-edge to an ancestor); only the guard can break this
114 listing = {
115 "": [FileSystemItem("A", "A", "", is_dir=True)],
116 "A": [FileSystemItem("B", "A/B", "", is_dir=True)],
117 "A/B": [FileSystemItem("A", "A", "", is_dir=True)],
118 }
119 scanned: list[str] = []
120
121 async def fake_scandir(path: str) -> list[FileSystemItem]:
122 scanned.append(path)
123 return listing[path]
124
125 provider._scandir = AsyncMock(side_effect=fake_scandir) # type: ignore[method-assign]
126
127 await _run_enumerate(provider)
128
129 # each directory is scanned exactly once despite the cycle
130 assert sorted(scanned) == ["", "A", "A/B"]
131
132
133async def test_enumerate_routes_metadata_files_separately_from_media() -> None:
134 """The WebDAV walk collects NFO/images as metadata files, never as media or present files."""
135 provider = _make_provider()
136 provider._sync_tracks = True
137 listing = {
138 "": [
139 FileSystemItem("Artist", "Artist", "", is_dir=True),
140 ],
141 "Artist": [
142 FileSystemItem("track.mp3", "Artist/track.mp3", "", is_dir=False, checksum="1"),
143 FileSystemItem("artist.nfo", "Artist/artist.nfo", "", is_dir=False, checksum="1"),
144 FileSystemItem("folder.jpg", "Artist/folder.jpg", "", is_dir=False, checksum="1"),
145 FileSystemItem("notes.txt", "Artist/notes.txt", "", is_dir=False, checksum="1"),
146 ],
147 }
148
149 async def fake_scandir(path: str) -> list[FileSystemItem]:
150 return listing[path]
151
152 provider._scandir = AsyncMock(side_effect=fake_scandir) # type: ignore[method-assign]
153 items_to_process: list[tuple[FileSystemItem, str | None]] = []
154 cur_filenames: set[str] = set()
155 metadata_files: list[FileSystemItem] = []
156
157 await provider._enumerate_files_for_sync(
158 file_checksums={},
159 cue_file_checksums={},
160 cur_filenames=cur_filenames,
161 items_to_process=items_to_process,
162 unchanged_cue_items=[],
163 cue_stems=set(),
164 scan_errors=ScanErrors(),
165 metadata_files=metadata_files,
166 )
167
168 assert [item.relative_path for item, _ in items_to_process] == ["Artist/track.mp3"]
169 assert sorted(item.relative_path for item in metadata_files) == [
170 "Artist/artist.nfo",
171 "Artist/folder.jpg",
172 ]
173 # metadata files are never recorded present (unrelated to deletion tracking); notes.txt
174 # is neither media nor a recognized metadata file, so the walk skips it entirely
175 assert cur_filenames == set()
176
177
178@pytest.mark.parametrize("special", ["Live; Unplugged", "Die drei ???", "Rock #1"])
179async def test_enumerate_does_not_loop_on_special_chars(special: str) -> None:
180 """Folders with reserved characters must be traversed without looping."""
181 provider = _make_provider()
182 track = FileSystemItem("t.mp3", f"{special}/t.mp3", "", is_dir=False, checksum="1")
183 listing = {
184 "": [FileSystemItem(special, special, "", is_dir=True)],
185 special: [track],
186 }
187
188 async def fake_scandir(path: str) -> list[FileSystemItem]:
189 return listing[path]
190
191 provider._scandir = AsyncMock(side_effect=fake_scandir) # type: ignore[method-assign]
192 cur_filenames: set[str] = set()
193
194 # the track is unchanged (checksum matches), so reaching it records it as present
195 await _run_enumerate(
196 provider, file_checksums={f"{special}/t.mp3": "1"}, cur_filenames=cur_filenames
197 )
198
199 assert cur_filenames == {f"{special}/t.mp3"}
200
201
202async def test_is_reachable_asks_the_server(monkeypatch: pytest.MonkeyPatch) -> None:
203 """A WebDAV url cannot be stat'ed, so reachability is a live request to the server."""
204 provider = _make_provider()
205 provider.verify_ssl = True
206 provider.mass = MagicMock()
207
208 test_connection = AsyncMock()
209 monkeypatch.setattr(
210 "music_assistant.providers.webdav.provider.webdav_test_connection", test_connection
211 )
212 assert await provider._is_reachable() is True
213 test_connection.assert_awaited_once()
214
215 test_connection.side_effect = ProviderUnavailableError("server down")
216 with pytest.raises(ProviderUnavailableError):
217 await provider._is_reachable()
218