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_recursive_iter_skips_names_that_are_not_valid_utf8(
478    tmp_path: Path, caplog: pytest.LogCaptureFixture
479) -> None:
480    """
481    Test that a filename which is not valid UTF-8 is skipped, naming it escaped.
482
483    Its path can be neither stored nor serialized, so letting it through only fails
484    deeper down, taking the events to the clients and the settings file with it
485    (#6042). Emoji are valid UTF-8 and must keep scanning.
486    """
487    (tmp_path / "track 🎧.mp3").write_bytes(b"x")
488    # 0xDF is "ß" in Latin-1 and not valid UTF-8, so os returns it surrogate-escaped
489    undecodable = os.path.join(os.fsencode(str(tmp_path)), b"Stra\xdfe.mp3")
490    with open(undecodable, "wb") as _file:
491        _file.write(b"x")
492
493    errors = helpers.ScanErrors()
494    with caplog.at_level(logging.WARNING):
495        items = list(
496            helpers.recursive_iter(
497                str(tmp_path), str(tmp_path), SUPPORTED, logging.getLogger("test"), errors
498            )
499        )
500
501    assert [item.relative_path for item in items] == ["track 🎧.mp3"]
502    assert "Stra\\xdfe.mp3" in caplog.text
503    # such a file can never have been indexed, so skipping it must not block deletions
504    assert not errors.incomplete
505
506
507def test_sorted_scandir_skips_names_that_are_not_valid_utf8(
508    tmp_path: Path, caplog: pytest.LogCaptureFixture
509) -> None:
510    """
511    Test that the directory listing skips a filename which is not valid UTF-8.
512
513    This listing feeds browse, podcast episodes, playlist and folder images, audiobooks
514    and chapters, so an item that can not be serialized would fail whichever of those
515    asked for it (#6042).
516    """
517    (tmp_path / "track 🎧.mp3").write_bytes(b"x")
518    undecodable = os.path.join(os.fsencode(str(tmp_path)), b"Stra\xdfe.mp3")
519    with open(undecodable, "wb") as _file:
520        _file.write(b"x")
521
522    with caplog.at_level(logging.WARNING):
523        items = helpers.sorted_scandir(str(tmp_path), str(tmp_path))
524
525    assert [item.relative_path for item in items] == ["track 🎧.mp3"]
526    assert "Stra\\xdfe.mp3" in caplog.text
527
528
529def test_scan_errors_reset_on_successful_read() -> None:
530    """Test that a directory read in between failures resets the abort threshold."""
531    errors = helpers.ScanErrors()
532    err = OSError(errno.EIO, "i/o error")
533    for _ in range(helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1):
534        errors.record_dir_error(err, is_root=False)
535    assert not errors.aborted
536
537    errors.record_dir_read()
538    assert errors.consecutive_failures == 0
539
540    for _ in range(helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1):
541        errors.record_dir_error(err, is_root=False)
542    assert not errors.aborted
543    assert errors.failed_dirs == (helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1) * 2
544