/
/
1"""Tests for the hourly duplicate track reconciliation maintenance task."""
2
3from __future__ import annotations
4
5from typing import NamedTuple
6from unittest.mock import AsyncMock, Mock, patch
7
8import pytest
9from music_assistant_models.enums import AlbumType, ExternalID, TaskStatus
10from music_assistant_models.errors import MediaNotFoundError, MusicAssistantError
11from music_assistant_models.helpers import create_safe_string
12from music_assistant_models.media_items import (
13 Album,
14 Artist,
15 MediaItemMetadata,
16 ProviderMapping,
17 Track,
18 UniqueList,
19)
20
21from music_assistant.constants import (
22 DB_TABLE_ALBUM_TRACKS,
23 DB_TABLE_ALBUMS,
24 DB_TABLE_TRACK_ARTISTS,
25 DB_TABLE_TRACKS,
26)
27from music_assistant.controllers.music.constants import TRACK_RECONCILIATION_BATCH_SIZE
28from music_assistant.controllers.music.controller import MusicController
29from music_assistant.mass import MusicAssistant
30
31_DUPLICATE_NAME = "Shared Track Title"
32_REPORT_FAILURE = "music_assistant.controllers.music.controller.report_current_task_failure"
33
34
35class _TrackSpec(NamedTuple):
36 """How one side of a duplicate pair should differ from the other."""
37
38 provider: str = "spotify_instance"
39 title: str = _DUPLICATE_NAME
40 album_name: str = "Shared Album"
41 duration: int = 200
42 version: str = ""
43 track_number: int = 1
44 explicit: bool | None = None
45 mbid: str | None = None
46
47
48def _mapping(provider_instance: str, item_id: str) -> ProviderMapping:
49 """Create a provider mapping for a library fixture item."""
50 return ProviderMapping(
51 item_id=item_id,
52 provider_domain=provider_instance.removesuffix("_instance"),
53 provider_instance=provider_instance,
54 in_library=True,
55 )
56
57
58async def _add_artist(mass: MusicAssistant, provider_instance: str) -> Artist:
59 """Create the shared fixture artist for a provider."""
60 return await mass.music.artists.add_item_to_library(
61 Artist(
62 item_id="0",
63 provider="library",
64 name="Shared Artist",
65 provider_mappings={_mapping(provider_instance, f"{provider_instance}-artist")},
66 )
67 )
68
69
70async def _add_album(
71 mass: MusicAssistant, provider_instance: str, artist: Artist, name: str = "Shared Album"
72) -> Album:
73 """Create a fixture album for a provider."""
74 slug = create_safe_string(name, True, True)
75 return await mass.music.albums.add_item_to_library(
76 Album(
77 item_id="0",
78 provider="library",
79 name=name,
80 album_type=AlbumType.ALBUM,
81 provider_mappings={_mapping(provider_instance, f"{provider_instance}-album-{slug}")},
82 external_ids={(ExternalID.BARCODE, f"{provider_instance}-barcode-{slug}")},
83 artists=UniqueList([artist]),
84 )
85 )
86
87
88async def _add_track(
89 mass: MusicAssistant,
90 provider_instance: str,
91 artist: Artist,
92 album: Album,
93 *,
94 name: str,
95 duration: int = 200,
96 version: str = "",
97 track_number: int = 1,
98 explicit: bool | None = None,
99 mbid: str | None = None,
100) -> Track:
101 """Create a fixture track under a unique name, so it is stored as its own row."""
102 return await mass.music.tracks.add_item_to_library(
103 Track(
104 item_id="0",
105 provider="library",
106 name=name,
107 version=version,
108 duration=duration,
109 metadata=MediaItemMetadata(explicit=explicit),
110 external_ids={(ExternalID.MB_RECORDING, mbid)} if mbid else set(),
111 provider_mappings={
112 _mapping(
113 provider_instance,
114 f"{provider_instance}-track-{create_safe_string(name, True, True)}",
115 )
116 },
117 artists=UniqueList([artist]),
118 album=album,
119 disc_number=1,
120 track_number=track_number,
121 )
122 )
123
124
125async def _make_titles_look_alike(mass: MusicAssistant, track: Track, title: str) -> None:
126 """
127 Give a track row its final title plus the shared normalized title.
128
129 Fixture tracks are inserted under unique titles so that insert-time matching keeps them
130 as separate rows; equalizing search_name afterwards reproduces the end state of a library
131 where that matching failed.
132 """
133 normalized = create_safe_string(_DUPLICATE_NAME, True, True)
134 await mass.music.database.update(
135 DB_TABLE_TRACKS,
136 {"item_id": int(track.item_id)},
137 {
138 "name": title,
139 "sort_name": title.lower(),
140 "search_name": normalized,
141 "search_sort_name": normalized,
142 },
143 )
144
145
146async def _build_duplicate_pair(
147 mass: MusicAssistant,
148 first: _TrackSpec | None = None,
149 second: _TrackSpec | None = None,
150) -> tuple[Track, Track]:
151 """Create two same-titled library tracks that differ only as the specs say."""
152 first = first or _TrackSpec()
153 second = second or _TrackSpec(provider="qobuz_instance")
154 artist = await _add_artist(mass, first.provider)
155 tracks: list[Track] = []
156 for index, spec in enumerate((first, second)):
157 album = await _add_album(mass, spec.provider, artist, name=spec.album_name)
158 track = await _add_track(
159 mass,
160 spec.provider,
161 artist,
162 album,
163 name=f"Fixture Title {index}",
164 duration=spec.duration,
165 version=spec.version,
166 track_number=spec.track_number,
167 explicit=spec.explicit,
168 mbid=spec.mbid,
169 )
170 await _make_titles_look_alike(mass, track, spec.title)
171 tracks.append(track)
172 return tracks[0], tracks[1]
173
174
175def _bare_controller(candidate_rows: list[dict[str, int]]) -> MusicController:
176 """Create a bare MusicController whose candidate query returns the given rows."""
177 ctrl = MusicController.__new__(MusicController)
178 ctrl._track_reconciliation_cursor = (0, 0)
179 ctrl._track_reconciliation_rescan_due = False
180 ctrl.logger = Mock()
181 ctrl._database = Mock(get_rows_from_query=AsyncMock(return_value=candidate_rows))
182 ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])))
183 return ctrl
184
185
186# --------------------------------------------------------------------------- #
187# candidate selection and merging (against a real library database) #
188# --------------------------------------------------------------------------- #
189
190
191async def test_merges_cross_provider_duplicate_tracks(mass: MusicAssistant) -> None:
192 """Two same-titled tracks on the same album position are merged into one row."""
193 track_1, track_2 = await _build_duplicate_pair(mass)
194
195 await mass.music._reconcile_duplicate_tracks()
196
197 surviving = await mass.music.tracks.get_library_item(track_1.item_id)
198 assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
199 "spotify",
200 "qobuz",
201 }
202 with pytest.raises(MediaNotFoundError):
203 await mass.music.tracks.get_library_item(track_2.item_id)
204
205
206async def test_keeps_different_versions_apart(mass: MusicAssistant) -> None:
207 """A remaster is never merged into the original recording."""
208 track_1, track_2 = await _build_duplicate_pair(
209 mass, second=_TrackSpec("qobuz_instance", version="Remastered 2011")
210 )
211
212 await mass.music._reconcile_duplicate_tracks()
213
214 assert await mass.music.tracks.get_library_item(track_1.item_id)
215 assert await mass.music.tracks.get_library_item(track_2.item_id)
216
217
218async def test_merges_despite_disagreement_on_the_explicit_flag(mass: MusicAssistant) -> None:
219 """Providers routinely disagree on the explicit flag; that alone must not block a merge."""
220 track_1, track_2 = await _build_duplicate_pair(
221 mass, _TrackSpec(explicit=False), _TrackSpec("qobuz_instance", explicit=True)
222 )
223
224 await mass.music._reconcile_duplicate_tracks()
225
226 surviving = await mass.music.tracks.get_library_item(track_1.item_id)
227 assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
228 "spotify",
229 "qobuz",
230 }
231 with pytest.raises(MediaNotFoundError):
232 await mass.music.tracks.get_library_item(track_2.item_id)
233
234
235async def test_keeps_an_album_edition_apart_from_the_original(mass: MusicAssistant) -> None:
236 """An edition lives on the album, not the track, so equal track titles are not enough."""
237 track_1, track_2 = await _build_duplicate_pair(mass)
238 album_id = (
239 await mass.music.database.get_rows(
240 DB_TABLE_ALBUM_TRACKS, {"track_id": int(track_2.item_id)}
241 )
242 )[0]["album_id"]
243 await mass.music.database.update(
244 DB_TABLE_ALBUMS, {"item_id": album_id}, {"version": "2014 Remaster"}
245 )
246
247 await mass.music._reconcile_duplicate_tracks()
248
249 assert await mass.music.tracks.get_library_item(track_1.item_id)
250 assert await mass.music.tracks.get_library_item(track_2.item_id)
251
252
253async def test_an_unrelated_appearance_cannot_approve_an_edition(mass: MusicAssistant) -> None:
254 """Only the album appearance the pair shares a position on may vouch for the edition."""
255 track_1, track_2 = await _build_duplicate_pair(mass)
256 artist = (await mass.music.artists.get_library_items_by_query(limit=1))[0]
257 # the appearance that made them candidates is an original against a remaster
258 for track, version in ((track_1, "Deluxe Edition"), (track_2, "2014 Remaster")):
259 album_id = (
260 await mass.music.database.get_rows(
261 DB_TABLE_ALBUM_TRACKS, {"track_id": int(track.item_id)}
262 )
263 )[0]["album_id"]
264 await mass.music.database.update(
265 DB_TABLE_ALBUMS, {"item_id": album_id}, {"version": version}
266 )
267 # both also turn up on one compilation, but at different positions on it
268 compilation = await _add_album(mass, "spotify_instance", artist, name="Some Compilation")
269 for track, track_number in ((track_1, 5), (track_2, 9)):
270 await mass.music.database.insert(
271 DB_TABLE_ALBUM_TRACKS,
272 {
273 "track_id": int(track.item_id),
274 "album_id": int(compilation.item_id),
275 "disc_number": 1,
276 "track_number": track_number,
277 },
278 )
279
280 await mass.music._reconcile_duplicate_tracks()
281
282 assert await mass.music.tracks.get_library_item(track_1.item_id)
283 assert await mass.music.tracks.get_library_item(track_2.item_id)
284
285
286async def test_merges_across_an_ignorable_album_edition(mass: MusicAssistant) -> None:
287 """A quality label like Hi-Res is not a different edition, so it must not block a merge."""
288 track_1, track_2 = await _build_duplicate_pair(mass)
289 album_id = (
290 await mass.music.database.get_rows(
291 DB_TABLE_ALBUM_TRACKS, {"track_id": int(track_2.item_id)}
292 )
293 )[0]["album_id"]
294 await mass.music.database.update(
295 DB_TABLE_ALBUMS, {"item_id": album_id}, {"version": "Hi-Res Version"}
296 )
297
298 await mass.music._reconcile_duplicate_tracks()
299
300 surviving = await mass.music.tracks.get_library_item(track_1.item_id)
301 assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
302 "spotify",
303 "qobuz",
304 }
305 with pytest.raises(MediaNotFoundError):
306 await mass.music.tracks.get_library_item(track_2.item_id)
307
308
309async def test_merges_titles_that_differ_only_in_punctuation(mass: MusicAssistant) -> None:
310 """A curly apostrophe against a straight one is the same title, so the rows still merge."""
311 track_1, track_2 = await _build_duplicate_pair(
312 mass,
313 _TrackSpec(title="You\u2019re My Best Friend"),
314 _TrackSpec("qobuz_instance", title="You're My Best Friend"),
315 )
316
317 await mass.music._reconcile_duplicate_tracks()
318
319 surviving = await mass.music.tracks.get_library_item(track_1.item_id)
320 assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
321 "spotify",
322 "qobuz",
323 }
324 with pytest.raises(MediaNotFoundError):
325 await mass.music.tracks.get_library_item(track_2.item_id)
326
327
328async def test_keeps_distinct_recordings_apart(mass: MusicAssistant) -> None:
329 """Two different MusicBrainz recordings are different tracks, whatever else lines up."""
330 track_1, track_2 = await _build_duplicate_pair(
331 mass,
332 _TrackSpec(mbid="11111111-1111-1111-1111-111111111111"),
333 _TrackSpec("qobuz_instance", mbid="22222222-2222-2222-2222-222222222222"),
334 )
335
336 await mass.music._reconcile_duplicate_tracks()
337
338 assert await mass.music.tracks.get_library_item(track_1.item_id)
339 assert await mass.music.tracks.get_library_item(track_2.item_id)
340
341
342async def test_keeps_tracks_with_different_artists_apart(mass: MusicAssistant) -> None:
343 """A shared title and album slot is not enough when the track artists differ."""
344 track_1, track_2 = await _build_duplicate_pair(mass)
345 other_artist = await mass.music.artists.add_item_to_library(
346 Artist(
347 item_id="0",
348 provider="library",
349 name="Different Artist",
350 provider_mappings={_mapping("qobuz_instance", "qobuz-other-artist")},
351 )
352 )
353 await mass.music.database.delete(DB_TABLE_TRACK_ARTISTS, {"track_id": int(track_2.item_id)})
354 await mass.music.database.insert(
355 DB_TABLE_TRACK_ARTISTS,
356 {"track_id": int(track_2.item_id), "artist_id": int(other_artist.item_id)},
357 )
358
359 await mass.music._reconcile_duplicate_tracks()
360
361 assert await mass.music.tracks.get_library_item(track_1.item_id)
362 assert await mass.music.tracks.get_library_item(track_2.item_id)
363
364
365async def test_treats_an_unreported_disc_number_as_disc_one(mass: MusicAssistant) -> None:
366 """A provider that reports no disc number still matches a track tagged as disc 1."""
367 track_1, track_2 = await _build_duplicate_pair(mass)
368 await mass.music.database.update(
369 DB_TABLE_ALBUM_TRACKS, {"track_id": int(track_2.item_id)}, {"disc_number": 0}
370 )
371
372 await mass.music._reconcile_duplicate_tracks()
373
374 surviving = await mass.music.tracks.get_library_item(track_1.item_id)
375 assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
376 "spotify",
377 "qobuz",
378 }
379
380
381async def test_ignores_tracks_with_an_unreported_album_position(mass: MusicAssistant) -> None:
382 """Two unknown track numbers agree on nothing, so they are no evidence of a duplicate."""
383 track_1, track_2 = await _build_duplicate_pair(mass)
384 for track in (track_1, track_2):
385 await mass.music.database.update(
386 DB_TABLE_ALBUM_TRACKS, {"track_id": int(track.item_id)}, {"track_number": 0}
387 )
388
389 await mass.music._reconcile_duplicate_tracks()
390
391 assert await mass.music.tracks.get_library_item(track_1.item_id)
392 assert await mass.music.tracks.get_library_item(track_2.item_id)
393
394
395async def test_ignores_albums_whose_title_normalizes_to_nothing(mass: MusicAssistant) -> None:
396 """Symbol-only album titles are not treated as agreement, they match everything."""
397 track_1, track_2 = await _build_duplicate_pair(
398 mass, second=_TrackSpec("qobuz_instance", album_name="+")
399 )
400 for track in (track_1, track_2):
401 album_id = (
402 await mass.music.database.get_rows(
403 DB_TABLE_ALBUM_TRACKS, {"track_id": int(track.item_id)}
404 )
405 )[0]["album_id"]
406 await mass.music.database.update(
407 DB_TABLE_ALBUMS, {"item_id": album_id}, {"name": "÷", "search_name": ""}
408 )
409
410 await mass.music._reconcile_duplicate_tracks()
411
412 assert await mass.music.tracks.get_library_item(track_1.item_id)
413 assert await mass.music.tracks.get_library_item(track_2.item_id)
414
415
416async def test_ignores_tracks_from_the_same_provider(mass: MusicAssistant) -> None:
417 """Two rows of the same provider are left alone, however alike they look."""
418 track_1, track_2 = await _build_duplicate_pair(mass, second=_TrackSpec("spotify_instance"))
419
420 await mass.music._reconcile_duplicate_tracks()
421
422 assert await mass.music.tracks.get_library_item(track_1.item_id)
423 assert await mass.music.tracks.get_library_item(track_2.item_id)
424
425
426async def test_requires_agreement_on_the_album(mass: MusicAssistant) -> None:
427 """Tracks that sit on differently titled albums are not treated as duplicates."""
428 track_1, track_2 = await _build_duplicate_pair(
429 mass, second=_TrackSpec("qobuz_instance", album_name="Greatest Hits")
430 )
431
432 await mass.music._reconcile_duplicate_tracks()
433
434 assert await mass.music.tracks.get_library_item(track_1.item_id)
435 assert await mass.music.tracks.get_library_item(track_2.item_id)
436
437
438@pytest.mark.parametrize("suffix", ["EP", "Single"])
439@pytest.mark.parametrize("suffix_on_second", [False, True])
440async def test_merges_across_a_spelled_out_retail_suffix(
441 mass: MusicAssistant, suffix: str, suffix_on_second: bool
442) -> None:
443 """A provider that appends the format to an EP or single title still names the same album."""
444 specs = [_TrackSpec(), _TrackSpec("qobuz_instance")]
445 index = 1 if suffix_on_second else 0
446 specs[index] = specs[index]._replace(album_name=f"{specs[index].album_name} - {suffix}")
447 track_1, track_2 = await _build_duplicate_pair(mass, *specs)
448
449 await mass.music._reconcile_duplicate_tracks()
450
451 surviving = await mass.music.tracks.get_library_item(track_1.item_id)
452 assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
453 "spotify",
454 "qobuz",
455 }
456 with pytest.raises(MediaNotFoundError):
457 await mass.music.tracks.get_library_item(track_2.item_id)
458
459
460async def test_keeps_an_ep_apart_from_the_single_of_the_same_name(mass: MusicAssistant) -> None:
461 """Two titles that each name their format and disagree are different releases."""
462 track_1, track_2 = await _build_duplicate_pair(
463 mass,
464 _TrackSpec(album_name="Shared Album - EP"),
465 _TrackSpec("qobuz_instance", album_name="Shared Album - Single"),
466 )
467
468 await mass.music._reconcile_duplicate_tracks()
469
470 assert await mass.music.tracks.get_library_item(track_1.item_id)
471 assert await mass.music.tracks.get_library_item(track_2.item_id)
472
473
474@pytest.mark.parametrize("suffix_on_second", [False, True])
475async def test_a_suffix_without_a_dash_is_left_to_the_album_comparison(
476 mass: MusicAssistant, suffix_on_second: bool
477) -> None:
478 """The query relates titles the album comparison still refuses to call the same album."""
479 specs = [_TrackSpec(), _TrackSpec("qobuz_instance")]
480 index = 1 if suffix_on_second else 0
481 specs[index] = specs[index]._replace(album_name=f"{specs[index].album_name} EP")
482 track_1, track_2 = await _build_duplicate_pair(mass, *specs)
483
484 await mass.music._reconcile_duplicate_tracks()
485
486 assert await mass.music.tracks.get_library_item(track_1.item_id)
487 assert await mass.music.tracks.get_library_item(track_2.item_id)
488
489
490async def test_a_title_merely_ending_in_a_suffix_word_is_left_alone(
491 mass: MusicAssistant,
492) -> None:
493 """Only a spelled-out suffix relates two titles, not a word that happens to end in one."""
494 track_1, track_2 = await _build_duplicate_pair(
495 mass, _TrackSpec(album_name="Sle"), _TrackSpec("qobuz_instance", album_name="Sleep")
496 )
497
498 await mass.music._reconcile_duplicate_tracks()
499
500 assert await mass.music.tracks.get_library_item(track_1.item_id)
501 assert await mass.music.tracks.get_library_item(track_2.item_id)
502
503
504async def test_keeps_albums_that_differ_only_by_a_symbol_apart(mass: MusicAssistant) -> None:
505 """Album titles that normalize alike are still held to the album comparison."""
506 track_1, track_2 = await _build_duplicate_pair(
507 mass,
508 _TrackSpec(album_name="Shared Album"),
509 _TrackSpec("qobuz_instance", album_name="Shared Album +"),
510 )
511
512 await mass.music._reconcile_duplicate_tracks()
513
514 assert await mass.music.tracks.get_library_item(track_1.item_id)
515 assert await mass.music.tracks.get_library_item(track_2.item_id)
516
517
518async def test_requires_agreement_on_the_album_position(mass: MusicAssistant) -> None:
519 """Tracks at a different position on the same album are not treated as duplicates."""
520 track_1, track_2 = await _build_duplicate_pair(
521 mass, second=_TrackSpec("qobuz_instance", track_number=4)
522 )
523
524 await mass.music._reconcile_duplicate_tracks()
525
526 assert await mass.music.tracks.get_library_item(track_1.item_id)
527 assert await mass.music.tracks.get_library_item(track_2.item_id)
528
529
530async def test_ignores_tracks_with_a_large_duration_difference(mass: MusicAssistant) -> None:
531 """A duration difference beyond the tolerance rules out a duplicate."""
532 track_1, track_2 = await _build_duplicate_pair(
533 mass, second=_TrackSpec("qobuz_instance", duration=260)
534 )
535
536 await mass.music._reconcile_duplicate_tracks()
537
538 assert await mass.music.tracks.get_library_item(track_1.item_id)
539 assert await mass.music.tracks.get_library_item(track_2.item_id)
540
541
542async def test_keeps_the_row_with_the_most_provider_mappings(mass: MusicAssistant) -> None:
543 """The richest row survives the merge, so the fewest mappings have to move."""
544 track_1, track_2 = await _build_duplicate_pair(mass)
545 await mass.music.tracks.add_provider_mapping(
546 track_2.item_id, _mapping("tidal_instance", "tidal-track")
547 )
548
549 await mass.music._reconcile_duplicate_tracks()
550
551 surviving = await mass.music.tracks.get_library_item(track_2.item_id)
552 assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
553 "spotify",
554 "qobuz",
555 "tidal",
556 }
557 with pytest.raises(MediaNotFoundError):
558 await mass.music.tracks.get_library_item(track_1.item_id)
559
560
561# --------------------------------------------------------------------------- #
562# batch mechanics #
563# --------------------------------------------------------------------------- #
564
565
566async def test_full_batch_advances_the_cursor() -> None:
567 """A full batch resumes after the last examined row on the next run."""
568 ctrl = _bare_controller(
569 [
570 {"item_id_1": index, "item_id_2": 1000 + index}
571 for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE + 1)
572 ]
573 )
574 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)) as merge:
575 await ctrl._reconcile_duplicate_tracks()
576
577 assert ctrl._track_reconciliation_cursor == (
578 TRACK_RECONCILIATION_BATCH_SIZE,
579 1000 + TRACK_RECONCILIATION_BATCH_SIZE,
580 )
581 assert merge.await_count == TRACK_RECONCILIATION_BATCH_SIZE
582
583
584async def test_full_batch_resumes_within_the_same_track() -> None:
585 """A batch boundary between two pairs of one track resumes at that exact pair."""
586 ctrl = _bare_controller(
587 [
588 {"item_id_1": 10, "item_id_2": 20 + offset}
589 for offset in range(TRACK_RECONCILIATION_BATCH_SIZE)
590 ]
591 )
592
593 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
594 await ctrl._reconcile_duplicate_tracks()
595
596 # resuming at (10, ...) rather than past track 10 keeps its remaining pairs reachable
597 assert ctrl._track_reconciliation_cursor == (10, 19 + TRACK_RECONCILIATION_BATCH_SIZE)
598
599
600async def test_defers_while_a_library_sync_is_running() -> None:
601 """Duplicates are judged against a settled library, never a half-synced one."""
602 ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
603 candidate_query = AsyncMock(return_value=[{"item_id_1": 1, "item_id_2": 2}])
604 ctrl._database = Mock(get_rows_from_query=candidate_query)
605 running = Mock(status=TaskStatus.RUNNING)
606 ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[running])))
607
608 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock()) as merge:
609 await ctrl._reconcile_duplicate_tracks()
610
611 assert not merge.called
612 assert not candidate_query.called
613
614
615async def test_no_candidate_pair_is_starved() -> None:
616 """Repeated runs reach every candidate, including pairs a batch boundary cut off."""
617 # one track with more duplicate partners than fit in a single batch, so the boundary
618 # falls in the middle of its pairs, followed by candidates that must stay reachable
619 pairs = [(10, 20 + offset) for offset in range(TRACK_RECONCILIATION_BATCH_SIZE + 3)]
620 pairs += [(11, 40), (12, 50)]
621 seen: set[tuple[int, int]] = set()
622
623 async def _candidates(_sql: str, params: dict[str, int], limit: int) -> list[dict[str, int]]:
624 cursor = (params["cursor_item_id_1"], params["cursor_item_id_2"])
625 return [
626 {"item_id_1": item_id_1, "item_id_2": item_id_2}
627 for item_id_1, item_id_2 in [pair for pair in pairs if pair > cursor][:limit]
628 ]
629
630 async def _refuse(item_id_1: int, item_id_2: int) -> bool:
631 # refusing every pair keeps the candidate set intact, so a starved pair stays starved
632 seen.add((item_id_1, item_id_2))
633 return False
634
635 ctrl = _bare_controller([])
636 ctrl._database = Mock(get_rows_from_query=AsyncMock(side_effect=_candidates))
637
638 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_refuse)):
639 for _ in range(4):
640 await ctrl._reconcile_duplicate_tracks()
641
642 assert seen == set(pairs)
643
644
645async def test_a_short_batch_ends_the_walk() -> None:
646 """A batch that is not full means the end of the library has been reached."""
647 ctrl = _bare_controller([{"item_id_1": 7, "item_id_2": 9}])
648 ctrl._track_reconciliation_cursor = (5, 6)
649
650 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)):
651 await ctrl._reconcile_duplicate_tracks()
652
653 assert ctrl._track_reconciliation_cursor is None
654
655
656async def test_a_finished_walk_does_not_query_again() -> None:
657 """A duplicate-free library must not pay for a scan every hour to prove it."""
658 ctrl = _bare_controller([])
659 candidate_query = AsyncMock(return_value=[])
660 ctrl._database = Mock(get_rows_from_query=candidate_query)
661 ctrl._track_reconciliation_cursor = None
662
663 await ctrl._reconcile_duplicate_tracks()
664
665 assert not candidate_query.called
666
667
668async def test_a_completed_sync_does_not_restart_a_walk_in_progress() -> None:
669 """Rewinding mid-walk would re-examine the same prefix forever, starving the rest."""
670 ctrl = _bare_controller(
671 [
672 {"item_id_1": index, "item_id_2": 900 + index}
673 for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE + 1)
674 ]
675 )
676 ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])))
677
678 with patch.object(ctrl, "_queue_database_cleanup_task", Mock()):
679 ctrl._handle_sync_completion_check()
680 assert ctrl._track_reconciliation_cursor == (0, 0)
681
682 # a full batch means the walk is still in progress, so it keeps its place
683 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
684 await ctrl._reconcile_duplicate_tracks()
685
686 assert ctrl._track_reconciliation_cursor == (
687 TRACK_RECONCILIATION_BATCH_SIZE,
688 900 + TRACK_RECONCILIATION_BATCH_SIZE,
689 )
690 assert ctrl._track_reconciliation_rescan_due
691
692
693async def test_the_next_pass_starts_once_the_walk_reaches_the_end() -> None:
694 """The rescan a sync asked for happens as soon as the current walk drains."""
695 ctrl = _bare_controller([{"item_id_1": 7, "item_id_2": 9}])
696 ctrl._track_reconciliation_cursor = None
697 ctrl._track_reconciliation_rescan_due = True
698
699 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
700 await ctrl._reconcile_duplicate_tracks()
701
702 # the rescan rewound to the top, walked to the end again and finished there
703 assert ctrl._track_reconciliation_cursor is None
704 assert not ctrl._track_reconciliation_rescan_due
705
706
707def _persisting_mass(stored: dict[str, object]) -> Mock:
708 """Build a mass stand-in whose raw core config survives being read back."""
709 return Mock(
710 tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])),
711 config=Mock(
712 set_raw_core_config_value=lambda _domain, key, value: stored.__setitem__(key, value),
713 get_raw_core_config_value=lambda _domain, key, default=None: stored.get(key, default),
714 ),
715 )
716
717
718async def test_an_unexpected_failure_keeps_the_pairs_it_never_reached() -> None:
719 """A run cut short resumes at the pair it got to, not past the whole batch."""
720 ctrl = _bare_controller(
721 [{"item_id_1": index, "item_id_2": 900 + index} for index in range(1, 6)]
722 )
723
724 async def _boom(item_id_1: int, _item_id_2: int) -> bool:
725 if item_id_1 == 3:
726 raise RuntimeError("something unexpected")
727 return False
728
729 with (
730 patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_boom)),
731 pytest.raises(RuntimeError),
732 ):
733 await ctrl._reconcile_duplicate_tracks()
734
735 assert ctrl._track_reconciliation_cursor == (2, 902)
736
737
738async def test_a_merge_asks_for_another_pass() -> None:
739 """Merging moves relations onto the surviving row, which may duplicate an earlier one."""
740 ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
741
742 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)):
743 await ctrl._reconcile_duplicate_tracks()
744
745 assert ctrl._track_reconciliation_rescan_due
746
747
748async def test_a_failed_pair_asks_for_another_pass() -> None:
749 """A pair that failed on something transient is looked at again rather than dropped."""
750 ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
751
752 with (
753 patch.object(
754 ctrl,
755 "_merge_duplicate_track_pair",
756 AsyncMock(side_effect=MusicAssistantError("transient")),
757 ),
758 patch(_REPORT_FAILURE),
759 ):
760 await ctrl._reconcile_duplicate_tracks()
761
762 assert ctrl._track_reconciliation_rescan_due
763
764
765async def test_a_run_without_merges_asks_for_nothing() -> None:
766 """A walk that changed nothing has no reason to start over."""
767 ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
768
769 with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
770 await ctrl._reconcile_duplicate_tracks()
771
772 assert not ctrl._track_reconciliation_rescan_due
773
774
775async def test_the_walk_resumes_after_a_restart() -> None:
776 """Progress is kept across restarts, so a long run of refusals is only crossed once."""
777 stored: dict[str, object] = {}
778 ctrl = _bare_controller([])
779 ctrl.mass = _persisting_mass(stored)
780 ctrl._set_track_reconciliation_state((42, 99), True)
781
782 restarted = _bare_controller([])
783 restarted.mass = _persisting_mass(stored)
784 restarted._restore_track_reconciliation_state()
785
786 assert restarted._track_reconciliation_cursor == (42, 99)
787 assert restarted._track_reconciliation_rescan_due
788
789
790async def test_a_finished_walk_stays_finished_after_a_restart() -> None:
791 """A library with nothing left to do does not start scanning again on every boot."""
792 stored: dict[str, object] = {}
793 ctrl = _bare_controller([])
794 ctrl.mass = _persisting_mass(stored)
795 ctrl._set_track_reconciliation_state(None, False)
796
797 restarted = _bare_controller([])
798 restarted.mass = _persisting_mass(stored)
799 restarted._restore_track_reconciliation_state()
800
801 assert restarted._track_reconciliation_cursor is None
802
803
804async def test_a_first_run_starts_at_the_beginning() -> None:
805 """With nothing stored yet the walk starts at the top of the library."""
806 ctrl = _bare_controller([])
807 ctrl.mass = _persisting_mass({})
808
809 ctrl._restore_track_reconciliation_state()
810
811 assert ctrl._track_reconciliation_cursor == (0, 0)
812 assert not ctrl._track_reconciliation_rescan_due
813
814
815async def test_no_candidate_survives_a_sync_driven_rewind() -> None:
816 """Repeated syncs must not keep a later candidate permanently out of reach."""
817 pairs = [(index, 900 + index) for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE * 3)]
818 seen: set[tuple[int, int]] = set()
819
820 async def _candidates(_sql: str, params: dict[str, int], limit: int) -> list[dict[str, int]]:
821 cursor = (params["cursor_item_id_1"], params["cursor_item_id_2"])
822 return [
823 {"item_id_1": one, "item_id_2": two}
824 for one, two in [pair for pair in pairs if pair > cursor][:limit]
825 ]
826
827 async def _refuse(item_id_1: int, item_id_2: int) -> bool:
828 seen.add((item_id_1, item_id_2))
829 return False
830
831 ctrl = _bare_controller([])
832 ctrl._database = Mock(get_rows_from_query=AsyncMock(side_effect=_candidates))
833 ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])))
834
835 with (
836 patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_refuse)),
837 patch.object(ctrl, "_queue_database_cleanup_task", Mock()),
838 ):
839 for _ in range(4):
840 # a sync completes before every run, as a busy library would do
841 ctrl._handle_sync_completion_check()
842 await ctrl._reconcile_duplicate_tracks()
843
844 assert seen == set(pairs)
845
846
847async def test_batch_continues_quietly_after_an_already_merged_row() -> None:
848 """A row absorbed by an earlier merge in the same batch is expected, so it stays silent."""
849 ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}, {"item_id_1": 3, "item_id_2": 4}])
850 logger = Mock()
851 ctrl.logger = logger
852
853 with (
854 patch.object(
855 ctrl,
856 "_merge_duplicate_track_pair",
857 AsyncMock(side_effect=[MediaNotFoundError("gone"), True]),
858 ) as merge,
859 patch(_REPORT_FAILURE) as report_failure,
860 ):
861 await ctrl._reconcile_duplicate_tracks()
862
863 assert merge.await_count == 2
864 assert not logger.warning.called
865 assert not report_failure.called
866
867
868async def test_batch_continues_after_a_failing_pair() -> None:
869 """A failing pair is reported but does not abort the rest of the batch."""
870 ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}, {"item_id_1": 3, "item_id_2": 4}])
871 logger = Mock()
872 ctrl.logger = logger
873
874 with (
875 patch.object(
876 ctrl,
877 "_merge_duplicate_track_pair",
878 AsyncMock(side_effect=[MusicAssistantError("boom"), True]),
879 ) as merge,
880 patch(_REPORT_FAILURE) as report_failure,
881 ):
882 await ctrl._reconcile_duplicate_tracks()
883
884 assert merge.await_count == 2
885 assert logger.warning.called
886 assert report_failure.called
887