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
438async def test_requires_agreement_on_the_album_position(mass: MusicAssistant) -> None:
439    """Tracks at a different position on the same album are not treated as duplicates."""
440    track_1, track_2 = await _build_duplicate_pair(
441        mass, second=_TrackSpec("qobuz_instance", track_number=4)
442    )
443
444    await mass.music._reconcile_duplicate_tracks()
445
446    assert await mass.music.tracks.get_library_item(track_1.item_id)
447    assert await mass.music.tracks.get_library_item(track_2.item_id)
448
449
450async def test_ignores_tracks_with_a_large_duration_difference(mass: MusicAssistant) -> None:
451    """A duration difference beyond the tolerance rules out a duplicate."""
452    track_1, track_2 = await _build_duplicate_pair(
453        mass, second=_TrackSpec("qobuz_instance", duration=260)
454    )
455
456    await mass.music._reconcile_duplicate_tracks()
457
458    assert await mass.music.tracks.get_library_item(track_1.item_id)
459    assert await mass.music.tracks.get_library_item(track_2.item_id)
460
461
462async def test_keeps_the_row_with_the_most_provider_mappings(mass: MusicAssistant) -> None:
463    """The richest row survives the merge, so the fewest mappings have to move."""
464    track_1, track_2 = await _build_duplicate_pair(mass)
465    await mass.music.tracks.add_provider_mapping(
466        track_2.item_id, _mapping("tidal_instance", "tidal-track")
467    )
468
469    await mass.music._reconcile_duplicate_tracks()
470
471    surviving = await mass.music.tracks.get_library_item(track_2.item_id)
472    assert {mapping.provider_domain for mapping in surviving.provider_mappings} == {
473        "spotify",
474        "qobuz",
475        "tidal",
476    }
477    with pytest.raises(MediaNotFoundError):
478        await mass.music.tracks.get_library_item(track_1.item_id)
479
480
481# --------------------------------------------------------------------------- #
482#  batch mechanics                                                             #
483# --------------------------------------------------------------------------- #
484
485
486async def test_full_batch_advances_the_cursor() -> None:
487    """A full batch resumes after the last examined row on the next run."""
488    ctrl = _bare_controller(
489        [
490            {"item_id_1": index, "item_id_2": 1000 + index}
491            for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE + 1)
492        ]
493    )
494    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)) as merge:
495        await ctrl._reconcile_duplicate_tracks()
496
497    assert ctrl._track_reconciliation_cursor == (
498        TRACK_RECONCILIATION_BATCH_SIZE,
499        1000 + TRACK_RECONCILIATION_BATCH_SIZE,
500    )
501    assert merge.await_count == TRACK_RECONCILIATION_BATCH_SIZE
502
503
504async def test_full_batch_resumes_within_the_same_track() -> None:
505    """A batch boundary between two pairs of one track resumes at that exact pair."""
506    ctrl = _bare_controller(
507        [
508            {"item_id_1": 10, "item_id_2": 20 + offset}
509            for offset in range(TRACK_RECONCILIATION_BATCH_SIZE)
510        ]
511    )
512
513    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
514        await ctrl._reconcile_duplicate_tracks()
515
516    # resuming at (10, ...) rather than past track 10 keeps its remaining pairs reachable
517    assert ctrl._track_reconciliation_cursor == (10, 19 + TRACK_RECONCILIATION_BATCH_SIZE)
518
519
520async def test_defers_while_a_library_sync_is_running() -> None:
521    """Duplicates are judged against a settled library, never a half-synced one."""
522    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
523    candidate_query = AsyncMock(return_value=[{"item_id_1": 1, "item_id_2": 2}])
524    ctrl._database = Mock(get_rows_from_query=candidate_query)
525    running = Mock(status=TaskStatus.RUNNING)
526    ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[running])))
527
528    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock()) as merge:
529        await ctrl._reconcile_duplicate_tracks()
530
531    assert not merge.called
532    assert not candidate_query.called
533
534
535async def test_no_candidate_pair_is_starved() -> None:
536    """Repeated runs reach every candidate, including pairs a batch boundary cut off."""
537    # one track with more duplicate partners than fit in a single batch, so the boundary
538    # falls in the middle of its pairs, followed by candidates that must stay reachable
539    pairs = [(10, 20 + offset) for offset in range(TRACK_RECONCILIATION_BATCH_SIZE + 3)]
540    pairs += [(11, 40), (12, 50)]
541    seen: set[tuple[int, int]] = set()
542
543    async def _candidates(_sql: str, params: dict[str, int], limit: int) -> list[dict[str, int]]:
544        cursor = (params["cursor_item_id_1"], params["cursor_item_id_2"])
545        return [
546            {"item_id_1": item_id_1, "item_id_2": item_id_2}
547            for item_id_1, item_id_2 in [pair for pair in pairs if pair > cursor][:limit]
548        ]
549
550    async def _refuse(item_id_1: int, item_id_2: int) -> bool:
551        # refusing every pair keeps the candidate set intact, so a starved pair stays starved
552        seen.add((item_id_1, item_id_2))
553        return False
554
555    ctrl = _bare_controller([])
556    ctrl._database = Mock(get_rows_from_query=AsyncMock(side_effect=_candidates))
557
558    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_refuse)):
559        for _ in range(4):
560            await ctrl._reconcile_duplicate_tracks()
561
562    assert seen == set(pairs)
563
564
565async def test_a_short_batch_ends_the_walk() -> None:
566    """A batch that is not full means the end of the library has been reached."""
567    ctrl = _bare_controller([{"item_id_1": 7, "item_id_2": 9}])
568    ctrl._track_reconciliation_cursor = (5, 6)
569
570    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)):
571        await ctrl._reconcile_duplicate_tracks()
572
573    assert ctrl._track_reconciliation_cursor is None
574
575
576async def test_a_finished_walk_does_not_query_again() -> None:
577    """A duplicate-free library must not pay for a scan every hour to prove it."""
578    ctrl = _bare_controller([])
579    candidate_query = AsyncMock(return_value=[])
580    ctrl._database = Mock(get_rows_from_query=candidate_query)
581    ctrl._track_reconciliation_cursor = None
582
583    await ctrl._reconcile_duplicate_tracks()
584
585    assert not candidate_query.called
586
587
588async def test_a_completed_sync_does_not_restart_a_walk_in_progress() -> None:
589    """Rewinding mid-walk would re-examine the same prefix forever, starving the rest."""
590    ctrl = _bare_controller(
591        [
592            {"item_id_1": index, "item_id_2": 900 + index}
593            for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE + 1)
594        ]
595    )
596    ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])))
597
598    with patch.object(ctrl, "_queue_database_cleanup_task", Mock()):
599        ctrl._handle_sync_completion_check()
600    assert ctrl._track_reconciliation_cursor == (0, 0)
601
602    # a full batch means the walk is still in progress, so it keeps its place
603    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
604        await ctrl._reconcile_duplicate_tracks()
605
606    assert ctrl._track_reconciliation_cursor == (
607        TRACK_RECONCILIATION_BATCH_SIZE,
608        900 + TRACK_RECONCILIATION_BATCH_SIZE,
609    )
610    assert ctrl._track_reconciliation_rescan_due
611
612
613async def test_the_next_pass_starts_once_the_walk_reaches_the_end() -> None:
614    """The rescan a sync asked for happens as soon as the current walk drains."""
615    ctrl = _bare_controller([{"item_id_1": 7, "item_id_2": 9}])
616    ctrl._track_reconciliation_cursor = None
617    ctrl._track_reconciliation_rescan_due = True
618
619    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
620        await ctrl._reconcile_duplicate_tracks()
621
622    # the rescan rewound to the top, walked to the end again and finished there
623    assert ctrl._track_reconciliation_cursor is None
624    assert not ctrl._track_reconciliation_rescan_due
625
626
627def _persisting_mass(stored: dict[str, object]) -> Mock:
628    """Build a mass stand-in whose raw core config survives being read back."""
629    return Mock(
630        tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])),
631        config=Mock(
632            set_raw_core_config_value=lambda _domain, key, value: stored.__setitem__(key, value),
633            get_raw_core_config_value=lambda _domain, key, default=None: stored.get(key, default),
634        ),
635    )
636
637
638async def test_an_unexpected_failure_keeps_the_pairs_it_never_reached() -> None:
639    """A run cut short resumes at the pair it got to, not past the whole batch."""
640    ctrl = _bare_controller(
641        [{"item_id_1": index, "item_id_2": 900 + index} for index in range(1, 6)]
642    )
643
644    async def _boom(item_id_1: int, _item_id_2: int) -> bool:
645        if item_id_1 == 3:
646            raise RuntimeError("something unexpected")
647        return False
648
649    with (
650        patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_boom)),
651        pytest.raises(RuntimeError),
652    ):
653        await ctrl._reconcile_duplicate_tracks()
654
655    assert ctrl._track_reconciliation_cursor == (2, 902)
656
657
658async def test_a_merge_asks_for_another_pass() -> None:
659    """Merging moves relations onto the surviving row, which may duplicate an earlier one."""
660    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
661
662    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=True)):
663        await ctrl._reconcile_duplicate_tracks()
664
665    assert ctrl._track_reconciliation_rescan_due
666
667
668async def test_a_failed_pair_asks_for_another_pass() -> None:
669    """A pair that failed on something transient is looked at again rather than dropped."""
670    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
671
672    with (
673        patch.object(
674            ctrl,
675            "_merge_duplicate_track_pair",
676            AsyncMock(side_effect=MusicAssistantError("transient")),
677        ),
678        patch(_REPORT_FAILURE),
679    ):
680        await ctrl._reconcile_duplicate_tracks()
681
682    assert ctrl._track_reconciliation_rescan_due
683
684
685async def test_a_run_without_merges_asks_for_nothing() -> None:
686    """A walk that changed nothing has no reason to start over."""
687    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}])
688
689    with patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(return_value=False)):
690        await ctrl._reconcile_duplicate_tracks()
691
692    assert not ctrl._track_reconciliation_rescan_due
693
694
695async def test_the_walk_resumes_after_a_restart() -> None:
696    """Progress is kept across restarts, so a long run of refusals is only crossed once."""
697    stored: dict[str, object] = {}
698    ctrl = _bare_controller([])
699    ctrl.mass = _persisting_mass(stored)
700    ctrl._set_track_reconciliation_state((42, 99), True)
701
702    restarted = _bare_controller([])
703    restarted.mass = _persisting_mass(stored)
704    restarted._restore_track_reconciliation_state()
705
706    assert restarted._track_reconciliation_cursor == (42, 99)
707    assert restarted._track_reconciliation_rescan_due
708
709
710async def test_a_finished_walk_stays_finished_after_a_restart() -> None:
711    """A library with nothing left to do does not start scanning again on every boot."""
712    stored: dict[str, object] = {}
713    ctrl = _bare_controller([])
714    ctrl.mass = _persisting_mass(stored)
715    ctrl._set_track_reconciliation_state(None, False)
716
717    restarted = _bare_controller([])
718    restarted.mass = _persisting_mass(stored)
719    restarted._restore_track_reconciliation_state()
720
721    assert restarted._track_reconciliation_cursor is None
722
723
724async def test_a_first_run_starts_at_the_beginning() -> None:
725    """With nothing stored yet the walk starts at the top of the library."""
726    ctrl = _bare_controller([])
727    ctrl.mass = _persisting_mass({})
728
729    ctrl._restore_track_reconciliation_state()
730
731    assert ctrl._track_reconciliation_cursor == (0, 0)
732    assert not ctrl._track_reconciliation_rescan_due
733
734
735async def test_no_candidate_survives_a_sync_driven_rewind() -> None:
736    """Repeated syncs must not keep a later candidate permanently out of reach."""
737    pairs = [(index, 900 + index) for index in range(1, TRACK_RECONCILIATION_BATCH_SIZE * 3)]
738    seen: set[tuple[int, int]] = set()
739
740    async def _candidates(_sql: str, params: dict[str, int], limit: int) -> list[dict[str, int]]:
741        cursor = (params["cursor_item_id_1"], params["cursor_item_id_2"])
742        return [
743            {"item_id_1": one, "item_id_2": two}
744            for one, two in [pair for pair in pairs if pair > cursor][:limit]
745        ]
746
747    async def _refuse(item_id_1: int, item_id_2: int) -> bool:
748        seen.add((item_id_1, item_id_2))
749        return False
750
751    ctrl = _bare_controller([])
752    ctrl._database = Mock(get_rows_from_query=AsyncMock(side_effect=_candidates))
753    ctrl.mass = Mock(tasks=Mock(get_tasks_by_metadata=Mock(return_value=[])))
754
755    with (
756        patch.object(ctrl, "_merge_duplicate_track_pair", AsyncMock(side_effect=_refuse)),
757        patch.object(ctrl, "_queue_database_cleanup_task", Mock()),
758    ):
759        for _ in range(4):
760            # a sync completes before every run, as a busy library would do
761            ctrl._handle_sync_completion_check()
762            await ctrl._reconcile_duplicate_tracks()
763
764    assert seen == set(pairs)
765
766
767async def test_batch_continues_quietly_after_an_already_merged_row() -> None:
768    """A row absorbed by an earlier merge in the same batch is expected, so it stays silent."""
769    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}, {"item_id_1": 3, "item_id_2": 4}])
770    logger = Mock()
771    ctrl.logger = logger
772
773    with (
774        patch.object(
775            ctrl,
776            "_merge_duplicate_track_pair",
777            AsyncMock(side_effect=[MediaNotFoundError("gone"), True]),
778        ) as merge,
779        patch(_REPORT_FAILURE) as report_failure,
780    ):
781        await ctrl._reconcile_duplicate_tracks()
782
783    assert merge.await_count == 2
784    assert not logger.warning.called
785    assert not report_failure.called
786
787
788async def test_batch_continues_after_a_failing_pair() -> None:
789    """A failing pair is reported but does not abort the rest of the batch."""
790    ctrl = _bare_controller([{"item_id_1": 1, "item_id_2": 2}, {"item_id_1": 3, "item_id_2": 4}])
791    logger = Mock()
792    ctrl.logger = logger
793
794    with (
795        patch.object(
796            ctrl,
797            "_merge_duplicate_track_pair",
798            AsyncMock(side_effect=[MusicAssistantError("boom"), True]),
799        ) as merge,
800        patch(_REPORT_FAILURE) as report_failure,
801    ):
802        await ctrl._reconcile_duplicate_tracks()
803
804    assert merge.await_count == 2
805    assert logger.warning.called
806    assert report_failure.called
807