music-assistant-server

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