669 lines
26 KiB
Python
669 lines
26 KiB
Python
import unittest
|
|
|
|
|
|
class MultiSourceSongResolverTests(unittest.TestCase):
|
|
def test_resolver_prefers_preferred_source_exact_candidate_before_cross_platform_search(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeClient:
|
|
def __init__(self, search_results=None, on_search=None):
|
|
self.search_results = list(search_results or [])
|
|
self.on_search = on_search
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
if self.on_search is not None:
|
|
self.on_search(keyword)
|
|
return list(self.search_results)
|
|
|
|
stale_song_info = SongInfo(
|
|
source="NeteaseMusicClient",
|
|
identifier="song-c",
|
|
song_name="Song C",
|
|
singers="Singer A / Singer B",
|
|
ext="mp3",
|
|
raw_data={"search": {"id": "song-c"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
netease_candidate = SongInfo(
|
|
source="NeteaseMusicClient",
|
|
identifier="song-c",
|
|
song_name="Song C",
|
|
singers="Singer A / Singer B",
|
|
ext="mp3",
|
|
file_size_bytes=1024,
|
|
file_size="1.00 MB",
|
|
raw_data={"quality": "standard"},
|
|
download_url="https://example.com/song-c.mp3",
|
|
download_url_status={"ok": True},
|
|
)
|
|
qq_candidate = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="qq-song-c",
|
|
song_name="Song C",
|
|
singers="Singer A / Singer B",
|
|
ext="flac",
|
|
file_size_bytes=4096,
|
|
file_size="4.00 MB",
|
|
raw_data={"quality": "lossless"},
|
|
download_url="https://example.com/song-c.flac",
|
|
download_url_status={"ok": True},
|
|
)
|
|
searched_sources: list[str] = []
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"netease": FakeClient([netease_candidate], on_search=lambda keyword: searched_sources.append("netease")),
|
|
"qq": FakeClient([qq_candidate], on_search=lambda keyword: searched_sources.append("qq")),
|
|
}[platform]
|
|
)
|
|
|
|
resolved_song_info = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "netease",
|
|
"name": "Song C",
|
|
"singers": "Singer A / Singer B",
|
|
"remote_song_id": "song-c",
|
|
},
|
|
snapshot_song_info=stale_song_info,
|
|
download_sources=["netease", "qq"],
|
|
)
|
|
|
|
self.assertEqual(["netease"], searched_sources)
|
|
self.assertEqual("NeteaseMusicClient", resolved_song_info.source)
|
|
self.assertEqual("mp3", resolved_song_info.ext)
|
|
self.assertEqual("song-c", resolved_song_info.identifier)
|
|
|
|
def test_resolver_stops_after_preferred_source_refresh_returns_downloadable_song(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class PreferredClient:
|
|
def __init__(self, refreshed_song):
|
|
self.refreshed_song = refreshed_song
|
|
self.search_called = False
|
|
|
|
def _parsewithofficialapiv1(self, search_result, request_overrides=None):
|
|
return self.refreshed_song
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
self.search_called = True
|
|
return []
|
|
|
|
class FallbackClient:
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
raise AssertionError("fallback source should not be searched")
|
|
|
|
snapshot_song_info = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="song-e",
|
|
song_name="Song E",
|
|
singers="Singer E",
|
|
ext="flac",
|
|
raw_data={"search": {"id": "song-e"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
refreshed_song = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="song-e",
|
|
song_name="Song E",
|
|
singers="Singer E",
|
|
ext="flac",
|
|
file_size_bytes=4096,
|
|
raw_data={"quality": "lossless"},
|
|
download_url="https://example.com/song-e.flac",
|
|
download_url_status={"ok": True},
|
|
)
|
|
preferred_client = PreferredClient(refreshed_song)
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"qq": preferred_client,
|
|
"kuwo": FallbackClient(),
|
|
}[platform]
|
|
)
|
|
|
|
resolved_song_info = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "qq",
|
|
"name": "Song E",
|
|
"singers": "Singer E",
|
|
"remote_song_id": "song-e",
|
|
},
|
|
snapshot_song_info=snapshot_song_info,
|
|
download_sources=["qq", "kuwo"],
|
|
)
|
|
|
|
self.assertFalse(preferred_client.search_called)
|
|
self.assertEqual("QQMusicClient", resolved_song_info.source)
|
|
self.assertEqual("song-e", resolved_song_info.identifier)
|
|
|
|
def test_resolver_attempts_preferred_source_first_even_when_not_in_download_sources(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeClient:
|
|
def __init__(self, source, search_results, calls):
|
|
self.source = source
|
|
self.search_results = list(search_results or [])
|
|
self.calls = calls
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
self.calls.append(self.source)
|
|
return list(self.search_results)
|
|
|
|
snapshot_song_info = SongInfo(
|
|
source="NeteaseMusicClient",
|
|
identifier="song-pref",
|
|
song_name="Song Preferred",
|
|
singers="Singer Preferred",
|
|
ext="mp3",
|
|
raw_data={"search": {"id": "song-pref"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
netease_candidate = SongInfo(
|
|
source="NeteaseMusicClient",
|
|
identifier="song-pref",
|
|
song_name="Song Preferred",
|
|
singers="Singer Preferred",
|
|
ext="mp3",
|
|
file_size_bytes=2048,
|
|
raw_data={"quality": "standard"},
|
|
download_url="https://example.com/song-pref.mp3",
|
|
download_url_status={"ok": True},
|
|
)
|
|
search_calls = []
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"netease": FakeClient("netease", [netease_candidate], search_calls),
|
|
"qq": FakeClient("qq", [], search_calls),
|
|
"kuwo": FakeClient("kuwo", [], search_calls),
|
|
}[platform]
|
|
)
|
|
|
|
resolved_song_info = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "netease",
|
|
"name": "Song Preferred",
|
|
"singers": "Singer Preferred",
|
|
"remote_song_id": "song-pref",
|
|
},
|
|
snapshot_song_info=snapshot_song_info,
|
|
download_sources=["qq", "kuwo"],
|
|
)
|
|
|
|
self.assertEqual(["netease"], search_calls)
|
|
self.assertEqual("NeteaseMusicClient", resolved_song_info.source)
|
|
self.assertEqual("song-pref", resolved_song_info.identifier)
|
|
|
|
def test_resolver_reports_source_attempts_to_progress_callback(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeClient:
|
|
def __init__(self, search_results=None):
|
|
self.search_results = list(search_results or [])
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
return list(self.search_results)
|
|
|
|
snapshot_song_info = SongInfo(
|
|
source="NeteaseMusicClient",
|
|
identifier="song-d",
|
|
song_name="Song D",
|
|
singers="Singer D",
|
|
ext="mp3",
|
|
raw_data={"search": {"id": "song-d"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
kuwo_candidate = SongInfo(
|
|
source="KuwoMusicClient",
|
|
identifier="kuwo-song-d",
|
|
song_name="Song D",
|
|
singers="Singer D",
|
|
ext="flac",
|
|
file_size_bytes=4096,
|
|
raw_data={"quality": "lossless"},
|
|
download_url="https://example.com/song-d.flac",
|
|
download_url_status={"ok": True},
|
|
)
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"netease": FakeClient([]),
|
|
"qq": FakeClient([]),
|
|
"kuwo": FakeClient([kuwo_candidate]),
|
|
}[platform]
|
|
)
|
|
progress_messages: list[str] = []
|
|
|
|
resolved_song_info = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "netease",
|
|
"name": "Song D",
|
|
"singers": "Singer D",
|
|
"remote_song_id": "song-d",
|
|
},
|
|
snapshot_song_info=snapshot_song_info,
|
|
download_sources=["qq", "kuwo"],
|
|
progress_callback=progress_messages.append,
|
|
)
|
|
|
|
self.assertEqual(
|
|
[
|
|
"resolving source netease (1/3)",
|
|
"resolving source qq (2/3)",
|
|
"resolving source kuwo (3/3)",
|
|
],
|
|
progress_messages,
|
|
)
|
|
self.assertEqual("KuwoMusicClient", resolved_song_info.source)
|
|
self.assertEqual("kuwo-song-d", resolved_song_info.identifier)
|
|
|
|
def test_resolver_uses_ranked_top_two_fallback_sources_after_warmup(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeStatsRepo:
|
|
def __init__(self):
|
|
self.rank_call = None
|
|
self.records = []
|
|
|
|
def rank_fallback_sources(self, origin_source, fallback_sources, warmup_attempts=1000):
|
|
self.rank_call = (origin_source, list(fallback_sources), warmup_attempts)
|
|
return ["migu", "kuwo", "qianqian"]
|
|
|
|
def record_fallback_result(self, origin_source, candidate_source, *, succeeded):
|
|
self.records.append((origin_source, candidate_source, succeeded))
|
|
|
|
class FakeClient:
|
|
def __init__(self, source, search_results, calls):
|
|
self.source = source
|
|
self.search_results = list(search_results or [])
|
|
self.calls = calls
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
self.calls.append(self.source)
|
|
return list(self.search_results)
|
|
|
|
snapshot = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="song-1",
|
|
song_name="Song 1",
|
|
singers="Singer 1",
|
|
raw_data={"search": {"id": "song-1"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
migu_hit = SongInfo(
|
|
source="MiguMusicClient",
|
|
identifier="migu-song-1",
|
|
song_name="Song 1",
|
|
singers="Singer 1",
|
|
ext="mp3",
|
|
download_url="https://example.com/song-1.mp3",
|
|
download_url_status={"ok": True},
|
|
)
|
|
search_calls = []
|
|
stats_repo = FakeStatsRepo()
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"qq": FakeClient("qq", [], search_calls),
|
|
"kuwo": FakeClient("kuwo", [], search_calls),
|
|
"migu": FakeClient("migu", [migu_hit], search_calls),
|
|
"qianqian": FakeClient("qianqian", [], search_calls),
|
|
}[platform],
|
|
resolver_stats_repo=stats_repo,
|
|
)
|
|
|
|
resolved = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "qq",
|
|
"name": "Song 1",
|
|
"singers": "Singer 1",
|
|
"remote_song_id": "song-1",
|
|
},
|
|
snapshot_song_info=snapshot,
|
|
download_sources=["qq", "kuwo", "migu", "qianqian"],
|
|
)
|
|
|
|
self.assertEqual(
|
|
("qq", ["kuwo", "migu", "qianqian"], 1000),
|
|
stats_repo.rank_call,
|
|
)
|
|
self.assertEqual(["qq", "migu"], search_calls)
|
|
self.assertEqual([("qq", "migu", True)], stats_repo.records)
|
|
self.assertEqual("MiguMusicClient", resolved.source)
|
|
|
|
def test_resolver_continues_after_ranked_top_two_fail(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeStatsRepo:
|
|
def __init__(self):
|
|
self.records = []
|
|
|
|
def rank_fallback_sources(self, origin_source, fallback_sources, warmup_attempts=1000):
|
|
return ["migu", "kuwo", "qianqian"]
|
|
|
|
def record_fallback_result(self, origin_source, candidate_source, *, succeeded):
|
|
self.records.append((origin_source, candidate_source, succeeded))
|
|
|
|
class FakeClient:
|
|
def __init__(self, source, search_results, calls):
|
|
self.source = source
|
|
self.search_results = list(search_results or [])
|
|
self.calls = calls
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
self.calls.append(self.source)
|
|
return list(self.search_results)
|
|
|
|
snapshot = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="song-2",
|
|
song_name="Song 2",
|
|
singers="Singer 2",
|
|
raw_data={"search": {"id": "song-2"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
qianqian_hit = SongInfo(
|
|
source="QianqianMusicClient",
|
|
identifier="qianqian-song-2",
|
|
song_name="Song 2",
|
|
singers="Singer 2",
|
|
ext="mp3",
|
|
download_url="https://example.com/song-2.mp3",
|
|
download_url_status={"ok": True},
|
|
)
|
|
search_calls = []
|
|
stats_repo = FakeStatsRepo()
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"qq": FakeClient("qq", [], search_calls),
|
|
"migu": FakeClient("migu", [], search_calls),
|
|
"kuwo": FakeClient("kuwo", [], search_calls),
|
|
"qianqian": FakeClient("qianqian", [qianqian_hit], search_calls),
|
|
}[platform],
|
|
resolver_stats_repo=stats_repo,
|
|
)
|
|
|
|
resolved = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "qq",
|
|
"name": "Song 2",
|
|
"singers": "Singer 2",
|
|
"remote_song_id": "song-2",
|
|
},
|
|
snapshot_song_info=snapshot,
|
|
download_sources=["qq", "kuwo", "migu", "qianqian"],
|
|
)
|
|
|
|
self.assertEqual(["qq", "migu", "kuwo", "qianqian"], search_calls)
|
|
self.assertEqual(
|
|
[
|
|
("qq", "migu", False),
|
|
("qq", "kuwo", False),
|
|
("qq", "qianqian", True),
|
|
],
|
|
stats_repo.records,
|
|
)
|
|
self.assertEqual("QianqianMusicClient", resolved.source)
|
|
|
|
def test_resolver_continues_to_fallback_when_preferred_client_factory_raises(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeClient:
|
|
def __init__(self, source, search_results, calls):
|
|
self.source = source
|
|
self.search_results = list(search_results or [])
|
|
self.calls = calls
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
self.calls.append(self.source)
|
|
return list(self.search_results)
|
|
|
|
snapshot_song_info = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="song-pref-fail",
|
|
song_name="Song Preferred Fail",
|
|
singers="Singer Preferred Fail",
|
|
raw_data={"search": {"id": "song-pref-fail"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
migu_hit = SongInfo(
|
|
source="MiguMusicClient",
|
|
identifier="migu-song-pref-fail",
|
|
song_name="Song Preferred Fail",
|
|
singers="Singer Preferred Fail",
|
|
ext="mp3",
|
|
download_url="https://example.com/song-pref-fail.mp3",
|
|
download_url_status={"ok": True},
|
|
)
|
|
search_calls = []
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"kuwo": FakeClient("kuwo", [], search_calls),
|
|
"migu": FakeClient("migu", [migu_hit], search_calls),
|
|
}[platform]
|
|
)
|
|
|
|
resolved_song_info = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "qq",
|
|
"name": "Song Preferred Fail",
|
|
"singers": "Singer Preferred Fail",
|
|
"remote_song_id": "song-pref-fail",
|
|
},
|
|
snapshot_song_info=snapshot_song_info,
|
|
download_sources=["kuwo", "migu"],
|
|
)
|
|
|
|
self.assertEqual(["kuwo", "migu"], search_calls)
|
|
self.assertEqual("MiguMusicClient", resolved_song_info.source)
|
|
self.assertEqual("migu-song-pref-fail", resolved_song_info.identifier)
|
|
|
|
def test_resolver_uses_configured_fallback_order_when_rank_lookup_raises(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeStatsRepo:
|
|
def __init__(self):
|
|
self.records = []
|
|
|
|
def rank_fallback_sources(self, origin_source, fallback_sources, warmup_attempts=1000):
|
|
raise RuntimeError("rank unavailable")
|
|
|
|
def record_fallback_result(self, origin_source, candidate_source, *, succeeded):
|
|
self.records.append((origin_source, candidate_source, succeeded))
|
|
|
|
class FakeClient:
|
|
def __init__(self, source, search_results, calls):
|
|
self.source = source
|
|
self.search_results = list(search_results or [])
|
|
self.calls = calls
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
self.calls.append(self.source)
|
|
return list(self.search_results)
|
|
|
|
snapshot_song_info = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="song-rank-fail",
|
|
song_name="Song Rank Fail",
|
|
singers="Singer Rank Fail",
|
|
raw_data={"search": {"id": "song-rank-fail"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
kuwo_hit = SongInfo(
|
|
source="KuwoMusicClient",
|
|
identifier="kuwo-song-rank-fail",
|
|
song_name="Song Rank Fail",
|
|
singers="Singer Rank Fail",
|
|
ext="mp3",
|
|
download_url="https://example.com/song-rank-fail.mp3",
|
|
download_url_status={"ok": True},
|
|
)
|
|
search_calls = []
|
|
stats_repo = FakeStatsRepo()
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"qq": FakeClient("qq", [], search_calls),
|
|
"kuwo": FakeClient("kuwo", [kuwo_hit], search_calls),
|
|
"migu": FakeClient("migu", [], search_calls),
|
|
}[platform],
|
|
resolver_stats_repo=stats_repo,
|
|
)
|
|
|
|
resolved_song_info = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "qq",
|
|
"name": "Song Rank Fail",
|
|
"singers": "Singer Rank Fail",
|
|
"remote_song_id": "song-rank-fail",
|
|
},
|
|
snapshot_song_info=snapshot_song_info,
|
|
download_sources=["qq", "kuwo", "migu"],
|
|
)
|
|
|
|
self.assertEqual(["qq", "kuwo"], search_calls)
|
|
self.assertEqual([("qq", "kuwo", True)], stats_repo.records)
|
|
self.assertEqual("KuwoMusicClient", resolved_song_info.source)
|
|
self.assertEqual("kuwo-song-rank-fail", resolved_song_info.identifier)
|
|
|
|
def test_resolver_continues_when_record_fallback_result_raises(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeStatsRepo:
|
|
def rank_fallback_sources(self, origin_source, fallback_sources, warmup_attempts=1000):
|
|
return ["migu", "kuwo"]
|
|
|
|
def record_fallback_result(self, origin_source, candidate_source, *, succeeded):
|
|
raise RuntimeError("record unavailable")
|
|
|
|
class FakeClient:
|
|
def __init__(self, source, search_results, calls):
|
|
self.source = source
|
|
self.search_results = list(search_results or [])
|
|
self.calls = calls
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
self.calls.append(self.source)
|
|
return list(self.search_results)
|
|
|
|
snapshot_song_info = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="song-record-fail",
|
|
song_name="Song Record Fail",
|
|
singers="Singer Record Fail",
|
|
raw_data={"search": {"id": "song-record-fail"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
migu_hit = SongInfo(
|
|
source="MiguMusicClient",
|
|
identifier="migu-song-record-fail",
|
|
song_name="Song Record Fail",
|
|
singers="Singer Record Fail",
|
|
ext="mp3",
|
|
download_url="https://example.com/song-record-fail.mp3",
|
|
download_url_status={"ok": True},
|
|
)
|
|
search_calls = []
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"qq": FakeClient("qq", [], search_calls),
|
|
"migu": FakeClient("migu", [migu_hit], search_calls),
|
|
"kuwo": FakeClient("kuwo", [], search_calls),
|
|
}[platform],
|
|
resolver_stats_repo=FakeStatsRepo(),
|
|
)
|
|
|
|
resolved_song_info = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "qq",
|
|
"name": "Song Record Fail",
|
|
"singers": "Singer Record Fail",
|
|
"remote_song_id": "song-record-fail",
|
|
},
|
|
snapshot_song_info=snapshot_song_info,
|
|
download_sources=["qq", "kuwo", "migu"],
|
|
)
|
|
|
|
self.assertEqual(["qq", "migu"], search_calls)
|
|
self.assertEqual("MiguMusicClient", resolved_song_info.source)
|
|
self.assertEqual("migu-song-record-fail", resolved_song_info.identifier)
|
|
|
|
def test_resolver_continues_when_first_fallback_client_factory_raises(self):
|
|
from musicdl.catalogsync.resolver import MultiSourceSongResolver
|
|
from musicdl.modules.utils.data import SongInfo
|
|
|
|
class FakeStatsRepo:
|
|
def rank_fallback_sources(self, origin_source, fallback_sources, warmup_attempts=1000):
|
|
return ["migu", "kuwo"]
|
|
|
|
def record_fallback_result(self, origin_source, candidate_source, *, succeeded):
|
|
return None
|
|
|
|
class FakeClient:
|
|
def __init__(self, source, search_results, calls):
|
|
self.source = source
|
|
self.search_results = list(search_results or [])
|
|
self.calls = calls
|
|
|
|
def search(self, keyword, num_threadings=1, request_overrides=None, rule=None, main_process_context=None):
|
|
self.calls.append(self.source)
|
|
return list(self.search_results)
|
|
|
|
snapshot_song_info = SongInfo(
|
|
source="QQMusicClient",
|
|
identifier="song-fallback-factory-fail",
|
|
song_name="Song Fallback Factory Fail",
|
|
singers="Singer Fallback Factory Fail",
|
|
raw_data={"search": {"id": "song-fallback-factory-fail"}},
|
|
download_url=None,
|
|
download_url_status={},
|
|
)
|
|
kuwo_hit = SongInfo(
|
|
source="KuwoMusicClient",
|
|
identifier="kuwo-song-fallback-factory-fail",
|
|
song_name="Song Fallback Factory Fail",
|
|
singers="Singer Fallback Factory Fail",
|
|
ext="mp3",
|
|
download_url="https://example.com/song-fallback-factory-fail.mp3",
|
|
download_url_status={"ok": True},
|
|
)
|
|
search_calls = []
|
|
resolver = MultiSourceSongResolver(
|
|
client_factory=lambda platform: {
|
|
"qq": FakeClient("qq", [], search_calls),
|
|
"kuwo": FakeClient("kuwo", [kuwo_hit], search_calls),
|
|
}[platform],
|
|
resolver_stats_repo=FakeStatsRepo(),
|
|
)
|
|
|
|
resolved_song_info = resolver.resolve_song_info(
|
|
row={
|
|
"platform": "qq",
|
|
"name": "Song Fallback Factory Fail",
|
|
"singers": "Singer Fallback Factory Fail",
|
|
"remote_song_id": "song-fallback-factory-fail",
|
|
},
|
|
snapshot_song_info=snapshot_song_info,
|
|
download_sources=["qq", "migu", "kuwo"],
|
|
)
|
|
|
|
self.assertEqual(["qq", "kuwo"], search_calls)
|
|
self.assertEqual("KuwoMusicClient", resolved_song_info.source)
|
|
self.assertEqual("kuwo-song-fallback-factory-fail", resolved_song_info.identifier)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|