/
/
/
1"""Tests for the WebDAV provider directory scan logic."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8
9from music_assistant.providers.filesystem_local.helpers import FileSystemItem, ScanErrors
10from music_assistant.providers.webdav.helpers import WebDAVItem
11from music_assistant.providers.webdav.provider import WebDAVFileSystemProvider
12
13BASE_URL = "https://host.example/dav"
14
15
16def _make_provider() -> WebDAVFileSystemProvider:
17 provider = WebDAVFileSystemProvider.__new__(WebDAVFileSystemProvider)
18 provider.base_url = BASE_URL
19 provider.username = None
20 provider.password = None
21 provider.logger = MagicMock()
22 provider.media_content_type = "music"
23 provider.config = MagicMock()
24 provider.config.get_value = MagicMock(return_value=False)
25 return provider
26
27
28async def _run_enumerate(
29 provider: WebDAVFileSystemProvider,
30 *,
31 file_checksums: dict[str, str] | None = None,
32 cur_filenames: set[str] | None = None,
33 scan_errors: ScanErrors | None = None,
34) -> None:
35 """Drive _enumerate_files_for_sync with empty sync buckets."""
36 await provider._enumerate_files_for_sync(
37 file_checksums=file_checksums or {},
38 cue_file_checksums={},
39 cur_filenames=cur_filenames if cur_filenames is not None else set(),
40 items_to_process=[],
41 unchanged_cue_items=[],
42 cue_stems=set(),
43 scan_errors=scan_errors if scan_errors is not None else ScanErrors(),
44 )
45
46
47def test_convert_skips_scanned_directory_with_special_chars() -> None:
48 """
49 The directory being scanned is returned by a depth-1 PROPFIND and must be skipped.
50
51 A name with a URL-reserved character (here ``;``) previously slipped past the skip
52 check, making the directory list itself and recurse until the recursion limit.
53 """
54 provider = _make_provider()
55 scan_path = "Live; Unplugged"
56 webdav_items = [
57 WebDAVItem(href="/dav/Live; Unplugged", name="Live; Unplugged", is_dir=True),
58 WebDAVItem(href="/dav/Live; Unplugged/01 track.mp3", name="01 track.mp3", is_dir=False),
59 ]
60
61 result = provider._convert_webdav_items(webdav_items, scan_path)
62
63 relative_paths = [item.relative_path for item in result]
64 assert relative_paths == ["Live; Unplugged/01 track.mp3"]
65 assert scan_path not in relative_paths
66
67
68def test_convert_handles_absolute_href_with_special_chars() -> None:
69 """
70 Some servers return absolute hrefs; the path must be extracted without a URL parser.
71
72 ``urlparse`` would treat the ``;`` as params and truncate the name, mis-scanning the
73 folder; the resolved relative path must keep reserved characters intact.
74 """
75 provider = _make_provider()
76 scan_path = "Live; Unplugged"
77 webdav_items = [
78 WebDAVItem(
79 href="https://host.example/dav/Live; Unplugged",
80 name="Live; Unplugged",
81 is_dir=True,
82 ),
83 WebDAVItem(
84 href="https://host.example/dav/Live; Unplugged/01 track.mp3",
85 name="01 track.mp3",
86 is_dir=False,
87 ),
88 ]
89
90 result = provider._convert_webdav_items(webdav_items, scan_path)
91
92 assert [item.relative_path for item in result] == ["Live; Unplugged/01 track.mp3"]
93
94
95def test_convert_skips_base_directory_at_root() -> None:
96 """The base directory itself must be skipped when scanning the root."""
97 provider = _make_provider()
98 webdav_items = [
99 WebDAVItem(href="/dav", name="dav", is_dir=True),
100 WebDAVItem(href="/dav/Artist", name="Artist", is_dir=True),
101 ]
102
103 result = provider._convert_webdav_items(webdav_items, "")
104
105 assert [item.relative_path for item in result] == ["Artist"]
106
107
108async def test_enumerate_stops_on_directory_cycle() -> None:
109 """A directory cycle must not exhaust the recursion limit."""
110 provider = _make_provider()
111 # A -> A/B -> A (back-edge to an ancestor); only the guard can break this
112 listing = {
113 "": [FileSystemItem("A", "A", "", is_dir=True)],
114 "A": [FileSystemItem("B", "A/B", "", is_dir=True)],
115 "A/B": [FileSystemItem("A", "A", "", is_dir=True)],
116 }
117 scanned: list[str] = []
118
119 async def fake_scandir(path: str) -> list[FileSystemItem]:
120 scanned.append(path)
121 return listing[path]
122
123 provider._scandir = AsyncMock(side_effect=fake_scandir) # type: ignore[method-assign]
124
125 await _run_enumerate(provider)
126
127 # each directory is scanned exactly once despite the cycle
128 assert sorted(scanned) == ["", "A", "A/B"]
129
130
131@pytest.mark.parametrize("special", ["Live; Unplugged", "Die drei ???", "Rock #1"])
132async def test_enumerate_does_not_loop_on_special_chars(special: str) -> None:
133 """Folders with reserved characters must be traversed without looping."""
134 provider = _make_provider()
135 track = FileSystemItem("t.mp3", f"{special}/t.mp3", "", is_dir=False, checksum="1")
136 listing = {
137 "": [FileSystemItem(special, special, "", is_dir=True)],
138 special: [track],
139 }
140
141 async def fake_scandir(path: str) -> list[FileSystemItem]:
142 return listing[path]
143
144 provider._scandir = AsyncMock(side_effect=fake_scandir) # type: ignore[method-assign]
145 cur_filenames: set[str] = set()
146
147 # the track is unchanged (checksum matches), so reaching it records it as present
148 await _run_enumerate(
149 provider, file_checksums={f"{special}/t.mp3": "1"}, cur_filenames=cur_filenames
150 )
151
152 assert cur_filenames == {f"{special}/t.mp3"}
153