/
/
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 )
46
47
48def test_convert_skips_scanned_directory_with_special_chars() -> None:
49 """
50 The directory being scanned is returned by a depth-1 PROPFIND and must be skipped.
51
52 A name with a URL-reserved character (here ``;``) previously slipped past the skip
53 check, making the directory list itself and recurse until the recursion limit.
54 """
55 provider = _make_provider()
56 scan_path = "Live; Unplugged"
57 webdav_items = [
58 WebDAVItem(href="/dav/Live; Unplugged", name="Live; Unplugged", is_dir=True),
59 WebDAVItem(href="/dav/Live; Unplugged/01 track.mp3", name="01 track.mp3", is_dir=False),
60 ]
61
62 result = provider._convert_webdav_items(webdav_items, scan_path)
63
64 relative_paths = [item.relative_path for item in result]
65 assert relative_paths == ["Live; Unplugged/01 track.mp3"]
66 assert scan_path not in relative_paths
67
68
69def test_convert_handles_absolute_href_with_special_chars() -> None:
70 """
71 Some servers return absolute hrefs; the path must be extracted without a URL parser.
72
73 ``urlparse`` would treat the ``;`` as params and truncate the name, mis-scanning the
74 folder; the resolved relative path must keep reserved characters intact.
75 """
76 provider = _make_provider()
77 scan_path = "Live; Unplugged"
78 webdav_items = [
79 WebDAVItem(
80 href="https://host.example/dav/Live; Unplugged",
81 name="Live; Unplugged",
82 is_dir=True,
83 ),
84 WebDAVItem(
85 href="https://host.example/dav/Live; Unplugged/01 track.mp3",
86 name="01 track.mp3",
87 is_dir=False,
88 ),
89 ]
90
91 result = provider._convert_webdav_items(webdav_items, scan_path)
92
93 assert [item.relative_path for item in result] == ["Live; Unplugged/01 track.mp3"]
94
95
96def test_convert_skips_base_directory_at_root() -> None:
97 """The base directory itself must be skipped when scanning the root."""
98 provider = _make_provider()
99 webdav_items = [
100 WebDAVItem(href="/dav", name="dav", is_dir=True),
101 WebDAVItem(href="/dav/Artist", name="Artist", is_dir=True),
102 ]
103
104 result = provider._convert_webdav_items(webdav_items, "")
105
106 assert [item.relative_path for item in result] == ["Artist"]
107
108
109async def test_enumerate_stops_on_directory_cycle() -> None:
110 """A directory cycle must not exhaust the recursion limit."""
111 provider = _make_provider()
112 # A -> A/B -> A (back-edge to an ancestor); only the guard can break this
113 listing = {
114 "": [FileSystemItem("A", "A", "", is_dir=True)],
115 "A": [FileSystemItem("B", "A/B", "", is_dir=True)],
116 "A/B": [FileSystemItem("A", "A", "", is_dir=True)],
117 }
118 scanned: list[str] = []
119
120 async def fake_scandir(path: str) -> list[FileSystemItem]:
121 scanned.append(path)
122 return listing[path]
123
124 provider._scandir = AsyncMock(side_effect=fake_scandir) # type: ignore[method-assign]
125
126 await _run_enumerate(provider)
127
128 # each directory is scanned exactly once despite the cycle
129 assert sorted(scanned) == ["", "A", "A/B"]
130
131
132@pytest.mark.parametrize("special", ["Live; Unplugged", "Die drei ???", "Rock #1"])
133async def test_enumerate_does_not_loop_on_special_chars(special: str) -> None:
134 """Folders with reserved characters must be traversed without looping."""
135 provider = _make_provider()
136 track = FileSystemItem("t.mp3", f"{special}/t.mp3", "", is_dir=False, checksum="1")
137 listing = {
138 "": [FileSystemItem(special, special, "", is_dir=True)],
139 special: [track],
140 }
141
142 async def fake_scandir(path: str) -> list[FileSystemItem]:
143 return listing[path]
144
145 provider._scandir = AsyncMock(side_effect=fake_scandir) # type: ignore[method-assign]
146 cur_filenames: set[str] = set()
147
148 # the track is unchanged (checksum matches), so reaching it records it as present
149 await _run_enumerate(
150 provider, file_checksums={f"{special}/t.mp3": "1"}, cur_filenames=cur_filenames
151 )
152
153 assert cur_filenames == {f"{special}/t.mp3"}
154
155
156async def test_is_reachable_asks_the_server(monkeypatch: pytest.MonkeyPatch) -> None:
157 """A WebDAV url cannot be stat'ed, so reachability is a live request to the server."""
158 provider = _make_provider()
159 provider.verify_ssl = True
160 provider.mass = MagicMock()
161
162 test_connection = AsyncMock()
163 monkeypatch.setattr(
164 "music_assistant.providers.webdav.provider.webdav_test_connection", test_connection
165 )
166 assert await provider._is_reachable() is True
167 test_connection.assert_awaited_once()
168
169 test_connection.side_effect = ProviderUnavailableError("server down")
170 with pytest.raises(ProviderUnavailableError):
171 await provider._is_reachable()
172