/
/
/
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.helpers.compare import compare_strings
14from music_assistant.providers.filesystem_local import helpers
15
16# ruff: noqa: S108
17
18
19def test_get_artist_dir() -> None:
20 """Test the extraction of an artist dir."""
21 album_path = "/tmp/Artist/Album"
22 artist_name = "Artist"
23 assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/Artist"
24 album_path = "/tmp/artist/Album"
25 assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/artist"
26 album_path = "/tmp/Album"
27 assert helpers.get_artist_dir(artist_name, album_path) is None
28 album_path = "/tmp/ARTIST!/Album"
29 assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/ARTIST!"
30 album_path = "/tmp/Artist/Album"
31 artist_name = "Artist!"
32 assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/Artist"
33 album_path = "/tmp/REM/Album"
34 artist_name = "R.E.M."
35 assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/REM"
36 album_path = "/tmp/ACDC/Album"
37 artist_name = "AC/DC"
38 assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/ACDC"
39 album_path = "/tmp/Celine Dion/Album"
40 artist_name = "Céline Dion"
41 assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/Celine Dion"
42 album_path = "/tmp/Antonin Dvorak/Album"
43 artist_name = "AntonÃn DvoÅák"
44 assert helpers.get_artist_dir(artist_name, album_path) == "/tmp/Antonin Dvorak"
45
46
47@pytest.mark.parametrize(
48 ("album_name", "track_dir", "expected"),
49 [
50 # Test literal match
51 (
52 "Selected Ambient Works 85-92",
53 "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92",
54 "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92",
55 ),
56 # Test artist - album format
57 (
58 "Selected Ambient Works 85-92",
59 "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92",
60 "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92",
61 ),
62 # Test artist - album (version) format
63 (
64 "Selected Ambient Works 85-92",
65 "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92 (Remastered)",
66 "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92 (Remastered)",
67 ),
68 # Test artist - album (version) format
69 (
70 "Selected Ambient Works 85-92",
71 "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92 (Remastered) - WEB",
72 "/home/user/Music/Aphex Twin - Selected Ambient Works 85-92 (Remastered) - WEB",
73 ),
74 # Test tokenizer - dirname with extras
75 (
76 "Fokus - Prewersje",
77 "/home/user/Fokus-Prewersje-PL-WEB-FLAC-2021-PS_INT",
78 "/home/user/Fokus-Prewersje-PL-WEB-FLAC-2021-PS_INT",
79 ),
80 # Test tokenizer - dirname with version and extras
81 (
82 "Layo And Bushwacka - Night Works",
83 "/home/music/Layo_And_Bushwacka-Night_Works_(Reissue)-(XLCD_154X)-FLAC-2003",
84 "/home/music/Layo_And_Bushwacka-Night_Works_(Reissue)-(XLCD_154X)-FLAC-2003",
85 ),
86 # Test tokenizer - extras and approximate match on diacratics
87 (
88 "Åona i Webber - WyÅlij Sobie PocztówkÄ",
89 "/usr/others/Lona-Discography-PL-FLAC-2020-INT/Lona_I_Webber-Wyslij_Sobie_Pocztowke-PL-WEB-FLAC-2014-PS",
90 "/usr/others/Lona-Discography-PL-FLAC-2020-INT/Lona_I_Webber-Wyslij_Sobie_Pocztowke-PL-WEB-FLAC-2014-PS",
91 ),
92 (
93 "NIC",
94 "/nas/downloads/others/Sokol-NIC-PL-WEB-FLAC-2021",
95 "/nas/downloads/others/Sokol-NIC-PL-WEB-FLAC-2021",
96 ),
97 # Test album (version) format
98 (
99 "Aphex Twin - Selected Ambient Works 85-92",
100 "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92 (Remastered)",
101 "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92 (Remastered)",
102 ),
103 # Test album name in dir
104 (
105 "Aphex Twin - Selected Ambient Works 85-92",
106 "/home/user/Music/RandomDirWithAphex Twin - Selected Ambient Works 85-92InIt",
107 "/home/user/Music/RandomDirWithAphex Twin - Selected Ambient Works 85-92InIt",
108 ),
109 # Test no match
110 (
111 "NonExistentAlbumName",
112 "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92",
113 None,
114 ),
115 # Test empty album name
116 ("", "/home/user/Music/Aphex Twin/Selected Ambient Works 85-92", None),
117 # Test empty track dir
118 ("Selected Ambient Works 85-92", "", None),
119 ],
120)
121def test_get_album_dir(album_name: str, track_dir: str, expected: str) -> None:
122 """Test the extraction of an album dir."""
123 assert helpers.get_album_dir(track_dir, album_name) == expected
124
125
126def test_get_album_dir_falls_back_to_sort_name_alias() -> None:
127 """When the plain album name does not match a folder, the sort-name alias is tried."""
128 track_dir = "/tmp/Artist/Wall, The"
129 assert helpers.get_album_dir(track_dir, "The Wall") is None
130 assert helpers.get_album_dir(track_dir, "The Wall", album_sort="Wall, The") == track_dir
131 # the plain name still wins when it matches, without needing the alias
132 assert helpers.get_album_dir("/tmp/Artist/The Wall", "The Wall", album_sort="Wall, The") == (
133 "/tmp/Artist/The Wall"
134 )
135
136
137def test_get_album_dir_plain_name_at_a_farther_level_outranks_a_nearer_sort_name_alias() -> None:
138 """
139 A nearer sort-name alias match must never outrank a farther, exact plain-name match.
140
141 Both levels are searched for the plain album name first; the sort-name alias is only
142 tried afterwards, and only if the plain name matched nowhere.
143 """
144 # "Wall, The" (the sort-name alias) is the track's own directory, one level nearer than
145 # "The Wall" (the plain, exact name) at its parent
146 track_dir = "/tmp/Artist/The Wall/Wall, The"
147 assert (
148 helpers.get_album_dir(track_dir, "The Wall", album_sort="Wall, The")
149 == "/tmp/Artist/The Wall"
150 )
151
152
153def test_get_artist_dir_falls_back_to_sort_name_alias() -> None:
154 """When the plain artist name does not match a folder, the sort-name alias is tried."""
155 album_path = "/tmp/Beatles, The/Album"
156 assert helpers.get_artist_dir("The Beatles", album_path) is None
157 assert helpers.get_artist_dir("The Beatles", album_path, sort_name="Beatles, The") == (
158 "/tmp/Beatles, The"
159 )
160
161
162def test_get_artist_dir_plain_name_outranks_a_farther_sort_name_alias() -> None:
163 """
164 A farther sort-name alias match must never outrank a nearer, exact plain-name match.
165
166 The plain artist name's own bounded (up to 3 ancestor levels) search completes in full
167 before the sort-name alias is tried at all.
168 """
169 # "The Beatles" (the plain, exact name) is the immediate parent; "Beatles, The" (the
170 # sort-name alias) is one level further up
171 album_path = "/tmp/Beatles, The/The Beatles/Album"
172 assert (
173 helpers.get_artist_dir("The Beatles", album_path, sort_name="Beatles, The")
174 == "/tmp/Beatles, The/The Beatles"
175 )
176
177
178def test_get_artist_dir_exact_only_ignores_the_sort_name_alias_and_fuzzy_matches() -> None:
179 """`exact_only` skips the sort-name alias and the relaxed (fuzzy) comparison entirely."""
180 # the sort-name alias itself is a relaxed heuristic and must not be tried
181 album_path = "/tmp/Beatles, The/Album"
182 assert (
183 helpers.get_artist_dir("The Beatles", album_path, sort_name="Beatles, The", exact_only=True)
184 is None
185 )
186 # a near-miss that only the fuzzy (non-strict) comparison would accept must not match
187 album_path = "/tmp/The Beetles/Album"
188 assert helpers.get_artist_dir("The Beatles", album_path, exact_only=True) is None
189 assert helpers.get_artist_dir("The Beatles", album_path) == "/tmp/The Beetles"
190 # an exact (normalized) match still succeeds
191 album_path = "/tmp/The Beatles/Album"
192 assert helpers.get_artist_dir("The Beatles", album_path, exact_only=True) == (
193 "/tmp/The Beatles"
194 )
195
196
197@pytest.mark.parametrize(
198 ("dirname", "expected"),
199 [
200 ("2025-03-14 Vaxis Act III The Father of Make Believe", True),
201 ("2025.03.14 Vaxis Act III The Father of Make Believe", True),
202 ("1995-03-13 Vaxis Act III The Father of Make Believe", True),
203 ("2025-03-14 VAXIS ACT III THE FATHER OF MAKE BELIEVE", True),
204 ("(2025) Vaxis Act III The Father of Make Believe", True),
205 ("[2025] Vaxis Act III The Father of Make Believe", True),
206 ("2025 Vaxis Act III The Father of Make Believe", True),
207 ("Vaxis Act III The Father of Make Believe", True),
208 ],
209)
210def test_dir_matches_album_strips_a_recognized_date_prefix(dirname: str, expected: bool) -> None:
211 """A leading release date/year, in any recognized format or case, is not part of the title."""
212 assert helpers._dir_matches_album(dirname, "Vaxis Act III: The Father of Make Believe") is (
213 expected
214 )
215
216
217def test_dir_matches_album_date_prefixed_king_for_a_day_example() -> None:
218 """The second #3994 reproduction: a date-prefixed folder with an ellipsis-free title."""
219 assert helpers._dir_matches_album(
220 "1995-03-13 King for a Day Fool for a Lifetime",
221 "King for a Day... Fool for a Lifetime",
222 )
223
224
225def test_dir_matches_album_king_for_a_day_example_without_date_prefix() -> None:
226 """The #3994 examples without a date prefix already matched before this change too."""
227 assert helpers._dir_matches_album(
228 "King for a Day Fool for a Lifetime",
229 "King for a Day... Fool for a Lifetime",
230 )
231
232
233def test_strip_date_prefix_does_not_touch_an_arbitrary_catalogue_prefix() -> None:
234 """A catalogue prefix like "CAT-1234" does not start with a 4-digit date/year at all."""
235 assert helpers._strip_date_prefix("CAT-1234 Album Name") == "CAT-1234 Album Name"
236
237
238def test_strip_date_prefix_requires_a_real_separator_after_the_year() -> None:
239 """A bare year glued directly onto the title (no separator) is not stripped."""
240 assert helpers._strip_date_prefix("2025Album") == "2025Album"
241
242
243def test_dir_matches_album_date_prefix_path_rejects_reordered_words() -> None:
244 """The new date-prefix comparison is strict normalized equality, not token matching."""
245 stripped = helpers._strip_date_prefix("2025-03-14 Beta Alpha")
246 assert stripped == "Beta Alpha"
247 assert compare_strings("Alpha Beta", stripped, True) is False
248
249
250@pytest.mark.parametrize(
251 ("name", "expected"),
252 [
253 ("Disc 1", True),
254 ("disc1", True),
255 ("CD2", True),
256 ("cd 03", True),
257 ("Disk 1", True),
258 ("DVD1", True),
259 ("Volume 2", True),
260 ("Vol. 2", True),
261 ("Album", False),
262 ("weird-disc-name", False),
263 ("", False),
264 ],
265)
266def test_is_disc_dir(name: str, expected: bool) -> None:
267 """Only a recognized disc/volume naming pattern is treated as a disc subfolder."""
268 assert helpers.is_disc_dir(name) is expected
269
270
271def test_parse_nfo_root_returns_the_named_root_element() -> None:
272 """A well-formed NFO with the expected root element returns it as a dict."""
273 data = b"<album><title>My Album</title></album>"
274 root = helpers.parse_nfo_root(data, "album")
275 assert root is not None
276 assert root["title"] == "My Album"
277
278
279@pytest.mark.parametrize(
280 ("data", "root_tag"),
281 [
282 (b"not xml at all <<<", "album"),
283 (b"<artist><title>Name</title></artist>", "album"), # wrong root element
284 (b"\xff\xfe not utf-8", "album"),
285 (b"<album>just text, no dict</album>", "album"),
286 ],
287)
288def test_parse_nfo_root_returns_none_for_malformed_content(data: bytes, root_tag: str) -> None:
289 """Malformed XML, an undecodable file or the wrong/non-dict root element returns None."""
290 assert helpers.parse_nfo_root(data, root_tag) is None
291
292
293SUPPORTED = {"mp3", "flac"}
294
295
296def _build_music_tree(root: Path) -> None:
297 """Create a small music tree fixture."""
298 (root / "Artist1" / "Album1").mkdir(parents=True)
299 (root / "Artist1" / "Album1" / "track1.mp3").write_bytes(b"x")
300 (root / "Artist1" / "Album1" / "track2.flac").write_bytes(b"x")
301 (root / "Artist2").mkdir()
302 (root / "Artist2" / "track3.mp3").write_bytes(b"x")
303
304
305def test_recursive_iter_happy_path(tmp_path: Path) -> None:
306 """Test that a healthy scan yields all supported files and records no errors."""
307 _build_music_tree(tmp_path)
308 errors = helpers.ScanErrors()
309 items = list(
310 helpers.recursive_iter(
311 str(tmp_path),
312 str(tmp_path),
313 SUPPORTED,
314 logging.getLogger("test"),
315 errors,
316 )
317 )
318 rel_paths = sorted(i.relative_path for i in items)
319 assert rel_paths == [
320 "Artist1/Album1/track1.mp3",
321 "Artist1/Album1/track2.flac",
322 "Artist2/track3.mp3",
323 ]
324 assert not errors.fatal
325 assert errors.failed_dirs == 0
326
327
328def test_recursive_iter_root_unreachable_records_error(tmp_path: Path) -> None:
329 """Test that a missing root path is reported via scan_errors."""
330 errors = helpers.ScanErrors()
331 missing = tmp_path / "does-not-exist"
332 items = list(
333 helpers.recursive_iter(
334 str(missing),
335 str(missing),
336 SUPPORTED,
337 logging.getLogger("test"),
338 errors,
339 )
340 )
341 assert items == []
342 assert isinstance(errors.fatal, OSError)
343 assert errors.fatal.errno == errno.ENOENT
344
345
346def test_recursive_iter_root_eacces_records_error() -> None:
347 """Test that permission-denied on the root path is reported via scan_errors."""
348 errors = helpers.ScanErrors()
349 with patch("os.scandir", side_effect=PermissionError(errno.EACCES, "denied")):
350 items = list(
351 helpers.recursive_iter(
352 "/fake/root",
353 "/fake/root",
354 SUPPORTED,
355 logging.getLogger("test"),
356 errors,
357 )
358 )
359 assert items == []
360 assert isinstance(errors.fatal, OSError)
361 assert errors.fatal.errno == errno.EACCES
362
363
364def test_recursive_iter_subfolder_failure_is_not_fatal(tmp_path: Path) -> None:
365 """Test that a single sub-folder scan failure is not fatal."""
366 _build_music_tree(tmp_path)
367 errors = helpers.ScanErrors()
368 real_scandir = os.scandir
369 bad_dir = str(tmp_path / "Artist1" / "Album1")
370
371 def fake_scandir(path: str | os.PathLike[str]): # type: ignore[no-untyped-def]
372 if str(path) == bad_dir:
373 raise OSError(errno.EIO, "i/o error")
374 return real_scandir(path)
375
376 with patch("os.scandir", side_effect=fake_scandir):
377 items = list(
378 helpers.recursive_iter(
379 str(tmp_path),
380 str(tmp_path),
381 SUPPORTED,
382 logging.getLogger("test"),
383 errors,
384 )
385 )
386
387 rel_paths = sorted(i.relative_path for i in items)
388 assert rel_paths == ["Artist2/track3.mp3"]
389 assert not errors.fatal
390 # the scan is incomplete, so callers must not run deletions
391 assert errors.failed_dirs == 1
392
393
394def test_recursive_iter_einval_is_ignored() -> None:
395 """Test that EINVAL from an unsupported path name is not recorded."""
396 errors = helpers.ScanErrors()
397 with patch("os.scandir", side_effect=OSError(errno.EINVAL, "invalid path")):
398 items = list(
399 helpers.recursive_iter(
400 "/weird/\udcff",
401 "/weird/\udcff",
402 SUPPORTED,
403 logging.getLogger("test"),
404 errors,
405 )
406 )
407 assert items == []
408 assert not errors.fatal
409 assert errors.failed_dirs == 0
410
411
412def _build_flat_tree(root: Path, count: int) -> None:
413 """Create a music tree with the given number of album folders."""
414 for index in range(count):
415 album_dir = root / f"Album{index:03d}"
416 album_dir.mkdir()
417 (album_dir / "track.mp3").write_bytes(b"x")
418
419
420def test_recursive_iter_aborts_after_consecutive_failures(tmp_path: Path) -> None:
421 """Test that storage disappearing mid-scan aborts the walk instead of grinding on."""
422 _build_flat_tree(tmp_path, helpers.MAX_CONSECUTIVE_SCAN_ERRORS + 10)
423 errors = helpers.ScanErrors()
424 real_scandir = os.scandir
425
426 def fake_scandir(path: str | os.PathLike[str]): # type: ignore[no-untyped-def]
427 if str(path) == str(tmp_path):
428 return real_scandir(path)
429 raise OSError(errno.EIO, "i/o error")
430
431 with patch("os.scandir", side_effect=fake_scandir):
432 items = list(
433 helpers.recursive_iter(
434 str(tmp_path),
435 str(tmp_path),
436 SUPPORTED,
437 logging.getLogger("test"),
438 errors,
439 )
440 )
441
442 assert items == []
443 assert errors.aborted
444 # the walk stopped at the threshold instead of trying every remaining folder
445 assert errors.failed_dirs == helpers.MAX_CONSECUTIVE_SCAN_ERRORS
446
447
448def test_recursive_iter_einval_does_not_abort(tmp_path: Path) -> None:
449 """Test that skipped (unsupported) path names never trip the abort threshold."""
450 _build_flat_tree(tmp_path, helpers.MAX_CONSECUTIVE_SCAN_ERRORS + 10)
451 (tmp_path / "root.mp3").write_bytes(b"x")
452 errors = helpers.ScanErrors()
453 real_scandir = os.scandir
454
455 def fake_scandir(path: str | os.PathLike[str]): # type: ignore[no-untyped-def]
456 if str(path) == str(tmp_path):
457 return real_scandir(path)
458 raise OSError(errno.EINVAL, "invalid argument")
459
460 with patch("os.scandir", side_effect=fake_scandir):
461 items = list(
462 helpers.recursive_iter(
463 str(tmp_path),
464 str(tmp_path),
465 SUPPORTED,
466 logging.getLogger("test"),
467 errors,
468 )
469 )
470
471 assert [item.relative_path for item in items] == ["root.mp3"]
472 assert not errors.aborted
473 assert errors.failed_dirs == 0
474
475
476def test_recursive_iter_permission_denied_does_not_abort(tmp_path: Path) -> None:
477 """Test that ACL-protected folders leave the scan incomplete without aborting it."""
478 _build_flat_tree(tmp_path, helpers.MAX_CONSECUTIVE_SCAN_ERRORS + 10)
479 (tmp_path / "root.mp3").write_bytes(b"x")
480 errors = helpers.ScanErrors()
481 real_scandir = os.scandir
482
483 def fake_scandir(path: str | os.PathLike[str]): # type: ignore[no-untyped-def]
484 if str(path) == str(tmp_path):
485 return real_scandir(path)
486 raise PermissionError(errno.EACCES, "denied")
487
488 with patch("os.scandir", side_effect=fake_scandir):
489 items = list(
490 helpers.recursive_iter(
491 str(tmp_path),
492 str(tmp_path),
493 SUPPORTED,
494 logging.getLogger("test"),
495 errors,
496 )
497 )
498
499 assert [item.relative_path for item in items] == ["root.mp3"]
500 assert not errors.aborted
501 assert errors.consecutive_failures == 0
502 # the folders were still missed, so callers must not run deletions
503 assert errors.failed_dirs == helpers.MAX_CONSECUTIVE_SCAN_ERRORS + 10
504
505
506class _BrokenEntry:
507 """Directory entry whose type check fails, as on a share that drops mid-listing."""
508
509 def __init__(self, path: str, err: OSError) -> None:
510 self.name = Path(path).name
511 self.path = path
512 self._err = err
513
514 def is_dir(self, follow_symlinks: bool = True) -> bool:
515 """Raise the configured error instead of answering."""
516 raise self._err
517
518 def is_file(self, follow_symlinks: bool = True) -> bool:
519 """Raise the configured error instead of answering."""
520 raise self._err
521
522
523class _NamedEntry:
524 """Directory entry carrying only a name, for names a filesystem may refuse to create."""
525
526 def __init__(self, parent: str, name: str) -> None:
527 self.name = name
528 self.path = os.path.join(parent, name)
529
530 # no is_dir/is_file on purpose: the name guard has to skip this entry before anything
531 # reads its type, so a call site that lost the guard fails loudly here instead of
532 # quietly dropping the entry and leaving the test green
533 def __getattr__(self, name: str) -> object:
534 raise AssertionError(f"'{self.name}' must be skipped on its name, before .{name}")
535
536
537_ScanEntry = _BrokenEntry | _NamedEntry | os.DirEntry[str]
538
539
540class _FakeScanDir:
541 """Stand-in for the os.scandir iterator, which is also a context manager."""
542
543 def __init__(self, entries: Sequence[_ScanEntry]) -> None:
544 self._entries = iter(entries)
545
546 def __enter__(self) -> Self:
547 return self
548
549 def __exit__(self, *_exc: object) -> None:
550 return None
551
552 def __iter__(self) -> Self:
553 return self
554
555 def __next__(self) -> _ScanEntry:
556 return next(self._entries)
557
558
559def test_recursive_iter_unreadable_file_is_recorded(
560 tmp_path: Path, caplog: pytest.LogCaptureFixture
561) -> None:
562 """Test that files that cannot be read leave the scan incomplete."""
563 _build_music_tree(tmp_path)
564 errors = helpers.ScanErrors()
565 real_from_dir_entry = helpers.FileSystemItem.from_dir_entry
566
567 def fake_from_dir_entry(entry: os.DirEntry[str], base_path: str) -> helpers.FileSystemItem:
568 if entry.name.startswith("track1") or entry.name.startswith("track2"):
569 raise OSError(errno.EIO, "i/o error")
570 return real_from_dir_entry(entry, base_path)
571
572 with (
573 caplog.at_level(logging.DEBUG, logger="test"),
574 patch.object(helpers.FileSystemItem, "from_dir_entry", fake_from_dir_entry),
575 ):
576 items = list(
577 helpers.recursive_iter(
578 str(tmp_path),
579 str(tmp_path),
580 SUPPORTED,
581 logging.getLogger("test"),
582 errors,
583 )
584 )
585
586 assert [item.relative_path for item in items] == ["Artist2/track3.mp3"]
587 assert not errors.aborted
588 assert errors.failed_dirs == 0
589 # the files are still on disk, so callers must not run deletions
590 assert errors.failed_entries == 2
591 assert errors.incomplete
592 # the summary names them so the user does not need the log to find them
593 assert "Artist1/Album1/track1.mp3" in errors.describe()
594 # both files failed in the same folder, so only the first one is a warning
595 warnings = [rec for rec in caplog.records if rec.levelno == logging.WARNING]
596 assert len(warnings) == 1
597
598
599@pytest.mark.parametrize("err", [OSError(errno.ENOENT, "gone"), OSError(errno.EINVAL, "invalid")])
600def test_recursive_iter_vanished_file_is_ignored(tmp_path: Path, err: OSError) -> None:
601 """Test that a file that is really gone does not block deletions."""
602 _build_music_tree(tmp_path)
603 errors = helpers.ScanErrors()
604 real_from_dir_entry = helpers.FileSystemItem.from_dir_entry
605
606 def fake_from_dir_entry(entry: os.DirEntry[str], base_path: str) -> helpers.FileSystemItem:
607 if entry.name == "track1.mp3":
608 raise err
609 return real_from_dir_entry(entry, base_path)
610
611 with patch.object(helpers.FileSystemItem, "from_dir_entry", fake_from_dir_entry):
612 items = list(
613 helpers.recursive_iter(
614 str(tmp_path),
615 str(tmp_path),
616 SUPPORTED,
617 logging.getLogger("test"),
618 errors,
619 )
620 )
621
622 assert "Artist1/Album1/track1.mp3" not in [item.relative_path for item in items]
623 assert not errors.incomplete
624
625
626def test_recursive_iter_unreadable_entry_type_is_recorded(tmp_path: Path) -> None:
627 """Test that an entry of unknown type leaves the scan incomplete."""
628 errors = helpers.ScanErrors()
629 entries = [_BrokenEntry(str(tmp_path / "Album1"), OSError(errno.EIO, "i/o error"))]
630
631 with patch("os.scandir", return_value=_FakeScanDir(entries)):
632 items = list(
633 helpers.recursive_iter(
634 str(tmp_path),
635 str(tmp_path),
636 SUPPORTED,
637 logging.getLogger("test"),
638 errors,
639 )
640 )
641
642 assert items == []
643 assert not errors.aborted
644 # the entry may be a folder full of tracks, so callers must not run deletions
645 assert errors.failed_entries == 1
646
647
648def test_scan_errors_describe_names_examples() -> None:
649 """Test that the summary names the failed paths it kept."""
650 errors = helpers.ScanErrors()
651 errors.record_dir_error(OSError(errno.EIO, "i/o error"), is_root=False, path="Artist1/Album1")
652 for index in range(helpers.MAX_REPORTED_FAILED_PATHS + 5):
653 errors.record_entry_error(OSError(errno.EIO, "i/o error"), f"Artist2/track{index}.mp3")
654
655 summary = errors.describe()
656 assert "1 folder(s)" in summary
657 assert f"{helpers.MAX_REPORTED_FAILED_PATHS + 5} file(s)" in summary
658 assert "Artist1/Album1" in summary
659 # only the first few paths are named, the counts carry the rest
660 assert len(errors.failed_paths) == helpers.MAX_REPORTED_FAILED_PATHS
661
662
663# 0xDF is "Ã" in Latin-1 and not valid UTF-8. os.fsdecode is what os.scandir uses to build
664# DirEntry.name, so this is exactly what a real scan hands the guard for such a file, while
665# needing no file on disk - filesystems that enforce UTF-8 names refuse to create one.
666UNDECODABLE_NAME = os.fsdecode(b"Stra\xdfe.mp3")
667
668
669def test_recursive_iter_skips_names_that_are_not_valid_utf8(
670 tmp_path: Path, caplog: pytest.LogCaptureFixture
671) -> None:
672 """
673 Test that a filename which is not valid UTF-8 is skipped, naming it escaped.
674
675 Its path can be neither stored nor serialized, so letting it through only fails
676 deeper down, taking the events to the clients and the settings file with it
677 (#6042). Emoji are valid UTF-8 and must keep scanning.
678 """
679 (tmp_path / "track ð§.mp3").write_bytes(b"x")
680 errors = helpers.ScanErrors()
681 real_scandir = os.scandir
682
683 def fake_scandir(path: str | os.PathLike[str]) -> _FakeScanDir:
684 with real_scandir(path) as entries:
685 return _FakeScanDir([*entries, _NamedEntry(str(path), UNDECODABLE_NAME)])
686
687 with (
688 caplog.at_level(logging.WARNING),
689 patch("os.scandir", side_effect=fake_scandir),
690 ):
691 items = list(
692 helpers.recursive_iter(
693 str(tmp_path), str(tmp_path), SUPPORTED, logging.getLogger("test"), errors
694 )
695 )
696
697 assert [item.relative_path for item in items] == ["track ð§.mp3"]
698 assert "Stra\\xdfe.mp3" in caplog.text
699 # such a file can never have been indexed, so skipping it must not block deletions
700 assert not errors.incomplete
701
702
703def test_sorted_scandir_skips_names_that_are_not_valid_utf8(
704 tmp_path: Path, caplog: pytest.LogCaptureFixture
705) -> None:
706 """
707 Test that the directory listing skips a filename which is not valid UTF-8.
708
709 This listing feeds browse, podcast episodes, playlist and folder images, audiobooks
710 and chapters, so an item that can not be serialized would fail whichever of those
711 asked for it (#6042).
712 """
713 (tmp_path / "track ð§.mp3").write_bytes(b"x")
714 real_scandir = os.scandir
715
716 def fake_scandir(path: str | os.PathLike[str]) -> _FakeScanDir:
717 with real_scandir(path) as entries:
718 return _FakeScanDir([*entries, _NamedEntry(str(path), UNDECODABLE_NAME)])
719
720 with (
721 caplog.at_level(logging.WARNING),
722 patch("os.scandir", side_effect=fake_scandir),
723 ):
724 items = helpers.sorted_scandir(str(tmp_path), str(tmp_path))
725
726 assert [item.relative_path for item in items] == ["track ð§.mp3"]
727 assert "Stra\\xdfe.mp3" in caplog.text
728
729
730def test_skip_undecodable_name_passes_valid_names(caplog: pytest.LogCaptureFixture) -> None:
731 """Test that the guard passes valid names without warning, whatever their encoding."""
732 log = logging.getLogger("test")
733 with caplog.at_level(logging.WARNING):
734 assert not helpers._skip_undecodable_name("track.mp3", log)
735 assert not helpers._skip_undecodable_name("StraÃe.mp3", log)
736 # 4-byte UTF-8 takes the same path as the 2-byte name above
737 assert not helpers._skip_undecodable_name("track ð§.mp3", log)
738
739 assert not caplog.records
740
741
742def test_scan_errors_reset_on_successful_read() -> None:
743 """Test that a directory read in between failures resets the abort threshold."""
744 errors = helpers.ScanErrors()
745 err = OSError(errno.EIO, "i/o error")
746 for _ in range(helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1):
747 errors.record_dir_error(err, is_root=False)
748 assert not errors.aborted
749
750 errors.record_dir_read()
751 assert errors.consecutive_failures == 0
752
753 for _ in range(helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1):
754 errors.record_dir_error(err, is_root=False)
755 assert not errors.aborted
756 assert errors.failed_dirs == (helpers.MAX_CONSECUTIVE_SCAN_ERRORS - 1) * 2
757