import copy import unittest from backend.app.matcher import MatchProviderError, Matcher DEFAULT_CONFIG = { 'advancedStrategy': { 'metadataFallback': True, 'downloadAssets': False }, 'metadata': { 'acoustidUrl': 'https://api.acoustid.org/v2', 'acoustidClientKey': 'client-key', 'musicbrainz': 'https://musicbrainz.org/ws/2/', 'netease': 'http://localhost:3000', 'qq': 'http://localhost:3300', 'spotifyUrl': 'https://api.spotify.com/v1', 'spotifyClientId': 'spotify-id', 'spotifySecret': 'spotify-secret', 'discogsUrl': 'https://api.discogs.com', 'discogsToken': '', 'lastfmUrl': 'https://ws.audioscrobbler.com/2.0/', 'lastfmKey': '', 'geniusUrl': 'https://api.genius.com', 'geniusToken': '' } } class MatcherTests(unittest.TestCase): def test_matches_authoritative_acoustid_candidate(self): item = build_item() candidate = build_candidate( provider='acoustid', is_authoritative=True, fingerprint_confidence=0.98 ) matcher = build_matcher( acoustid_candidates=[candidate] ) result = matcher.match_item(item, [item], DEFAULT_CONFIG) self.assertEqual(result['status'], 'matched') self.assertEqual(result['source'], 'acoustid') self.assertTrue(result['is_authoritative']) self.assertGreaterEqual(result['confidence'], 85) def test_matches_musicbrainz_text_candidate_without_fingerprint(self): item = build_item() candidate = build_candidate( provider='musicbrainz', is_authoritative=True, search_confidence=0.92 ) matcher = build_matcher( musicbrainz_candidates=[candidate] ) result = matcher.match_item(item, [item], DEFAULT_CONFIG) self.assertEqual(result['status'], 'matched') self.assertEqual(result['source'], 'musicbrainz') self.assertTrue(result['is_authoritative']) self.assertEqual(result['matched_metadata_json']['release_id'], 'release-main') def test_matches_fallback_candidate_when_authoritative_missing(self): item = build_item() candidate = build_candidate( provider='spotify', is_authoritative=False, search_confidence=0.9, source_ids={ 'spotify_track_id': 'track-1', 'spotify_album_id': 'album-1' } ) matcher = build_matcher( spotify_candidates=[candidate] ) result = matcher.match_item(item, [item], DEFAULT_CONFIG) self.assertEqual(result['status'], 'matched_fallback') self.assertEqual(result['source'], 'spotify') self.assertFalse(result['is_authoritative']) def test_respects_repair_provider_scope(self): item = build_item() authoritative = build_candidate( provider='musicbrainz', is_authoritative=True, search_confidence=0.95 ) fallback = build_candidate( provider='spotify', is_authoritative=False, search_confidence=0.9, source_ids={ 'spotify_track_id': 'track-1', 'spotify_album_id': 'album-1' } ) matcher = build_matcher( musicbrainz_candidates=[authoritative], spotify_candidates=[fallback] ) scoped_config = copy.deepcopy(DEFAULT_CONFIG) scoped_config['repair_provider_scope'] = ['spotify'] result = matcher.match_item(item, [item], scoped_config) self.assertEqual(result['status'], 'matched_fallback') self.assertEqual(result['source'], 'spotify') self.assertFalse(result['is_authoritative']) def test_returns_low_score_when_gap_is_too_small(self): item = build_item() first = build_candidate( provider='musicbrainz', is_authoritative=True, search_confidence=0.95, release_id='release-a' ) second = build_candidate( provider='musicbrainz', is_authoritative=True, search_confidence=0.94, release_id='release-b', source_ids={ 'musicbrainz_recording_id': 'recording-b', 'musicbrainz_release_id': 'release-b', 'musicbrainz_release_group_id': 'release-group-b' } ) matcher = build_matcher( musicbrainz_candidates=[first, second] ) result = matcher.match_item(item, [item], DEFAULT_CONFIG) self.assertEqual(result['status'], 'low_score') self.assertEqual(result['reason'], 'score_gap_too_small') self.assertEqual(len(result['match_candidates_json']), 2) def test_returns_not_found_when_no_candidates_exist(self): item = build_item() matcher = build_matcher() result = matcher.match_item(item, [item], DEFAULT_CONFIG) self.assertEqual(result['status'], 'not_found') self.assertIsNone(result['matched_metadata_json']) self.assertEqual(result['match_candidates_json'], []) def test_skips_acoustid_error_and_matches_musicbrainz_text_candidate(self): item = build_item() candidate = build_candidate( provider='musicbrainz', is_authoritative=True, search_confidence=0.92 ) matcher = build_matcher( acoustid_error=MatchProviderError('acoustid', 'acoustid failed'), musicbrainz_candidates=[candidate] ) result = matcher.match_item(item, [item], DEFAULT_CONFIG) self.assertEqual(result['status'], 'matched') self.assertEqual(result['source'], 'musicbrainz') self.assertEqual(len(result['provider_warnings']), 1) self.assertEqual(result['provider_warnings'][0]['provider'], 'acoustid') def test_skips_musicbrainz_error_and_uses_fallback_candidate(self): item = build_item() candidate = build_candidate( provider='spotify', is_authoritative=False, search_confidence=0.9, source_ids={ 'spotify_track_id': 'track-1', 'spotify_album_id': 'album-1' } ) matcher = build_matcher( musicbrainz_error=MatchProviderError('musicbrainz', 'musicbrainz failed'), spotify_candidates=[candidate] ) result = matcher.match_item(item, [item], DEFAULT_CONFIG) self.assertEqual(result['status'], 'matched_fallback') self.assertEqual(result['source'], 'spotify') self.assertEqual( [warning['provider'] for warning in result['provider_warnings']], ['musicbrainz', 'musicbrainz'] ) def test_returns_not_found_when_all_providers_fail(self): item = build_item() matcher = build_matcher( acoustid_error=MatchProviderError('acoustid', 'acoustid failed'), musicbrainz_error=MatchProviderError('musicbrainz', 'musicbrainz failed'), netease_error=MatchProviderError('netease', 'netease failed'), qq_error=MatchProviderError('qq', 'qq failed'), spotify_error=MatchProviderError('spotify', 'spotify failed') ) result = matcher.match_item(item, [item], DEFAULT_CONFIG) self.assertEqual(result['status'], 'not_found') self.assertEqual( [warning['provider'] for warning in result['provider_warnings']], ['acoustid', 'musicbrainz', 'netease', 'qq', 'spotify'] ) def test_album_context_converges_to_single_release(self): item_one = build_item(title='Song Title', track_number=1, duration_seconds=201) item_two = build_item( title='Song Title', track_number=2, duration_seconds=233, relative_path='Artist/Album/02.flac', filename='02.flac' ) group = [item_one, item_two] def dynamic_musicbrainz_candidates(item_metadata, _config, **_kwargs): track_number = item_metadata.get('track_number') duration_seconds = item_metadata.get('duration_seconds') release_a = build_candidate( provider='musicbrainz', is_authoritative=True, search_confidence=0.91, track_number=track_number, duration_seconds=duration_seconds, recording_id=f'recording-a-{track_number}', release_id='release-a', release_group_id='group-a', release_tracklist=[ { 'title': 'Song Title', 'track_number': 1, 'disc_number': 1, 'duration_seconds': 201 }, { 'title': 'Song Title', 'track_number': 2, 'disc_number': 1, 'duration_seconds': 233 } ], source_ids={ 'musicbrainz_recording_id': f'recording-a-{track_number}', 'musicbrainz_release_id': 'release-a', 'musicbrainz_release_group_id': 'group-a' } ) release_b = build_candidate( provider='musicbrainz', is_authoritative=True, search_confidence=0.91, track_number=track_number, duration_seconds=duration_seconds, recording_id=f'recording-b-{track_number}', release_id='release-b', release_group_id='group-b', release_tracklist=[ { 'title': 'Track Zero', 'track_number': 1, 'disc_number': 1, 'duration_seconds': 120 }, { 'title': 'Track Extra', 'track_number': 2, 'disc_number': 1, 'duration_seconds': 310 } ], source_ids={ 'musicbrainz_recording_id': f'recording-b-{track_number}', 'musicbrainz_release_id': 'release-b', 'musicbrainz_release_group_id': 'group-b' } ) return [release_a, release_b] matcher = build_matcher( musicbrainz_candidates=dynamic_musicbrainz_candidates ) result_one = matcher.match_item(item_one, group, DEFAULT_CONFIG) result_two = matcher.match_item(item_two, group, DEFAULT_CONFIG) self.assertEqual(result_one['status'], 'matched') self.assertEqual(result_two['status'], 'matched') self.assertEqual(result_one['matched_metadata_json']['release_id'], 'release-a') self.assertEqual(result_two['matched_metadata_json']['release_id'], 'release-a') def build_matcher( *, acoustid_candidates=None, musicbrainz_candidates=None, aligned_candidate=None, netease_candidates=None, qq_candidates=None, spotify_candidates=None, acoustid_error=None, musicbrainz_error=None, netease_error=None, qq_error=None, spotify_error=None ): return Matcher( acoustid_provider=StaticSearchProvider(acoustid_candidates, error=acoustid_error), musicbrainz_provider=StaticMusicBrainzProvider( musicbrainz_candidates, aligned_candidate=aligned_candidate, error=musicbrainz_error ), netease_provider=StaticSearchProvider(netease_candidates, error=netease_error), qq_provider=StaticSearchProvider(qq_candidates, error=qq_error), spotify_provider=StaticSearchProvider(spotify_candidates, error=spotify_error), discogs_provider=StaticEnrichmentProvider(), lastfm_provider=StaticEnrichmentProvider(), genius_provider=StaticEnrichmentProvider() ) def build_item( *, title='Song Title', artist='Song Artist', album='Album Name', track_number=1, disc_number=1, duration_seconds=201, relative_path='Artist/Album/01.flac', filename='01.flac' ): return { 'id': 1, 'task_id': 'task-1', 'original_path': f'/tmp/{filename}', 'current_file_path': f'/tmp/{filename}', 'relative_path': relative_path, 'filename': filename, 'original_tags_json': { 'title': title, 'artist': artist, 'album': album, 'album_artist': artist, 'track_number': str(track_number), 'disc_number': str(disc_number) }, 'audio_props_json': { 'duration_seconds': duration_seconds }, 'acoustic_fingerprint': 'fingerprint', 'fingerprint_duration_seconds': duration_seconds, 'scan_status': 'queued', 'preprocess_status': 'completed' } def build_candidate( *, provider, is_authoritative, title='Song Title', artist='Song Artist', album='Album Name', track_number=1, disc_number=1, duration_seconds=201, recording_id='recording-main', release_id='release-main', release_group_id='release-group-main', fingerprint_confidence=None, search_confidence=None, release_tracklist=None, source_ids=None ): return { 'provider': provider, 'is_authoritative': is_authoritative, 'title': title, 'artist': artist, 'artists': [artist], 'album': album, 'album_artist': artist, 'track_number': track_number, 'disc_number': disc_number, 'release_date': '2024-01-01', 'year': 2024, 'duration_seconds': duration_seconds, 'recording_id': recording_id, 'release_id': release_id, 'release_group_id': release_group_id, 'source_ids': source_ids or { 'musicbrainz_recording_id': recording_id, 'musicbrainz_release_id': release_id, 'musicbrainz_release_group_id': release_group_id }, 'fingerprint_confidence': fingerprint_confidence, 'search_confidence': search_confidence, 'release_tracklist': release_tracklist or [ { 'title': title, 'track_number': track_number, 'disc_number': disc_number, 'duration_seconds': duration_seconds } ] } class StaticSearchProvider: def __init__(self, candidates=None, *, error=None): self.candidates = candidates or [] self.error = error def search(self, *args, **kwargs): if self.error: raise self.error if callable(self.candidates): return copy.deepcopy(self.candidates(*args, **kwargs)) return copy.deepcopy(self.candidates) class StaticMusicBrainzProvider: def __init__(self, candidates=None, *, aligned_candidate=None, error=None): self.candidates = candidates or [] self.aligned_candidate = aligned_candidate self.error = error def search_text(self, *args, **kwargs): if self.error: raise self.error if callable(self.candidates): return copy.deepcopy(self.candidates(*args, **kwargs)) return copy.deepcopy(self.candidates) def align_candidate(self, *args, **kwargs): if self.error: raise self.error return copy.deepcopy(self.aligned_candidate) class StaticEnrichmentProvider: def enrich(self, *args, **kwargs): return None if __name__ == '__main__': unittest.main()