music-assistant-server

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
460@pytest.mark.parametrize("spelling", ["-EP", "(EP)", "[EP]"])
461async def test_merges_when_providers_set_off_the_suffix_differently(
462    mass: MusicAssistant, spelling: str
463) -> None:
464    """Providers disagreeing on how the retail suffix is written still share an album."""
465    track_1, track_2 = await _build_duplicate_pair(
466        mass,
467        _TrackSpec(album_name="Shared Album - EP"),
468        _TrackSpec("qobuz_instance", album_name=f"Shared Album {spelling}"),
469    )
470
471    await mass.music._reconcile_duplicate_tracks()
472
473    surviving = await mass.music.tracks.get_library_item(track_1.item_id)
474    assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
475        "spotify",
476        "qobuz",
477    }
478    with pytest.raises(MediaNotFoundError):
479        await mass.music.tracks.get_library_item(track_2.item_id)
480
481
482async def test_keeps_an_ep_apart_from_the_single_of_the_same_name(mass: MusicAssistant) -> None:
483    """Two titles that each name their format and disagree are different releases."""
484    track_1, track_2 = await _build_duplicate_pair(
485        mass,
486        _TrackSpec(album_name="Shared Album - EP"),
487        _TrackSpec("qobuz_instance", album_name="Shared Album - Single"),
488    )
489
490    await mass.music._reconcile_duplicate_tracks()
491
492    assert await mass.music.tracks.get_library_item(track_1.item_id)
493    assert await mass.music.tracks.get_library_item(track_2.item_id)
494
495
496@pytest.mark.parametrize("spelling", ["-EP", "(EP)", "[EP]"])
497@pytest.mark.parametrize("suffix_on_second", [False, True])
498async def test_merges_a_plain_title_with_any_spelled_out_suffix(
499    mass: MusicAssistant, spelling: str, suffix_on_second: bool
500) -> None:
501    """A plain title and one naming the format are the same album, however it is written."""
502    specs = [_TrackSpec(), _TrackSpec("qobuz_instance")]
503    index = 1 if suffix_on_second else 0
504    specs[index] = specs[index]._replace(album_name=f"{specs[index].album_name} {spelling}")
505    track_1, track_2 = await _build_duplicate_pair(mass, *specs)
506
507    await mass.music._reconcile_duplicate_tracks()
508
509    surviving = await mass.music.tracks.get_library_item(track_1.item_id)
510    assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
511        "spotify",
512        "qobuz",
513    }
514    with pytest.raises(MediaNotFoundError):
515        await mass.music.tracks.get_library_item(track_2.item_id)
516
517
518@pytest.mark.parametrize("suffix_on_second", [False, True])
519async def test_a_suffix_without_a_dash_is_left_to_the_album_comparison(
520    mass: MusicAssistant, suffix_on_second: bool
521) -> None:
522    """The query relates titles the album comparison still refuses to call the same album."""
523    specs = [_TrackSpec(), _TrackSpec("qobuz_instance")]
524    index = 1 if suffix_on_second else 0
525    specs[index] = specs[index]._replace(album_name=f"{specs[index].album_name} EP")
526    track_1, track_2 = await _build_duplicate_pair(mass, *specs)
527
528    await mass.music._reconcile_duplicate_tracks()
529
530    assert await mass.music.tracks.get_library_item(track_1.item_id)
531    assert await mass.music.tracks.get_library_item(track_2.item_id)
532
533
534async def test_a_title_merely_ending_in_a_suffix_word_is_left_alone(
535    mass: MusicAssistant,
536) -> None:
537    """Only a spelled-out suffix relates two titles, not a word that happens to end in one."""
538    track_1, track_2 = await _build_duplicate_pair(
539        mass, _TrackSpec(album_name="Sle"), _TrackSpec("qobuz_instance", album_name="Sleep")
540    )
541
542    await mass.music._reconcile_duplicate_tracks()
543
544    assert await mass.music.tracks.get_library_item(track_1.item_id)
545    assert await mass.music.tracks.get_library_item(track_2.item_id)
546
547
548async def test_keeps_albums_that_differ_only_by_a_symbol_apart(mass: MusicAssistant) -> None:
549    """Album titles that normalize alike are still held to the album comparison."""
550    track_1, track_2 = await _build_duplicate_pair(
551        mass,
552        _TrackSpec(album_name="Shared Album"),
553        _TrackSpec("qobuz_instance", album_name="Shared Album +"),
554    )
555
556    await mass.music._reconcile_duplicate_tracks()
557
558    assert await mass.music.tracks.get_library_item(track_1.item_id)
559    assert await mass.music.tracks.get_library_item(track_2.item_id)
560
561
562async def test_requires_agreement_on_the_album_position(mass: MusicAssistant) -> None:
563    """Tracks at a different position on the same album are not treated as duplicates."""
564    track_1, track_2 = await _build_duplicate_pair(
565        mass, second=_TrackSpec("qobuz_instance", track_number=4)
566    )
567
568    await mass.music._reconcile_duplicate_tracks()
569
570    assert await mass.music.tracks.get_library_item(track_1.item_id)
571    assert await mass.music.tracks.get_library_item(track_2.item_id)
572
573
574async def test_ignores_tracks_with_a_large_duration_difference(mass: MusicAssistant) -> None:
575    """A duration difference beyond the tolerance rules out a duplicate."""
576    track_1, track_2 = await _build_duplicate_pair(
577        mass, second=_TrackSpec("qobuz_instance", duration=260)
578    )
579
580    await mass.music._reconcile_duplicate_tracks()
581
582    assert await mass.music.tracks.get_library_item(track_1.item_id)
583    assert await mass.music.tracks.get_library_item(track_2.item_id)
584
585
586async def test_keeps_the_row_with_the_most_provider_mappings(mass: MusicAssistant) -> None:
587    """The richest row survives the merge, so the fewest mappings have to move."""
588    track_1, track_2 = await _build_duplicate_pair(mass)
589    await mass.music.tracks.add_provider_mapping(
590        track_2.item_id, _mapping("tidal_instance", "tidal-track")
591    )
592
593    await mass.music._reconcile_duplicate_tracks()
594
595    surviving = await mass.music.tracks.get_library_item(track_2.item_id)
596    assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
597        "spotify",
598        "qobuz",
599        "tidal",
600    }
601    with pytest.raises(MediaNotFoundError):
602        await mass.music.tracks.get_library_item(track_1.item_id)
603
604
605# --------------------------------------------------------------------------- #
606#  batch mechanics                                                             #
607# --------------------------------------------------------------------------- #
608
609
610async def test_full_batch_advances_the_cursor() -> None:
611    """A full batch resumes after the last examined row on the next run."""
612    ctrl = _bare_controller(
613        [
614            {"item_id_1": index, "item_id_2": 1000 + index}
615            for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE + 1)
616        ]
617    )
618    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)) as merge:
619        await ctrl._reconcile_duplicate_tracks()
620
621    assert ctrl._track_reconciliation_cursor == (
622        TRACK_RECONCILIATION_BATCH_SIZE,
623        1000 + TRACK_RECONCILIATION_BATCH_SIZE,
624    )
625    assert merge.await_count == TRACK_RECONCILIATION_BATCH_SIZE
626
627
628async def test_full_batch_resumes_within_the_same_track() -> None:
629    """A batch boundary between two pairs of one track resumes at that exact pair."""
630    ctrl = _bare_controller(
631        [
632            {"item_id_1": 10, "item_id_2": 20 + offset}
633            for offset in range(TRACK_RECONCILIATION_BATCH_SIZE)
634        ]
635    )
636
637    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
638        await ctrl._reconcile_duplicate_tracks()
639
640    # resuming at (10, ...) rather than past track 10 keeps its remaining pairs reachable
641    assert ctrl._track_reconciliation_cursor == (10, 19 + TRACK_RECONCILIATION_BATCH_SIZE)
642
643
644async def test_defers_while_a_library_sync_is_running() -> None:
645    """Duplicates are judged against a settled library, never a half-synced one."""
646    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
647    candidate_query = AsyncMock(return_value=[{"item_id_1": 1, "item_id_2": 2}])
648    ctrl._database = Mock(get_rows_from_query=candidate_query)
649    running = Mock(status=TaskStatus.RUNNING)
650    ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[running])))
651
652    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock()) as merge:
653        await ctrl._reconcile_duplicate_tracks()
654
655    assert not merge.called
656    assert not candidate_query.called
657
658
659async def test_no_candidate_pair_is_starved() -> None:
660    """Repeated runs reach every candidate, including pairs a batch boundary cut off."""
661    # one track with more duplicate partners than fit in a single batch, so the boundary
662    # falls in the middle of its pairs, followed by candidates that must stay reachable
663    pairs = [(10, 20 + offset) for offset in range(TRACK_RECONCILIATION_BATCH_SIZE + 3)]
664    pairs += [(11, 40), (12, 50)]
665    seen: set[tuple[int, int]] = set()
666
667    async def _candidates(_sql: str, params: dict[str, int], limit: int) -> list[dict[str, int]]:
668        cursor = (params["cursor_item_id_1"], params["cursor_item_id_2"])
669        return [
670            {"item_id_1": item_id_1, "item_id_2": item_id_2}
671            for item_id_1, item_id_2 in [pair for pair in pairs if pair > cursor][:limit]
672        ]
673
674    async def _refuse(item_id_1: int, item_id_2: int) -> bool:
675        # refusing every pair keeps the candidate set intact, so a starved pair stays starved
676        seen.add((item_id_1, item_id_2))
677        return False
678
679    ctrl = _bare_controller([])
680    ctrl._database = Mock(get_rows_from_query=AsyncMock(side_effect=_candidates))
681
682    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_refuse)):
683        for _ in range(4):
684            await ctrl._reconcile_duplicate_tracks()
685
686    assert seen == set(pairs)
687
688
689async def test_a_short_batch_ends_the_walk() -> None:
690    """A batch that is not full means the end of the library has been reached."""
691    ctrl = _bare_controller([{"item_id_1": 7, "item_id_2": 9}])
692    ctrl._track_reconciliation_cursor = (5, 6)
693
694    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)):
695        await ctrl._reconcile_duplicate_tracks()
696
697    assert ctrl._track_reconciliation_cursor is None
698
699
700async def test_a_finished_walk_does_not_query_again() -> None:
701    """A duplicate-free library must not pay for a scan every hour to prove it."""
702    ctrl = _bare_controller([])
703    candidate_query = AsyncMock(return_value=[])
704    ctrl._database = Mock(get_rows_from_query=candidate_query)
705    ctrl._track_reconciliation_cursor = None
706
707    await ctrl._reconcile_duplicate_tracks()
708
709    assert not candidate_query.called
710
711
712async def test_a_completed_sync_does_not_restart_a_walk_in_progress() -> None:
713    """Rewinding mid-walk would re-examine the same prefix forever, starving the rest."""
714    ctrl = _bare_controller(
715        [
716            {"item_id_1": index, "item_id_2": 900 + index}
717            for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE + 1)
718        ]
719    )
720    ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])))
721
722    with patch.object(ctrl, "_queue_database_cleanup_task", Mock()):
723        ctrl._handle_sync_completion_check()
724    assert ctrl._track_reconciliation_cursor == (0, 0)
725
726    # a full batch means the walk is still in progress, so it keeps its place
727    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
728        await ctrl._reconcile_duplicate_tracks()
729
730    assert ctrl._track_reconciliation_cursor == (
731        TRACK_RECONCILIATION_BATCH_SIZE,
732        900 + TRACK_RECONCILIATION_BATCH_SIZE,
733    )
734    assert ctrl._track_reconciliation_rescan_due
735
736
737async def test_the_next_pass_starts_once_the_walk_reaches_the_end() -> None:
738    """The rescan a sync asked for happens as soon as the current walk drains."""
739    ctrl = _bare_controller([{"item_id_1": 7, "item_id_2": 9}])
740    ctrl._track_reconciliation_cursor = None
741    ctrl._track_reconciliation_rescan_due = True
742
743    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
744        await ctrl._reconcile_duplicate_tracks()
745
746    # the rescan rewound to the top, walked to the end again and finished there
747    assert ctrl._track_reconciliation_cursor is None
748    assert not ctrl._track_reconciliation_rescan_due
749
750
751def _persisting_mass(stored: dict[str, object]) -> Mock:
752    """Build a mass stand-in whose raw core config survives being read back."""
753    return Mock(
754        tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])),
755        config=Mock(
756            set_raw_core_config_value=lambda _domain, key, value: stored.__setitem__(key, value),
757            get_raw_core_config_value=lambda _domain, key, default=None: stored.get(key, default),
758        ),
759    )
760
761
762async def test_an_unexpected_failure_keeps_the_pairs_it_never_reached() -> None:
763    """A run cut short resumes at the pair it got to, not past the whole batch."""
764    ctrl = _bare_controller(
765        [{"item_id_1": index, "item_id_2": 900 + index} for index in range(1, 6)]
766    )
767
768    async def _boom(item_id_1: int, _item_id_2: int) -> bool:
769        if item_id_1 == 3:
770            raise RuntimeError("something unexpected")
771        return False
772
773    with (
774        patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_boom)),
775        pytest.raises(RuntimeError),
776    ):
777        await ctrl._reconcile_duplicate_tracks()
778
779    assert ctrl._track_reconciliation_cursor == (2, 902)
780
781
782async def test_a_merge_asks_for_another_pass() -> None:
783    """Merging moves relations onto the surviving row, which may duplicate an earlier one."""
784    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
785
786    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)):
787        await ctrl._reconcile_duplicate_tracks()
788
789    assert ctrl._track_reconciliation_rescan_due
790
791
792async def test_a_failed_pair_asks_for_another_pass() -> None:
793    """A pair that failed on something transient is looked at again rather than dropped."""
794    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
795
796    with (
797        patch.object(
798            ctrl,
799            "_merge_duplicate_track_pair",
800            AsyncMock(side_effect=MusicAssistantError("transient")),
801        ),
802        patch(_REPORT_FAILURE),
803    ):
804        await ctrl._reconcile_duplicate_tracks()
805
806    assert ctrl._track_reconciliation_rescan_due
807
808
809async def test_a_run_without_merges_asks_for_nothing() -> None:
810    """A walk that changed nothing has no reason to start over."""
811    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
812
813    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
814        await ctrl._reconcile_duplicate_tracks()
815
816    assert not ctrl._track_reconciliation_rescan_due
817
818
819async def test_the_walk_resumes_after_a_restart() -> None:
820    """Progress is kept across restarts, so a long run of refusals is only crossed once."""
821    stored: dict[str, object] = {}
822    ctrl = _bare_controller([])
823    ctrl.mass = _persisting_mass(stored)
824    ctrl._set_track_reconciliation_state((42, 99), True)
825
826    restarted = _bare_controller([])
827    restarted.mass = _persisting_mass(stored)
828    restarted._restore_track_reconciliation_state()
829
830    assert restarted._track_reconciliation_cursor == (42, 99)
831    assert restarted._track_reconciliation_rescan_due
832
833
834async def test_a_finished_walk_stays_finished_after_a_restart() -> None:
835    """A library with nothing left to do does not start scanning again on every boot."""
836    stored: dict[str, object] = {}
837    ctrl = _bare_controller([])
838    ctrl.mass = _persisting_mass(stored)
839    ctrl._set_track_reconciliation_state(None, False)
840
841    restarted = _bare_controller([])
842    restarted.mass = _persisting_mass(stored)
843    restarted._restore_track_reconciliation_state()
844
845    assert restarted._track_reconciliation_cursor is None
846
847
848async def test_a_first_run_starts_at_the_beginning() -> None:
849    """With nothing stored yet the walk starts at the top of the library."""
850    ctrl = _bare_controller([])
851    ctrl.mass = _persisting_mass({})
852
853    ctrl._restore_track_reconciliation_state()
854
855    assert ctrl._track_reconciliation_cursor == (0, 0)
856    assert not ctrl._track_reconciliation_rescan_due
857
858
859async def test_no_candidate_survives_a_sync_driven_rewind() -> None:
860    """Repeated syncs must not keep a later candidate permanently out of reach."""
861    pairs = [(index, 900 + index) for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE * 3)]
862    seen: set[tuple[int, int]] = set()
863
864    async def _candidates(_sql: str, params: dict[str, int], limit: int) -> list[dict[str, int]]:
865        cursor = (params["cursor_item_id_1"], params["cursor_item_id_2"])
866        return [
867            {"item_id_1": one, "item_id_2": two}
868            for one, two in [pair for pair in pairs if pair > cursor][:limit]
869        ]
870
871    async def _refuse(item_id_1: int, item_id_2: int) -> bool:
872        seen.add((item_id_1, item_id_2))
873        return False
874
875    ctrl = _bare_controller([])
876    ctrl._database = Mock(get_rows_from_query=AsyncMock(side_effect=_candidates))
877    ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])))
878
879    with (
880        patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_refuse)),
881        patch.object(ctrl, "_queue_database_cleanup_task", Mock()),
882    ):
883        for _ in range(4):
884            # a sync completes before every run, as a busy library would do
885            ctrl._handle_sync_completion_check()
886            await ctrl._reconcile_duplicate_tracks()
887
888    assert seen == set(pairs)
889
890
891async def test_batch_continues_quietly_after_an_already_merged_row() -> None:
892    """A row absorbed by an earlier merge in the same batch is expected, so it stays silent."""
893    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}, {"item_id_1": 3, "item_id_2": 4}])
894    logger = Mock()
895    ctrl.logger = logger
896
897    with (
898        patch.object(
899            ctrl,
900            "_merge_duplicate_track_pair",
901            AsyncMock(side_effect=[MediaNotFoundError("gone"), True]),
902        ) as merge,
903        patch(_REPORT_FAILURE) as report_failure,
904    ):
905        await ctrl._reconcile_duplicate_tracks()
906
907    assert merge.await_count == 2
908    assert not logger.warning.called
909    assert not report_failure.called
910
911
912async def test_batch_continues_after_a_failing_pair() -> None:
913    """A failing pair is reported but does not abort the rest of the batch."""
914    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}, {"item_id_1": 3, "item_id_2": 4}])
915    logger = Mock()
916    ctrl.logger = logger
917
918    with (
919        patch.object(
920            ctrl,
921            "_merge_duplicate_track_pair",
922            AsyncMock(side_effect=[MusicAssistantError("boom"), True]),
923        ) as merge,
924        patch(_REPORT_FAILURE) as report_failure,
925    ):
926        await ctrl._reconcile_duplicate_tracks()
927
928    assert merge.await_count == 2
929    assert logger.warning.called
930    assert report_failure.called
931