music-assistant-server

1"""Tests for utility/helper functions."""
2
3import errno
4import logging
5import os
6from pathlib import Path
7from typing import Self
8from unittest.mock import patch
9
10import pytest
11
12from music_assistant.providers.filesystem_local import helpers
13
14# ruff: noqa: S108
15
16
17def test_get_artist_dir() -> None:
18    """Test the extraction of an artist dir."""
19    album_path = "/tmp/Artist/Album"
20    artist_name = "Artist"
21    assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/Artist"
22    album_path = "/tmp/artist/Album"
23    assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/artist"
24    album_path = "/tmp/Album"
25    assert helpers.get_artist_dir(artist_name, album_path) is None
26    album_path = "/tmp/ARTIST!/Album"
27    assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/ARTIST!"
28    album_path = "/tmp/Artist/Album"
29    artist_name = "Artist!"
30    assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/Artist"
31    album_path = "/tmp/REM/Album"
32    artist_name = "R.E.M."
33    assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/REM"
34    album_path = "/tmp/ACDC/Album"
35    artist_name = "AC/DC"
36    assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/ACDC"
37    album_path = "/tmp/Celine Dion/Album"
38    artist_name = "Céline Dion"
39    assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/Celine Dion"
40    album_path = "/tmp/Antonin Dvorak/Album"
41    artist_name = "Antonín Dvořák"
42    assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/Antonin Dvorak"
43
44
45@pytest.mark.parametrize(
46    ("album_name", "track_dir", "expected"),
47    [
48        # Test literal match
49        (
50            "Selected Ambient Works 85-92",
51            "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92",
52            "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92",
53        ),
54        # Test artist - album format
55        (
56            "Selected Ambient Works 85-92",
57            "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92",
58            "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92",
59        ),
60        # Test artist - album (version) format
61        (
62            "Selected Ambient Works 85-92",
63            "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92 (Remastered)",
64            "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92 (Remastered)",
65        ),
66        # Test artist - album (version) format
67        (
68            "Selected Ambient Works 85-92",
69            "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92 (Remastered) - WEB",
70            "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92 (Remastered) - WEB",
71        ),
72        # Test tokenizer - dirname with extras
73        (
74            "Fokus - Prewersje",
75            "/home/user/Fokus-Prewersje-PL-WEB-FLAC-2021-PS_INT",
76            "/home/user/Fokus-Prewersje-PL-WEB-FLAC-2021-PS_INT",
77        ),
78        # Test tokenizer - dirname with version and extras
79        (
80            "Layo And Bushwacka - Night Works",
81            "/home/music/Layo_And_Bushwacka-Night_Works_(Reissue)-(XLCD_154X)-FLAC-2003",
82            "/home/music/Layo_And_Bushwacka-Night_Works_(Reissue)-(XLCD_154X)-FLAC-2003",
83        ),
84        # Test tokenizer - extras and approximate match on diacratics
85        (
86            "Łona i Webber - Wyślij Sobie Pocztówkę",
87            "/usr/others/Lona-Discography-PL-FLAC-2020-INT/Lona_I_Webber-Wyslij_Sobie_Pocztowke-PL-WEB-FLAC-2014-PS",
88            "/usr/others/Lona-Discography-PL-FLAC-2020-INT/Lona_I_Webber-Wyslij_Sobie_Pocztowke-PL-WEB-FLAC-2014-PS",
89        ),
90        (
91            "NIC",
92            "/nas/downloads/others/Sokol-NIC-PL-WEB-FLAC-2021",
93            "/nas/downloads/others/Sokol-NIC-PL-WEB-FLAC-2021",
94        ),
95        # Test album (version) format
96        (
97            "Aphex Twin - Selected Ambient Works 85-92",
98            "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92 (Remastered)",
99            "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92 (Remastered)",
100        ),
101        # Test album name in dir
102        (
103            "Aphex Twin - Selected Ambient Works 85-92",
104            "/home/user/Music/RandomDirWithAphex Twin - Selected Ambient Works 85-92InIt",
105            "/home/user/Music/RandomDirWithAphex Twin - Selected Ambient Works 85-92InIt",
106        ),
107        # Test no match
108        (
109            "NonExistentAlbumName",
110            "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92",
111            None,
112        ),
113        # Test empty album name
114        ("", "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92", None),
115        # Test empty track dir
116        ("Selected Ambient Works 85-92", "", None),
117    ],
118)
119def test_get_album_dir(album_name: str, track_dir: str, expected: str) -> None:
120    """Test the extraction of an album dir."""
121    assert helpers.get_album_dir(track_dir, album_name) == expected
122
123
124SUPPORTED = {"mp3", "flac"}
125
126
127def _build_music_tree(root: Path) -> None:
128    """Create a small music tree fixture."""
129    (root / "Artist1" / "Album1").mkdir(parents=True)
130    (root / "Artist1" / "Album1" / "track1.mp3").write_bytes(b"x")
131    (root / "Artist1" / "Album1" / "track2.flac").write_bytes(b"x")
132    (root / "Artist2").mkdir()
133    (root / "Artist2" / "track3.mp3").write_bytes(b"x")
134
135
136def test_recursive_iter_happy_path(tmp_path: Path) -> None:
137    """Test that a healthy scan yields all supported files and records no errors."""
138    _build_music_tree(tmp_path)
139    errors = helpers.ScanErrors()
140    items = list(
141        helpers.recursive_iter(
142            str(tmp_path),
143            str(tmp_path),
144            SUPPORTED,
145            logging.getLogger("test"),
146            errors,
147        )
148    )
149    rel_paths = sorted(i.relative_path for i in items)
150    assert rel_paths == [
151        "Artist1/Album1/track1.mp3",
152        "Artist1/Album1/track2.flac",
153        "Artist2/track3.mp3",
154    ]
155    assert not errors.fatal
156    assert errors.failed_dirs == 0
157
158
159def test_recursive_iter_root_unreachable_records_error(tmp_path: Path) -> None:
160    """Test that a missing root path is reported via scan_errors."""
161    errors = helpers.ScanErrors()
162    missing = tmp_path / "does-not-exist"
163    items = list(
164        helpers.recursive_iter(
165            str(missing),
166            str(missing),
167            SUPPORTED,
168            logging.getLogger("test"),
169            errors,
170        )
171    )
172    assert items == []
173    assert isinstance(errors.fatal, OSError)
174    assert errors.fatal.errno == errno.ENOENT
175
176
177def test_recursive_iter_root_eacces_records_error() -> None:
178    """Test that permission-denied on the root path is reported via scan_errors."""
179    errors = helpers.ScanErrors()
180    with patch("os.scandir", side_effect=PermissionError(errno.EACCES, "denied")):
181        items = list(
182            helpers.recursive_iter(
183                "/fake/root",
184                "/fake/root",
185                SUPPORTED,
186                logging.getLogger("test"),
187                errors,
188            )
189        )
190    assert items == []
191    assert isinstance(errors.fatal, OSError)
192    assert errors.fatal.errno == errno.EACCES
193
194
195def test_recursive_iter_subfolder_failure_is_not_fatal(tmp_path: Path) -> None:
196    """Test that a single sub-folder scan failure is not fatal."""
197    _build_music_tree(tmp_path)
198    errors = helpers.ScanErrors()
199    real_scandir = os.scandir
200    bad_dir = str(tmp_path / "Artist1" / "Album1")
201
202    def fake_scandir(path: str | os.PathLike[str]):  # type: ignore[no-untyped-def]
203        if str(path) == bad_dir:
204            raise OSError(errno.EIO, "i/o error")
205        return real_scandir(path)
206
207    with patch("os.scandir", side_effect=fake_scandir):
208        items = list(
209            helpers.recursive_iter(
210                str(tmp_path),
211                str(tmp_path),
212                SUPPORTED,
213                logging.getLogger("test"),
214                errors,
215            )
216        )
217
218    rel_paths = sorted(i.relative_path for i in items)
219    assert rel_paths == ["Artist2/track3.mp3"]
220    assert not errors.fatal
221    # the scan is incomplete, so callers must not run deletions
222    assert errors.failed_dirs == 1
223
224
225def test_recursive_iter_einval_is_ignored() -> None:
226    """Test that EINVAL from an unsupported path name is not recorded."""
227    errors = helpers.ScanErrors()
228    with patch("os.scandir", side_effect=OSError(errno.EINVAL, "invalid path")):
229        items = list(
230            helpers.recursive_iter(
231                "/weird/\udcff",
232                "/weird/\udcff",
233                SUPPORTED,
234                logging.getLogger("test"),
235                errors,
236            )
237        )
238    assert items == []
239    assert not errors.fatal
240    assert errors.failed_dirs == 0
241
242
243def _build_flat_tree(root: Path, count: int) -> None:
244    """Create a music tree with the given number of album folders."""
245    for index in range(count):
246        album_dir = root / f"Album{index:03d}"
247        album_dir.mkdir()
248        (album_dir / "track.mp3").write_bytes(b"x")
249
250
251def test_recursive_iter_aborts_after_consecutive_failures(tmp_path: Path) -> None:
252    """Test that storage disappearing mid-scan aborts the walk instead of grinding on."""
253    _build_flat_tree(tmp_path, helpers.MAX_CONSECUTIVE_SCAN_ERRORS + 10)
254    errors = helpers.ScanErrors()
255    real_scandir = os.scandir
256
257    def fake_scandir(path: str | os.PathLike[str]):  # type: ignore[no-untyped-def]
258        if str(path) == str(tmp_path):
259            return real_scandir(path)
260        raise OSError(errno.EIO, "i/o error")
261
262    with patch("os.scandir", side_effect=fake_scandir):
263        items = list(
264            helpers.recursive_iter(
265                str(tmp_path),
266                str(tmp_path),
267                SUPPORTED,
268                logging.getLogger("test"),
269                errors,
270            )
271        )
272
273    assert items == []
274    assert errors.aborted
275    # the walk stopped at the threshold instead of trying every remaining folder
276    assert errors.failed_dirs == helpers.MAX_CONSECUTIVE_SCAN_ERRORS
277
278
279def test_recursive_iter_einval_does_not_abort(tmp_path: Path) -> None:
280    """Test that skipped (unsupported) path names never trip the abort threshold."""
281    _build_flat_tree(tmp_path, helpers.MAX_CONSECUTIVE_SCAN_ERRORS + 10)
282    (tmp_path / "root.mp3").write_bytes(b"x")
283    errors = helpers.ScanErrors()
284    real_scandir = os.scandir
285
286    def fake_scandir(path: str | os.PathLike[str]):  # type: ignore[no-untyped-def]
287        if str(path) == str(tmp_path):
288            return real_scandir(path)
289        raise OSError(errno.EINVAL, "invalid argument")
290
291    with patch("os.scandir", side_effect=fake_scandir):
292        items = list(
293            helpers.recursive_iter(
294                str(tmp_path),
295                str(tmp_path),
296                SUPPORTED,
297                logging.getLogger("test"),
298                errors,
299            )
300        )
301
302    assert [item.relative_path for item in items] == ["root.mp3"]
303    assert not errors.aborted
304    assert errors.failed_dirs == 0
305
306
307def test_recursive_iter_permission_denied_does_not_abort(tmp_path: Path) -> None:
308    """Test that ACL-protected folders leave the scan incomplete without aborting it."""
309    _build_flat_tree(tmp_path, helpers.MAX_CONSECUTIVE_SCAN_ERRORS + 10)
310    (tmp_path / "root.mp3").write_bytes(b"x")
311    errors = helpers.ScanErrors()
312    real_scandir = os.scandir
313
314    def fake_scandir(path: str | os.PathLike[str]):  # type: ignore[no-untyped-def]
315        if str(path) == str(tmp_path):
316            return real_scandir(path)
317        raise PermissionError(errno.EACCES, "denied")
318
319    with patch("os.scandir", side_effect=fake_scandir):
320        items = list(
321            helpers.recursive_iter(
322                str(tmp_path),
323                str(tmp_path),
324                SUPPORTED,
325                logging.getLogger("test"),
326                errors,
327            )
328        )
329
330    assert [item.relative_path for item in items] == ["root.mp3"]
331    assert not errors.aborted
332    assert errors.consecutive_failures == 0
333    # the folders were still missed, so callers must not run deletions
334    assert errors.failed_dirs == helpers.MAX_CONSECUTIVE_SCAN_ERRORS + 10
335
336
337class _BrokenEntry:
338    """Directory entry whose type check fails, as on a share that drops mid-listing."""
339
340    def __init__(self, path: str, err: OSError) -> None:
341        self.name = Path(path).name
342        self.path = path
343        self._err = err
344
345    def is_dir(self, follow_symlinks: bool = True) -> bool:
346        """Raise the configured error instead of answering."""
347        raise self._err
348
349    def is_file(self, follow_symlinks: bool = True) -> bool:
350        """Raise the configured error instead of answering."""
351        raise self._err
352
353
354class _FakeScanDir:
355    """Stand-in for the os.scandir iterator, which is also a context manager."""
356
357    def __init__(self, entries: list[_BrokenEntry]) -> None:
358        self._entries = iter(entries)
359
360    def __enter__(self) -> Self:
361        return self
362
363    def __exit__(self, *_exc: object) -> None:
364        return None
365
366    def __iter__(self) -> Self:
367        return self
368
369    def __next__(self) -> _BrokenEntry:
370        return next(self._entries)
371
372
373def test_recursive_iter_unreadable_file_is_recorded(
374    tmp_path: Path, caplog: pytest.LogCaptureFixture
375) -> None:
376    """Test that files that cannot be read leave the scan incomplete."""
377    _build_music_tree(tmp_path)
378    errors = helpers.ScanErrors()
379    real_from_dir_entry = helpers.FileSystemItem.from_dir_entry
380
381    def fake_from_dir_entry(entry: os.DirEntry[str], base_path: str) -> helpers.FileSystemItem:
382        if entry.name.startswith("track1") or entry.name.startswith("track2"):
383            raise OSError(errno.EIO, "i/o error")
384        return real_from_dir_entry(entry, base_path)
385
386    with (
387        caplog.at_level(logging.DEBUG, logger="test"),
388        patch.object(helpers.FileSystemItem, "from_dir_entry", fake_from_dir_entry),
389    ):
390        items = list(
391            helpers.recursive_iter(
392                str(tmp_path),
393                str(tmp_path),
394                SUPPORTED,
395                logging.getLogger("test"),
396                errors,
397            )
398        )
399
400    assert [item.relative_path for item in items] == ["Artist2/track3.mp3"]
401    assert not errors.aborted
402    assert errors.failed_dirs == 0
403    # the files are still on disk, so callers must not run deletions
404    assert errors.failed_entries == 2
405    assert errors.incomplete
406    # the summary names them so the user does not need the log to find them
407    assert "Artist1/Album1/track1.mp3" in errors.describe()
408    # both files failed in the same folder, so only the first one is a warning
409    warnings = [rec for rec in caplog.records if rec.levelno == logging.WARNING]
410    assert len(warnings) == 1
411
412
413@pytest.mark.parametrize("err", [OSError(errno.ENOENT, "gone"), OSError(errno.EINVAL, "invalid")])
414def test_recursive_iter_vanished_file_is_ignored(tmp_path: Path, err: OSError) -> None:
415    """Test that a file that is really gone does not block deletions."""
416    _build_music_tree(tmp_path)
417    errors = helpers.ScanErrors()
418    real_from_dir_entry = helpers.FileSystemItem.from_dir_entry
419
420    def fake_from_dir_entry(entry: os.DirEntry[str], base_path: str) -> helpers.FileSystemItem:
421        if entry.name == "track1.mp3":
422            raise err
423        return real_from_dir_entry(entry, base_path)
424
425    with patch.object(helpers.FileSystemItem, "from_dir_entry", fake_from_dir_entry):
426        items = list(
427            helpers.recursive_iter(
428                str(tmp_path),
429                str(tmp_path),
430                SUPPORTED,
431                logging.getLogger("test"),
432                errors,
433            )
434        )
435
436    assert "Artist1/Album1/track1.mp3" not in [item.relative_path for item in items]
437    assert not errors.incomplete
438
439
440def test_recursive_iter_unreadable_entry_type_is_recorded(tmp_path: Path) -> None:
441    """Test that an entry of unknown type leaves the scan incomplete."""
442    errors = helpers.ScanErrors()
443    entries = [_BrokenEntry(str(tmp_path / "Album1"), OSError(errno.EIO, "i/o error"))]
444
445    with patch("os.scandir", return_value=_FakeScanDir(entries)):
446        items = list(
447            helpers.recursive_iter(
448                str(tmp_path),
449                str(tmp_path),
450                SUPPORTED,
451                logging.getLogger("test"),
452                errors,
453            )
454        )
455
456    assert items == []
457    assert not errors.aborted
458    # the entry may be a folder full of tracks, so callers must not run deletions
459    assert errors.failed_entries == 1
460
461
462def test_scan_errors_describe_names_examples() -> None:
463    """Test that the summary names the failed paths it kept."""
464    errors = helpers.ScanErrors()
465    errors.record_dir_error(OSError(errno.EIO, "i/o error"), is_root=False, path="Artist1/Album1")
466    for index in range(helpers.MAX_REPORTED_FAILED_PATHS + 5):
467        errors.record_entry_error(OSError(errno.EIO, "i/o error"), f"Artist2/track{index}.mp3")
468
469    summary = errors.describe()
470    assert "1 folder(s)" in summary
471    assert f"{helpers.MAX_REPORTED_FAILED_PATHS + 5} file(s)" in summary
472    assert "Artist1/Album1" in summary
473    # only the first few paths are named, the counts carry the rest
474    assert len(errors.failed_paths) == helpers.MAX_REPORTED_FAILED_PATHS
475
476
477def test_scan_errors_reset_on_successful_read() -> None:
478    """Test that a directory read in between failures resets the abort threshold."""
479    errors = helpers.ScanErrors()
480    err = OSError(errno.EIO, "i/o error")
481    for _ in range(helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1):
482        errors.record_dir_error(err, is_root=False)
483    assert not errors.aborted
484
485    errors.record_dir_read()
486    assert errors.consecutive_failures == 0
487
488    for _ in range(helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1):
489        errors.record_dir_error(err, is_root=False)
490    assert not errors.aborted
491    assert errors.failed_dirs == (helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1) * 2
492