/
/
/
1"""Tests for the NFS provider's mountpoint / scan root split."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock, patch
7
8import pytest
9from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
10from music_assistant_models.enums import ConfigEntryType, ProviderType
11from music_assistant_models.errors import SetupFailedError
12
13from music_assistant.constants import CONF_LOG_LEVEL
14from music_assistant.providers.filesystem_local import LocalFileSystemProvider
15from music_assistant.providers.filesystem_nfs.provider import NFSFileSystemProvider
16
17INSTANCE_ID = "filesystem_nfs--test"
18MOUNT_PATH = f"/tmp/{INSTANCE_ID}" # noqa: S108
19HOST = "nas.local"
20EXPORT_PATH = "/mnt/vault"
21
22
23def _provider(
24 subfolder: str | None = None, export_path: str = EXPORT_PATH
25) -> NFSFileSystemProvider:
26 """Create an NFS provider instance backed by in-memory setup data."""
27 setup_data: dict[str, Any] = {
28 "content_type": "music",
29 "host": HOST,
30 "export_path": export_path,
31 "nfs_version": "3",
32 }
33 if subfolder is not None:
34 setup_data["subfolder"] = subfolder
35 config = ProviderConfig(
36 values={
37 # required, and CRITICAL is the only value that cannot lower the root logger
38 CONF_LOG_LEVEL: ConfigEntry(
39 key=CONF_LOG_LEVEL, type=ConfigEntryType.STRING, value="CRITICAL"
40 )
41 },
42 type=ProviderType.MUSIC,
43 domain="filesystem_nfs",
44 instance_id=INSTANCE_ID,
45 )
46 manifest = MagicMock()
47 manifest.domain = "filesystem_nfs"
48 mass = MagicMock()
49 mass.config.get = MagicMock(return_value=setup_data)
50 mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
51 provider = NFSFileSystemProvider(mass, manifest, config, MOUNT_PATH)
52 provider.logger = MagicMock()
53 return provider
54
55
56@pytest.mark.parametrize("subfolder", [None, "", " ", "/"])
57def test_scan_root_is_the_mountpoint_without_subfolder(subfolder: str | None) -> None:
58 """Without a subfolder the scan root is the mountpoint itself, with no trailing separator."""
59 provider = _provider(subfolder)
60 assert provider.mount_path == MOUNT_PATH
61 assert provider.base_path == MOUNT_PATH
62
63
64@pytest.mark.parametrize(
65 ("subfolder", "expected"),
66 [
67 ("Music", "Music"),
68 # a leading slash is tolerated
69 ("/Music", "Music"),
70 ("Music/", "Music"),
71 ("albums/A-K", "albums/A-K"),
72 ],
73)
74def test_scan_root_is_the_subfolder_inside_the_mount(subfolder: str, expected: str) -> None:
75 """A configured subfolder becomes the scan root inside the (unchanged) mountpoint."""
76 provider = _provider(subfolder)
77 assert provider.mount_path == MOUNT_PATH
78 assert provider.base_path == f"{MOUNT_PATH}/{expected}"
79
80
81def test_instance_name_postfix_survives_the_config_migration() -> None:
82 """
83 Folding the subfolder into the export path must not rename the instance.
84
85 The settings migration rewrites `Export=/mnt/vault` + `Subfolder=Music` to
86 `Export=/mnt/vault/Music`; both must yield the same default instance name postfix.
87 """
88 assert _provider("Music").instance_name_postfix == "Music"
89 assert _provider(export_path=f"{EXPORT_PATH}/Music").instance_name_postfix == "Music"
90
91
92async def test_mount_source_never_includes_the_subfolder() -> None:
93 """The mount source is the export as configured; the subfolder stays out of the argv."""
94 provider = _provider("Music")
95 with patch(
96 "music_assistant.providers.filesystem_nfs.provider.check_output",
97 AsyncMock(return_value=(0, b"")),
98 ) as mock_check_output:
99 await provider.mount()
100
101 argv = mock_check_output.call_args.args
102 assert f"{HOST}:{EXPORT_PATH}" in argv
103 assert argv[-1] == MOUNT_PATH
104 assert not any("Music" in str(arg) for arg in argv)
105
106
107async def test_unmount_targets_the_mountpoint() -> None:
108 """Umount must be given the mountpoint; a subdirectory of a mount cannot be unmounted."""
109 provider = _provider("Music")
110 mock_unmount = AsyncMock()
111 with patch("music_assistant.providers.filesystem_nfs.provider.unmount", mock_unmount):
112 await provider.unload()
113
114 mock_unmount.assert_awaited_once_with(MOUNT_PATH, provider.logger)
115
116
117async def test_only_the_mountpoint_is_ever_created() -> None:
118 """MA creates the mountpoint but never the subfolder, on either side of the mount."""
119 provider = _provider("Music")
120 mock_makedirs = AsyncMock()
121 with (
122 patch(
123 "music_assistant.providers.filesystem_nfs.provider.get_ip_from_host",
124 AsyncMock(return_value="192.0.2.10"),
125 ),
126 patch.object(provider, "mount", AsyncMock()),
127 patch("music_assistant.providers.filesystem_nfs.provider.unmount", AsyncMock()),
128 patch.object(provider, "check_write_access", AsyncMock()),
129 patch("music_assistant.providers.filesystem_nfs.provider.makedirs", mock_makedirs),
130 patch(
131 "music_assistant.providers.filesystem_nfs.provider.isdir",
132 AsyncMock(return_value=True),
133 ),
134 ):
135 await provider.handle_async_init()
136
137 # created before the mount it would be masked; after it, MA writes to the user's share
138 mock_makedirs.assert_awaited_once_with(MOUNT_PATH, exist_ok=True)
139
140
141async def test_diagnostics_report_the_mountpoint_mount_state() -> None:
142 """A subdirectory of a mount is not itself a mountpoint, so ismount must get mount_path."""
143 provider = _provider("Music")
144 with (
145 patch(
146 "music_assistant.providers.filesystem_nfs.provider.ismount",
147 AsyncMock(return_value=True),
148 ) as mock_ismount,
149 patch.object(
150 LocalFileSystemProvider,
151 "get_diagnostics",
152 AsyncMock(return_value={"write_access": True}),
153 ),
154 ):
155 diagnostics = await provider.get_diagnostics()
156
157 mock_ismount.assert_awaited_once_with(MOUNT_PATH)
158 assert diagnostics["mounted"] is True
159
160
161def test_traversal_subfolder_is_rejected_before_mounting() -> None:
162 """A subfolder escaping the mount is refused while deriving the scan root."""
163 with pytest.raises(SetupFailedError) as exc_info:
164 _provider("../../etc")
165
166 assert exc_info.value.translation_key == "invalid_subfolder"
167 assert exc_info.value.translation_owner == "provider.filesystem_nfs"
168 assert MOUNT_PATH not in str(exc_info.value)
169
170
171@pytest.mark.parametrize(
172 "cleanup_error",
173 [
174 # the mountpoint turned out to be busy
175 SetupFailedError("Unable to unmount"),
176 # umount itself is missing or not executable
177 FileNotFoundError("umount"),
178 PermissionError("umount"),
179 ],
180)
181async def test_failed_cleanup_never_masks_the_missing_subfolder(
182 cleanup_error: Exception,
183) -> None:
184 """The best-effort unmount must not replace the one error the user can act on."""
185 provider = _provider("Music")
186 with (
187 patch(
188 "music_assistant.providers.filesystem_nfs.provider.get_ip_from_host",
189 AsyncMock(return_value="192.0.2.10"),
190 ),
191 patch.object(provider, "mount", AsyncMock()),
192 patch(
193 "music_assistant.providers.filesystem_nfs.provider.unmount",
194 # only the post-mount cleanup fails; a blocked pre-mount unmount is a real error
195 AsyncMock(side_effect=[None, cleanup_error]),
196 ),
197 patch("music_assistant.providers.filesystem_nfs.provider.makedirs", AsyncMock()),
198 patch(
199 "music_assistant.providers.filesystem_nfs.provider.isdir",
200 AsyncMock(return_value=False),
201 ),
202 pytest.raises(SetupFailedError) as exc_info,
203 ):
204 await provider.handle_async_init()
205
206 assert exc_info.value.translation_key == "subfolder_not_found"
207
208
209async def test_missing_subfolder_fails_setup_instead_of_an_empty_library() -> None:
210 """A subfolder that does not exist inside the mount surfaces a translated setup error."""
211 provider = _provider("Music")
212 mock_unmount = AsyncMock()
213 with (
214 patch(
215 "music_assistant.providers.filesystem_nfs.provider.get_ip_from_host",
216 AsyncMock(return_value="192.0.2.10"),
217 ),
218 patch.object(provider, "mount", AsyncMock()),
219 patch("music_assistant.providers.filesystem_nfs.provider.unmount", mock_unmount),
220 patch("music_assistant.providers.filesystem_nfs.provider.makedirs", AsyncMock()),
221 patch(
222 "music_assistant.providers.filesystem_nfs.provider.isdir",
223 AsyncMock(return_value=False),
224 ),
225 pytest.raises(SetupFailedError) as exc_info,
226 ):
227 await provider.handle_async_init()
228
229 err = exc_info.value
230 assert err.translation_key == "subfolder_not_found"
231 assert err.translation_owner == "provider.filesystem_nfs"
232 assert err.translation_args == ["Music"]
233 # the internal mountpoint must not leak into user-facing text
234 assert MOUNT_PATH not in str(err)
235 mock_unmount.assert_awaited()
236