GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-21 16:23:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(music): complete album subscription workflow

Browse Source
This commit is contained in:
jxxghp
2026-08-09 20:43:33 +08:00
parent a94202a28d
commit 393a4ca4e2
30 changed files with 1402 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
tests/test_mediaserver_sync_incremental.py
View File

@@ -57,10 +57,61 @@ def test_media_count_reuses_existing_server_statistics():
"""整服同步应复用现有媒体统计并排除剧集集数。"""
chain = object.__new__(MediaServerChain)
chain.run_module = lambda *_args, **_kwargs: [
schemas.Statistic(movie_count=12, tv_count=8, episode_count=200)
schemas.Statistic(movie_count=12, tv_count=8, music_count=3, episode_count=200)
]
assert chain.media_count("plex") == 20
assert chain.media_count("plex") == 23
@pytest.mark.parametrize(
("raw_type", "expected"),
[
("Series", "电视剧"),
("show", "电视剧"),
("音乐", "音乐"),
("MusicAlbum", "音乐"),
("Audio", "音乐"),
("Movie", "电影"),
],
)
def test_sync_normalizes_movie_tv_and_music_item_types(raw_type, expected):
"""同步缓存应保留音乐类型，并兼容不同媒体服务器的原始类型名称。"""
assert MediaServerChain._normalize_item_type(raw_type) == expected
def test_sync_persists_music_without_querying_tv_episodes(database):
"""Navidrome 专辑同步应写成音乐条目，且不能触发电视剧分集查询。"""
chain = object.__new__(MediaServerChain)
chain.librarys = lambda _server: [SimpleNamespace(id="music", name="音乐")]
chain.media_count = lambda _server: 1
chain.items_count = lambda **_kwargs: pytest.fail("整服统计存在时不应逐库计数")
chain.items = lambda **_kwargs: iter(
[
schemas.MediaServerItem(
server="navidrome",
library="music",
item_id="album-1",
item_type="音乐",
title="叶惠美",
year="2003",
)
]
)
chain.episodes = lambda *_args, **_kwargs: pytest.fail("音乐条目不应查询电视剧分集")
with patch("app.db.ScopedSession", database), patch.object(
MEDIA_SERVER_CHAIN_MODULE.ServiceConfigHelper,
"get_mediaserver_configs",
return_value=[SimpleNamespace(name="navidrome", enabled=True, sync_libraries=["all"])],
):
chain.sync()
with database() as db:
item = db.query(MediaServerItem).one()
assert item.item_type == "音乐"
assert item.title == "叶惠美"
assert item.seasoninfo == {}
def test_sync_updates_rows_and_removes_stale_entries(database):

44
tests/test_music_chain.py
View File

@@ -40,6 +40,38 @@ def test_build_site_keywords_prefers_artist_album():
]
def test_album_resource_match_requires_selected_album_title():
"""专辑订阅只接受包含目标专辑名的站点资源，忽略大小写、空格和标点差异。"""
album = MusicInfo(
music_type="album",
title="Random Access Memories",
album="Random Access Memories",
names=["Random-Access Memories"],
)
assert MusicChain.matches_site_resource(
album,
"Daft.Punk-Random.Access.Memories-2013-FLAC",
) is True
assert MusicChain.matches_site_resource(album, "Daft Punk - Discovery - FLAC") is False
def test_recording_resource_match_does_not_treat_album_name_as_track_alias():
"""单曲候选的兼容 names 即使包含专辑名，也不能让整专标题冒充目标单曲。"""
recording = MusicInfo(
music_type="recording",
title="Get Lucky",
album="Random Access Memories",
names=["Get Lucky", "Random Access Memories"],
)
assert MusicChain.matches_site_resource(recording, "Daft Punk - Get Lucky FLAC") is True
assert MusicChain.matches_site_resource(
recording,
"Daft Punk - Random Access Memories FLAC",
) is False
def test_normalize_candidates_deduplicates_source_identity():
"""同一来源和媒体 ID 的音乐候选应只保留一次。"""
results = MusicChain.normalize_candidates(
@@ -58,6 +90,18 @@ def test_normalize_candidates_deduplicates_source_identity():
assert results[0].title == "A"
def test_normalize_candidates_keeps_different_entities_with_same_source_id():
"""同一来源 ID 在不同音乐实体命名空间下不能互相去重。"""
results = MusicChain.normalize_candidates(
[
MusicInfo(source="musicbrainz", media_id="shared-id", music_type="recording", title="Song"),
MusicInfo(source="musicbrainz", media_id="shared-id", music_type="album", title="Album"),
]
)
assert [item.music_type for item in results] == ["recording", "album"]
def test_normalize_candidates_deduplicates_metadata_without_id():
"""缺少来源 ID 时应按标题、艺术家和专辑去重。"""
results = MusicChain.normalize_candidates(

117
tests/test_music_download.py
View File

@@ -3,7 +3,8 @@ from unittest.mock import Mock, patch
from app.api.endpoints.download import download
from app.chain.download import DownloadChain
from app.chain.music import MusicChain
from app.core.context import MusicInfo
from app.core.context import MUSIC_ENTITY_ALBUM, Context, MusicInfo
from app.schemas import ExistMediaInfo
from app.schemas.context import TorrentInfo
from app.schemas.music import MusicInfo as MusicInfoSchema
from app.schemas.types import MediaType
@@ -23,6 +24,19 @@ def _music_info() -> MusicInfo:
)
def _album_info(total_tracks: int | None = 3) -> MusicInfo:
"""构造整张专辑下载校验使用的目标信息。"""
return MusicInfo(
source="musicbrainz",
media_id="release-group-1",
music_type=MUSIC_ENTITY_ALBUM,
title="叶惠美",
album="叶惠美",
artists=["周杰伦"],
total_tracks=total_tracks,
)
def test_music_info_exposes_download_chain_compatibility_fields():
"""音乐信息应安全兼容下载链现有的视频身份字段访问。"""
info = _music_info()
@@ -49,6 +63,73 @@ def test_download_note_keeps_versioned_music_context():
assert "raw_data" not in note["music"]["media"]
def test_album_resource_requires_all_independent_audio_tracks():
"""整专资源只有在独立音频文件数覆盖专辑曲目数时才可标记完整。"""
context = Context(media_info=_album_info(total_tracks=3))
error = DownloadChain._validate_music_album_resource(
context,
["叶惠美/01.flac", "叶惠美/02.flac", "叶惠美/03.m4a", "叶惠美/cover.jpg"],
)
assert error is None
assert context.confirmed_full_coverage is True
def test_album_resource_rejects_incomplete_or_unverifiable_pack():
"""曲目不足、未知曲目总数或无文件清单时不得把专辑订阅判定为完成。"""
incomplete = Context(media_info=_album_info(total_tracks=3))
unknown = Context(media_info=_album_info(total_tracks=None))
assert "仅包含 1 个独立音频文件" in (
DownloadChain._validate_music_album_resource(incomplete, ["叶惠美/disc.flac"]) or ""
)
assert incomplete.confirmed_full_coverage is False
assert "总曲目数未知" in (
DownloadChain._validate_music_album_resource(unknown, ["叶惠美/01.flac"]) or ""
)
assert "未提供文件清单" in (
DownloadChain._validate_music_album_resource(
Context(media_info=_album_info(total_tracks=3)),
[],
) or ""
)
def test_download_single_stops_before_client_when_album_pack_is_incomplete():
"""下载入口应在添加任务前拒绝不完整专辑，并记录可供后续候选继续尝试的失败原因。"""
context = Context(
media_info=_album_info(total_tracks=3),
meta_info=MusicChain.to_meta(_album_info(total_tracks=3)),
torrent_info=TorrentInfo(
title="周杰伦 - 叶惠美 FLAC",
category=MediaType.MUSIC.value,
),
)
chain = DownloadChain()
chain._record_download_failure = Mock()
media_chain = Mock()
media_chain.supplement_tmdb_info.return_value = context.media_info
torrent_helper = Mock()
torrent_helper.get_fileinfo_from_torrent_content.return_value = (
"叶惠美",
["叶惠美/整轨.flac", "叶惠美/整轨.cue"],
)
with patch("app.chain.download.MediaChain", return_value=media_chain), \
patch("app.chain.download.TorrentHelper", return_value=torrent_helper), \
patch("app.chain.download.eventmanager.send_event", return_value=None):
task_id, error = chain.download_single(
context,
torrent_content=b"torrent",
return_detail=True,
)
assert task_id is None
assert "专辑资源不完整" in error
chain._record_download_failure.assert_called_once()
def test_download_endpoint_builds_music_context():
"""现有添加下载接口应使用 MusicInfo 和 MetaMusic 构造音乐上下文。"""
chain = Mock()
@@ -74,3 +155,37 @@ def test_download_endpoint_builds_music_context():
assert context.media_info.media_id == "recording-1"
assert context.meta_info.type == MediaType.MUSIC
assert context.meta_info.org_string == "周杰伦 - 叶惠美 FLAC"
def test_music_library_exists_uses_atomic_album_lookup():
"""整专存在性检查应按音乐条目判断，不能落入电视剧季集补全分支。"""
album = _album_info(total_tracks=11)
chain = DownloadChain()
chain.media_exists = Mock(
return_value=ExistMediaInfo(
type=MediaType.MUSIC,
server_type="navidrome",
server="music",
itemid="album-item-1",
)
)
mediaserver = Mock()
mediaserver.get_item_id.return_value = "album-item-1"
with patch("app.chain.download.MediaServerOper", return_value=mediaserver):
exists, no_exists = chain.get_no_exists_info(
meta=MusicChain.to_meta(album),
mediainfo=album,
)
assert exists is True
assert no_exists == {}
mediaserver.get_item_id.assert_called_once_with(
mtype=MediaType.MUSIC.value,
title="叶惠美",
year=None,
)
chain.media_exists.assert_called_once_with(
mediainfo=album,
itemid="album-item-1",
)

12
tests/test_music_recognize_routing.py
View File

@@ -84,22 +84,22 @@ def test_musicbrainz_module_recognize_media_uses_detail_when_meta_has_identity(m
meta = MetaMusic(title="晴天", media_source="musicbrainz", media_id="recording-1")
expected = _music_info()
monkeypatch.setattr(module, "recognize_music", Mock(return_value=expected))
search_mock = Mock(return_value=[])
monkeypatch.setattr(module, "search_music", search_mock)
recording_search = Mock(return_value=[])
monkeypatch.setattr(module, "_search_recordings", recording_search)
result = module.recognize_media(meta=meta, source="musicbrainz")
module.recognize_music.assert_called_once_with("musicbrainz", "recording-1")
search_mock.assert_not_called()
recording_search.assert_not_called()
assert result is expected
def test_musicbrainz_module_recognize_media_matches_search_candidate(monkeypatch):
"""无身份时应按标题搜索并选择匹配候选。"""
"""无身份时应从 Recording 搜索中选择匹配候选。"""
module = MusicBrainzModule()
meta = MetaMusic(title="晴天", artists=["周杰伦"], album="叶惠美")
candidate = _music_info()
monkeypatch.setattr(module, "search_music", Mock(return_value=[candidate]))
monkeypatch.setattr(module, "_search_recordings", Mock(return_value=[candidate]))
result = module.recognize_media(meta=meta)
@@ -110,7 +110,7 @@ def test_musicbrainz_module_recognize_media_falls_back_to_offline_when_no_match(
"""搜索无候选时应返回元数据兜底，且兜底结果不带远端身份。"""
module = MusicBrainzModule()
meta = MetaMusic(title="未知曲目", artists=["未知艺术家"])
monkeypatch.setattr(module, "search_music", Mock(return_value=[]))
monkeypatch.setattr(module, "_search_recordings", Mock(return_value=[]))
result = module.recognize_media(meta=meta)

104
tests/test_music_scrape.py Normal file
View File

@@ -0,0 +1,104 @@
from types import SimpleNamespace
from unittest.mock import Mock
from app.chain.media import MediaChain
from app.core.context import MUSIC_ENTITY_ALBUM, MusicInfo
from app.core.meta import MetaMusic
from app.schemas import FileItem
def _media_chain() -> MediaChain:
"""构造不注册全局单例的音乐刮削链测试实例。"""
return object.__new__(MediaChain)
def _album_info() -> MusicInfo:
"""构造专辑批量刮削使用的标准目标。"""
return MusicInfo(
source="musicbrainz",
media_id="release-group-1",
music_type=MUSIC_ENTITY_ALBUM,
title="叶惠美",
artists=["周杰伦"],
album="叶惠美",
album_artist="周杰伦",
year=2003,
total_tracks=11,
cover_url="https://example.com/album.jpg",
)
def test_album_scrape_merge_preserves_track_fields_and_applies_album_identity() -> None:
"""专辑批量刮削应保留每首歌自己的曲名和曲序，只统一专辑级字段。"""
local = MetaMusic(
title="晴天",
artists=["周杰伦"],
album="错误专辑",
album_artist="错误艺术家",
year=1999,
track_number=3,
total_tracks=99,
)
merged = MediaChain._merge_music_album_metadata(local, _album_info())
assert merged.title == "晴天"
assert merged.artists == ["周杰伦"]
assert merged.track_number == 3
assert merged.album == "叶惠美"
assert merged.album_artist == "周杰伦"
assert merged.year == 2003
assert merged.total_tracks == 11
assert merged.media_source == "musicbrainz"
assert merged.media_id == "release-group-1"
def test_album_directory_scrape_processes_each_track_and_reuses_cover() -> None:
"""显式选择专辑刮削目录时应逐曲写标签，并让整批文件共用一次封面下载。"""
chain = _media_chain()
chain.storagechain = Mock()
chain.scraping_policies = Mock()
chain.scraping_policies.option.return_value = SimpleNamespace(
is_skip=False,
is_overwrite=False,
)
audio_files = [
FileItem(storage="local", path="/music/叶惠美/01.flac", type="file", name="01.flac"),
FileItem(storage="local", path="/music/叶惠美/02.m4a", type="file", name="02.m4a"),
]
chain.storagechain.list_files.return_value = audio_files
chain._download_music_cover = Mock(return_value=(b"cover", "image/jpeg"))
chain._scrape_music_file = Mock(return_value=True)
album = _album_info()
success, message = chain.scrape_music_metadata(
FileItem(storage="local", path="/music/叶惠美", type="dir", name="叶惠美"),
mediainfo=album,
)
assert success is True
assert message == "已刮削 2 个音频文件"
chain._download_music_cover.assert_called_once_with(album.cover_url)
assert chain._scrape_music_file.call_count == 2
assert all(
call.args[1] is album and call.kwargs["cover"] == (b"cover", "image/jpeg")
for call in chain._scrape_music_file.call_args_list
)
def test_recording_identity_rejects_multi_track_directory_scrape() -> None:
"""单曲身份不得覆盖整目录，否则会把同一首歌的标签写到专辑内所有文件。"""
chain = _media_chain()
chain.storagechain = Mock()
chain.storagechain.list_files.return_value = [
FileItem(storage="local", path="/music/01.flac", type="file"),
FileItem(storage="local", path="/music/02.flac", type="file"),
]
success, message = chain.scrape_music_metadata(
FileItem(storage="local", path="/music", type="dir"),
mediainfo=MusicInfo(title="晴天", music_type="recording"),
)
assert success is False
assert message == "单曲 MusicBrainz ID 仅支持刮削单个音频文件，整目录请选择专辑"

40
tests/test_music_search.py
View File

@@ -19,7 +19,12 @@ def test_music_context_builder_keeps_only_music_category():
)
torrents = [
TorrentInfo(
title="Daft Punk - Random Access Memories FLAC",
title="Daft Punk - Get Lucky - Random Access Memories FLAC",
category=MediaType.MUSIC.value,
site_name="MusicSite",
),
TorrentInfo(
title="Daft Punk - Discovery FLAC",
category=MediaType.MUSIC.value,
site_name="MusicSite",
),
@@ -44,6 +49,39 @@ def test_music_context_builder_keeps_only_music_category():
assert contexts[0].torrent_info.category == MediaType.MUSIC.value
def test_music_search_continues_after_unrelated_first_keyword_results():
"""首组关键词只命中其它专辑时应继续尝试后续关键词，不能提前返回空结果。"""
chain = SearchChain()
music = MusicInfo(
source="musicbrainz",
media_id="recording-1",
title="Get Lucky",
artists=["Daft Punk"],
album="Random Access Memories",
)
unrelated = TorrentInfo(
title="Daft Punk - Discovery FLAC",
category=MediaType.MUSIC.value,
site_name="MusicSite",
)
matched = TorrentInfo(
title="Daft Punk - Get Lucky FLAC",
category=MediaType.MUSIC.value,
site_name="MusicSite",
)
with patch.object(
chain,
"_SearchChain__search_all_sites",
side_effect=[[unrelated], [matched]],
) as search_sites, patch("app.chain.search.time.sleep"):
contexts = chain._process_music(music, rule_groups=[])
assert search_sites.call_count == 2
assert len(contexts) == 1
assert contexts[0].torrent_info.title == matched.title
def test_search_by_id_routes_music_identity_to_recognize_and_process():
"""MusicBrainz 精确身份搜索应经统一识别入口识别后进入现有搜索处理链。"""
chain = SearchChain()

139
tests/test_music_subscribe.py
View File

@@ -2,7 +2,7 @@ from types import SimpleNamespace
from unittest.mock import Mock, patch
from app.chain.subscribe import SubscribeChain, build_subscribe_meta
from app.core.context import Context, TorrentInfo
from app.core.context import MUSIC_ENTITY_ALBUM, Context, TorrentInfo
from app.core.meta import MetaMusic
from app.core.context import MusicInfo
from app.schemas.types import MediaType
@@ -20,9 +20,9 @@ def _music_info() -> MusicInfo:
)
def _subscribe() -> SimpleNamespace:
def _subscribe(**overrides) -> SimpleNamespace:
"""构造不依赖数据库的音乐订阅对象。"""
return SimpleNamespace(
values = dict(
id=7,
name="晴天",
year="2003",
@@ -30,6 +30,8 @@ def _subscribe() -> SimpleNamespace:
keyword=None,
media_source="musicbrainz",
media_id="recording-1",
music_type="recording",
total_tracks=None,
season=None,
episode_group=None,
tmdbid=None,
@@ -53,7 +55,11 @@ def _subscribe() -> SimpleNamespace:
best_version=0,
state="R",
note=None,
poster=None,
backdrop=None,
)
values.update(overrides)
return SimpleNamespace(**values)
def test_build_subscribe_meta_returns_music_meta():
@@ -72,7 +78,7 @@ def test_music_subscribe_reuses_search_download_and_finish_flow():
target = _music_info()
context = Context(
torrent_info=TorrentInfo(
title="周杰伦 - 叶惠美 FLAC",
title="周杰伦 - 晴天 FLAC",
category=MediaType.MUSIC.value,
)
)
@@ -98,7 +104,7 @@ def test_music_subscribe_reuses_search_download_and_finish_flow():
)
assert context.media_info is target
assert isinstance(context.meta_info, MetaMusic)
assert context.meta_info.org_string == "周杰伦 - 叶惠美 FLAC"
assert context.meta_info.org_string == "周杰伦 - 晴天 FLAC"
download_chain.batch_download.assert_called_once()
chain.finish_subscribe_or_not.assert_called_once()
@@ -124,6 +130,129 @@ def test_music_subscribe_ignores_non_music_category():
download_chain.assert_not_called()
def test_music_subscribe_ignores_unrelated_music_title():
"""即使站点分类为音乐，资源标题不含目标单曲或专辑名时也不得自动下载。"""
subscribe = _subscribe()
context = Context(
torrent_info=TorrentInfo(
title="周杰伦 - 七里香 FLAC",
category=MediaType.MUSIC.value,
)
)
search_chain = Mock()
search_chain.search_by_title.return_value = [context]
with patch.object(SubscribeChain, "_recognize_music_subscribe", return_value=_music_info()), \
patch("app.chain.subscribe.SearchChain", return_value=search_chain), \
patch("app.chain.subscribe.DownloadChain") as download_chain:
SubscribeChain()._search_music_subscribe(subscribe)
download_chain.assert_not_called()
def test_album_subscription_uses_persisted_snapshot_when_remote_detail_is_unavailable():
"""远端详情短暂失败时应从订阅快照恢复专辑语义，不能按标题猜成第一首单曲。"""
subscribe = _subscribe(
name="叶惠美",
media_id="release-group-1",
music_type=MUSIC_ENTITY_ALBUM,
total_tracks=11,
)
media_chain = Mock()
media_chain.recognize_media.return_value = None
with patch("app.chain.subscribe.MediaChain", return_value=media_chain), \
patch("app.chain.subscribe.MusicChain.search") as search:
restored = SubscribeChain._recognize_music_subscribe(subscribe)
assert restored.music_type == MUSIC_ENTITY_ALBUM
assert restored.album == "叶惠美"
assert restored.total_tracks == 11
search.assert_not_called()
def test_legacy_music_identity_failure_does_not_guess_entity_from_title():
"""旧订阅有标准 ID 却无实体类型时，识别失败后应保留订阅而不是误选标题搜索首项。"""
subscribe = _subscribe(music_type=None)
media_chain = Mock()
media_chain.recognize_media.return_value = None
with patch("app.chain.subscribe.MediaChain", return_value=media_chain), \
patch("app.chain.subscribe.MusicChain.search") as search:
restored = SubscribeChain._recognize_music_subscribe(subscribe)
assert restored is None
search.assert_not_called()
def test_album_subscription_without_remote_id_uses_persisted_entity_snapshot():
"""专辑快照缺少远端 ID 时也不得退化为单曲识别。"""
subscribe = _subscribe(
name="叶惠美",
media_source=None,
media_id=None,
music_type=MUSIC_ENTITY_ALBUM,
total_tracks=11,
)
with patch("app.chain.subscribe.MediaChain") as media_chain:
restored = SubscribeChain._recognize_music_subscribe(subscribe)
assert restored.music_type == MUSIC_ENTITY_ALBUM
assert restored.total_tracks == 11
media_chain.assert_not_called()
def test_legacy_music_without_identity_uses_recording_recognition_boundary():
"""旧订阅缺少标准身份时只能恢复为单曲，不能消费全局搜索中的专辑或艺术家候选。"""
subscribe = _subscribe(
media_source=None,
media_id=None,
music_type=None,
)
recording = _music_info()
media_chain = Mock()
media_chain.recognize_media.return_value = recording
with patch("app.chain.subscribe.MediaChain", return_value=media_chain), \
patch("app.chain.subscribe.MusicChain.search") as mixed_search:
restored = SubscribeChain._recognize_music_subscribe(subscribe)
assert restored is recording
mixed_search.assert_not_called()
media_chain.recognize_media.assert_called_once()
call = media_chain.recognize_media.call_args
assert isinstance(call.kwargs["meta"], MetaMusic)
assert call.kwargs["mtype"] == MediaType.MUSIC
def test_album_subscription_finishes_only_after_confirmed_full_pack():
"""专辑与电视剧全集相同，必须确认整专覆盖；单曲仍在任一成功下载后完成。"""
album_subscribe = _subscribe(music_type=MUSIC_ENTITY_ALBUM, total_tracks=11)
album = MusicInfo(
music_type=MUSIC_ENTITY_ALBUM,
title="叶惠美",
album="叶惠美",
total_tracks=11,
)
assert SubscribeChain._is_music_download_complete(
album_subscribe,
album,
[Context(confirmed_full_coverage=False)],
) is False
assert SubscribeChain._is_music_download_complete(
album_subscribe,
album,
[Context(confirmed_full_coverage=True)],
) is True
assert SubscribeChain._is_music_download_complete(
_subscribe(),
_music_info(),
[Context()],
) is True
def test_subscribe_add_music_uses_unified_recognize_by_meta():
"""音乐订阅新增应走统一 recognize_by_meta，并把媒体身份落到 MetaMusic 上。"""
target = _music_info()

53
tests/test_music_transfer.py
View File

@@ -133,6 +133,59 @@ def test_restore_music_context_from_download_history():
assert restored_info.album == "Random Access Memories"
def test_restore_album_context_keeps_album_identity_and_track_specific_tags(tmp_path, monkeypatch):
"""整专整理应保留选中的专辑身份，同时使用每个文件自己的曲名、艺术家和曲序。"""
album = MusicInfo(
source="musicbrainz",
media_id="release-group-1",
music_type="album",
title="叶惠美",
artists=["周杰伦"],
album="叶惠美",
album_artist="周杰伦",
year=2003,
total_tracks=11,
)
meta = MusicChain.to_meta(album)
history = SimpleNamespace(note={
"music": {
"version": 1,
"meta": meta.to_dict(),
"media": album.to_dict(),
}
})
audio_file = tmp_path / "03. 晴天.flac"
audio_file.write_bytes(b"fake-flac")
from app.helper.audio import AudioMetadataHelper
monkeypatch.setattr(
AudioMetadataHelper,
"read",
lambda path: MetaMusic(
org_string=path.name,
title="晴天",
artists=["周杰伦"],
album="错误专辑",
album_artist="错误艺术家",
year=1999,
track_number=3,
total_tracks=99,
),
)
restored_meta, restored_info = TransferChain._restore_music_download_context(history, audio_file)
assert restored_meta.title == "晴天"
assert restored_meta.track_number == 3
assert restored_meta.album == "叶惠美"
assert restored_meta.album_artist == "周杰伦"
assert restored_meta.year == 2003
assert restored_meta.total_tracks == 11
assert restored_info.music_type == "album"
assert restored_info.media_id == "release-group-1"
def test_restore_music_context_uses_file_title_over_subscription_title(tmp_path, monkeypatch):
"""曲目标题应优先取当前文件自身的标签/文件名，而非沿用订阅时的单曲标题。"""
meta, info = _music_context()

76
tests/test_musicbrainz_module.py
View File

@@ -82,6 +82,82 @@ def test_search_music_normalizes_candidates(monkeypatch):
assert results[0].title == "晴天"
def test_search_music_interleaves_recordings_albums_and_artists(monkeypatch):
"""全局音乐搜索应交错返回三类实体，避免单曲结果挤掉整专和艺术家入口。"""
module = MusicBrainzModule()
requested = []
def fake_request(path, params=None):
"""按 MusicBrainz 实体路径返回可区分的搜索结果。"""
requested.append((path, params))
if path == "/recording":
return {
"recordings": [
{"id": "recording-1", "title": "晴天"},
{"id": "recording-2", "title": "轨迹"},
]
}
if path == "/release-group":
return {
"release-groups": [
{
"id": "album-1",
"title": "叶惠美",
"primary-type": "Album",
"artist-credit": [{"artist": {"id": "artist-1", "name": "周杰伦"}}],
},
{"id": "album-2", "title": "七里香", "primary-type": "Album"},
]
}
if path == "/artist":
return {
"artists": [
{"id": "artist-1", "name": "周杰伦", "type": "Person"},
{"id": "artist-2", "name": "Jay Chou", "type": "Person"},
]
}
return None
monkeypatch.setattr(module, "_request_json", fake_request)
results = module.search_music(
MetaMusic(title="晴天", artists=["周杰伦"]),
limit=5,
)
assert [item.music_type for item in results] == [
"recording",
"album",
"artist",
"recording",
"album",
]
assert results[1].album == "叶惠美"
assert results[2].title == "周杰伦"
assert results[2].artists == []
assert requested[1][1]["query"] == 'releasegroup:"晴天" AND artist:"周杰伦"'
assert requested[2][1]["query"] == 'artist:"周杰伦"'
def test_file_recognition_searches_recordings_only(monkeypatch):
"""本地音轨识别不得把同名专辑或艺术家候选当成 Recording。"""
module = MusicBrainzModule()
requested_paths = []
def fake_request(path, params=None):
"""记录文件识别实际访问的 MusicBrainz 实体。"""
requested_paths.append(path)
return {"recordings": [{"id": "recording-1", "title": "晴天"}]}
monkeypatch.setattr(module, "_request_json", fake_request)
result = module.recognize_media(meta=MetaMusic(title="晴天"))
assert result is not None
assert result.music_type == "recording"
assert requested_paths == ["/recording"]
def test_recognize_music_ignores_other_sources(monkeypatch):
"""MusicBrainz 模块不应处理其他元数据源的详情请求。"""
module = MusicBrainzModule()

112
tests/test_navidrome_module.py
View File

@@ -1,6 +1,11 @@
"""Navidrome 媒体服务器模块接入测试。"""
from unittest.mock import Mock
from app import schemas
from app.core.module import ModuleManager
from app.core.context import MusicInfo
from app.modules.navidrome import NavidromeModule
from app.modules.navidrome.navidrome import Navidrome
from app.schemas.types import MediaServerType, ModuleType
@@ -30,3 +35,110 @@ def test_navidrome_module_ignores_non_music_media():
mediainfo.type = MediaType.MOVIE
assert NavidromeModule().media_exists(mediainfo) is None
def test_navidrome_refresh_requests_incremental_scan(monkeypatch):
"""音乐入库完成后应通过 Subsonic startScan 触发 Navidrome 增量扫描。"""
client = object.__new__(Navidrome)
requested = []
def fake_call(method, **kwargs):
"""记录媒体库刷新使用的 Subsonic 方法和参数。"""
requested.append((method, kwargs))
return {"status": "ok"}
monkeypatch.setattr(client, "_call", fake_call)
assert client.refresh_root_library() is True
assert requested == [("startScan", {"fullScan": False})]
def test_navidrome_search_filters_exact_album_and_song(monkeypatch):
"""Navidrome 模糊搜索结果必须按实体名称和艺术家精确过滤。"""
client = object.__new__(Navidrome)
monkeypatch.setattr(
client,
"_call",
lambda *_args, **_kwargs: {
"searchResult3": {
"album": [
{"id": "wrong", "name": "叶惠美 演唱会", "artist": "周杰伦", "songCount": 12},
{"id": "album-1", "name": "叶惠美", "artist": "周杰伦", "songCount": 11},
],
"song": [
{"id": "song-wrong", "title": "晴天 Live", "artist": "周杰伦"},
{"id": "song-1", "title": "晴天", "artist": "周杰伦"},
],
}
},
)
albums = client.search_music(album="叶惠美", artist="周杰伦")
songs = client.search_music(title="晴天", artist="周杰伦")
assert [item.item_id for item in albums] == ["album-1"]
assert albums[0].note["song_count"] == 11
assert [item.item_id for item in songs] == ["song-1"]
assert songs[0].title == "晴天"
def test_navidrome_now_playing_uses_song_title_instead_of_album(monkeypatch):
"""正在播放接口返回单曲时，仪表盘标题必须显示曲名而不是所属专辑名。"""
client = object.__new__(Navidrome)
client._play_host = "https://music.example.com"
monkeypatch.setattr(
client,
"_call",
lambda *_args, **_kwargs: {
"nowPlaying": {
"entry": [
{
"id": "song-1",
"title": "晴天",
"album": "叶惠美",
"artist": "周杰伦",
"coverArt": "cover-1",
}
]
}
},
)
monkeypatch.setattr(client, "_album_cover", lambda _item: "cover-url")
items = client.get_resume()
assert len(items) == 1
assert items[0].item_id == "song-1"
assert items[0].title == "晴天"
assert items[0].subtitle == "周杰伦"
assert items[0].image == "cover-url"
def test_navidrome_album_exists_requires_complete_track_count(monkeypatch):
"""同名专辑曲目不足时不得把整专订阅判定为已完整入库。"""
module = NavidromeModule()
service = Mock()
service.get_iteminfo.return_value = schemas.MediaServerItem(
item_id="album-1",
title="叶惠美",
item_type="音乐",
note={"artist": "周杰伦", "song_count": 10},
)
service.search_music.return_value = []
monkeypatch.setattr(module, "get_instances", lambda: {"music": service})
album = MusicInfo(
source="musicbrainz",
media_id="release-group-1",
music_type="album",
title="叶惠美",
artists=["周杰伦"],
total_tracks=11,
)
assert module.media_exists(album, itemid="album-1") is None
service.get_iteminfo.return_value.note["song_count"] = 11
exists = module.media_exists(album, itemid="album-1")
assert exists is not None
assert exists.itemid == "album-1"

5
tests/test_subscribe_chain.py
View File

@@ -78,6 +78,7 @@ def _load_subscribe_chain_class():
context_module.Context = SimpleNamespace
context_module.MediaInfo = SimpleNamespace
context_module.MusicInfo = SimpleNamespace
context_module.MUSIC_ENTITY_ALBUM = "album"
event_module = ensure_module("app.core.event", types.ModuleType("app.core.event"))
@@ -165,6 +166,8 @@ def _load_subscribe_chain_class():
"custom_words",
"media_category",
"filter_groups",
"music_type",
"total_tracks",
}
def __init__(self, **kwargs):
@@ -299,6 +302,8 @@ def _load_subscribe_chain_class():
self.media_source = None
self.media_id = None
self.mediaid = None
self.music_type = None
self.total_tracks = None
self.episode_group = None
for key, value in kwargs.items():
setattr(self, key, value)

5
tests/test_subscribe_endpoint.py
View File

@@ -1129,6 +1129,8 @@ def test_create_subscribe_accepts_music_payload_with_empty_strings():
subscribe_in = Subscribe(
name="Random Access Memories",
type=MediaType.MUSIC.value,
music_type="album",
total_tracks=13,
tmdbid="",
season="",
total_episode="",
@@ -1154,4 +1156,5 @@ def test_create_subscribe_accepts_music_payload_with_empty_strings():
assert payload["total_episode"] == 0
assert payload["sites"] == []
assert payload["type"] == MediaType.MUSIC.value
assert payload["music_type"] == "album"
assert payload["total_tracks"] == 13

25
tests/test_subscribe_oper.py
View File

@@ -143,6 +143,31 @@ def test_music_subscribe_persists_numeric_year_as_string():
assert payload["year"] == "2025"
def test_music_album_subscription_persists_entity_and_track_count():
"""专辑订阅必须保存实体类型和总曲目数，供搜索校验与完成判定复用。"""
persisted = SimpleNamespace(id=94)
created = SimpleNamespace(create=MagicMock())
media = MusicInfo(
source="musicbrainz",
media_id="release-group-1",
music_type="album",
title="叶惠美",
album="叶惠美",
total_tracks=11,
)
with patch("app.db.subscribe_oper.Subscribe") as subscribe_model:
subscribe_model.exists.side_effect = [None, persisted]
subscribe_model.return_value = created
sid, _ = SubscribeOper(db=object()).add(mediainfo=media, season=None)
assert sid == 94
payload = subscribe_model.call_args.kwargs
assert payload["music_type"] == "album"
assert payload["total_tracks"] == 11
@pytest.mark.parametrize("episode_group", [None, "eg-1"])
def test_async_add_scopes_duplicate_lookup_by_episode_group(episode_group):
"""异步新增与同步路径使用相同的剧集组身份契约。"""