mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
602 lines
29 KiB
Python
602 lines
29 KiB
Python
"""
|
||
自动音乐来源顺序、隔离诊断和不可重置的请求预算。
|
||
"""
|
||
|
||
import asyncio
|
||
import shutil
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, Mock
|
||
|
||
import pytest
|
||
|
||
from app.application.music.observation import (
|
||
capture_music_recognition,
|
||
capture_music_source,
|
||
music_request_timeout,
|
||
report_music_recognition,
|
||
)
|
||
from app.application.music.recognition import async_recognize_music_sources, recognize_music_sources
|
||
from app.chain.media import MediaChain
|
||
from app.chain.media.cache import AlbumDirectoryCache
|
||
from app.chain.media.path import _music_path_plan
|
||
from app.domain.context import MusicAlbumInfo, MusicInfo
|
||
from app.domain.media import music_recognition_sources
|
||
from app.domain.meta.metamusic import MetaMusic
|
||
from app.modules.douban.apiv2 import DoubanApi
|
||
from app.modules.theaudiodb import TheAudioDbModule
|
||
from app.schemas.types import MediaSource, MediaType
|
||
|
||
|
||
def _accepts(info, source):
|
||
"""
|
||
测试入口采用实际的来源与主身份判据，不凭对象非空判定远端命中。
|
||
"""
|
||
return bool(info and info.media_source == source and info.media_id)
|
||
|
||
|
||
@pytest.mark.parametrize("configured,expected", [
|
||
("themoviedb", (MediaSource.MusicBrainz,)),
|
||
("theaudiodb,musicbrainz,theaudiodb,doubanmusic", (MediaSource.TheAudioDB, MediaSource.MusicBrainz, MediaSource.DoubanMusic)),
|
||
("", (MediaSource.MusicBrainz,)),
|
||
("invalid,musicbrainz,anilist", (MediaSource.MusicBrainz,)),
|
||
])
|
||
def test_configured_music_sources_keep_order_and_ignore_video_sources(configured, expected):
|
||
"""
|
||
混合来源设置不会把影视来源派发为音乐服务，默认设置仍保持MusicBrainz。
|
||
"""
|
||
assert music_recognition_sources(configured) == expected
|
||
|
||
|
||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||
def test_source_failure_does_not_poison_following_source(asynchronous):
|
||
"""
|
||
首源故障不阻断后源的实际HTTP额度，两次请求仍计入同一操作总量。
|
||
"""
|
||
calls = []
|
||
|
||
def recognize(source):
|
||
"""
|
||
第一来源记录服务故障，第二来源以自己的身份成功命中。
|
||
"""
|
||
calls.append(source)
|
||
assert music_request_timeout() is not None
|
||
if source == MediaSource.MusicBrainz:
|
||
report_music_recognition("service_error", "服务暂不可用")
|
||
return None
|
||
return MusicInfo(media_source=source, media_id="audio-db-track", title="Song")
|
||
|
||
sources = (MediaSource.MusicBrainz, MediaSource.TheAudioDB, MediaSource.DoubanMusic)
|
||
result, diagnostic = asyncio.run(async_recognize_music_sources(sources, AsyncMock(side_effect=recognize), _accepts)) if asynchronous else (
|
||
recognize_music_sources(sources, recognize, _accepts))
|
||
|
||
assert calls == list(sources[:2])
|
||
assert result.media_source == MediaSource.TheAudioDB
|
||
assert diagnostic["status"] == "matched" and diagnostic["requests"] == 2
|
||
assert [row["status"] for row in diagnostic["sources"]] == ["service_error", "matched"]
|
||
assert [row["requests"] for row in diagnostic["sources"]] == [1, 1]
|
||
|
||
|
||
@pytest.mark.parametrize("state", ["ambiguous", "conflict"])
|
||
def test_ambiguous_or_conflicting_source_stops_weaker_fallback(state):
|
||
"""
|
||
已知冲突不能因为另一个来源愿意返回首个同名结果而被掩盖。
|
||
"""
|
||
pending = MusicInfo(title="Song", raw_data={"recognition": {"status": state, "message": "需要确认"}})
|
||
recognize = Mock(side_effect=[pending, MusicInfo(media_source="theaudiodb", media_id="wrong", title="Song")])
|
||
|
||
result, diagnostic = recognize_music_sources((MediaSource.MusicBrainz, MediaSource.TheAudioDB), recognize, _accepts)
|
||
|
||
assert result is None and diagnostic["status"] == state
|
||
assert recognize.call_count == 1
|
||
|
||
|
||
def test_outer_budget_cannot_be_reset_by_changing_source():
|
||
"""
|
||
来源范围独立失败，但不能通过新建子范围恢复已用完的外层请求额度。
|
||
"""
|
||
with capture_music_recognition(seconds=5, request_limit=1) as total:
|
||
with capture_music_source("first") as first:
|
||
assert music_request_timeout() is not None
|
||
report_music_recognition("service_error", "失败")
|
||
with capture_music_source("second") as second:
|
||
assert music_request_timeout() is None
|
||
assert total.requests == first.requests == 1
|
||
assert second.requests == 0 and second.status == "budget_exhausted"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_concurrent_source_scopes_claim_shared_budget_atomically():
|
||
"""
|
||
并行继承同一观察范围时也只能消耗总额度，不能分别领取一份完整预算。
|
||
"""
|
||
async def claim(source):
|
||
"""
|
||
让两个来源交错执行后领取共享HTTP名额。
|
||
"""
|
||
with capture_music_source(source):
|
||
await asyncio.sleep(0)
|
||
return music_request_timeout()
|
||
|
||
with capture_music_recognition(seconds=5, request_limit=1) as total:
|
||
allowed = await asyncio.gather(claim("one"), claim("two"))
|
||
assert sum(value is not None for value in allowed) == total.requests == 1
|
||
|
||
|
||
def test_all_empty_sources_keep_service_error_and_fresh_operation_recovers():
|
||
"""
|
||
没有成功回退时保留故障语义，下一次独立操作不继承上次状态。
|
||
"""
|
||
def failed(source):
|
||
"""
|
||
只有第一个来源出现故障，后面的来源正常无匹配。
|
||
"""
|
||
if source == MediaSource.MusicBrainz:
|
||
raise OSError("offline")
|
||
return None
|
||
|
||
sources = (MediaSource.MusicBrainz, MediaSource.TheAudioDB)
|
||
result, diagnostic = recognize_music_sources(sources, failed, _accepts)
|
||
assert result is None and diagnostic["status"] == "service_error"
|
||
expected = MusicInfo(media_source="musicbrainz", media_id="recording", title="Song")
|
||
result, diagnostic = recognize_music_sources(sources, lambda _source: expected, _accepts)
|
||
assert result is expected and diagnostic["status"] == "matched"
|
||
|
||
|
||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||
def test_native_recognition_uses_configured_fallback_without_mutating_provider_result(monkeypatch, asynchronous):
|
||
"""
|
||
自动识别使用配置顺序；成功保留来源ID，附加诊断不会污染来源缓存对象。
|
||
"""
|
||
chain = object.__new__(MediaChain)
|
||
monkeypatch.setattr("app.chain.media.recognition.get_chain_runtime_config_snapshot",
|
||
lambda: SimpleNamespace(search_source="musicbrainz,theaudiodb,doubanmusic"))
|
||
expected = MusicInfo(media_source="theaudiodb", media_id="track-42", title="Song")
|
||
callback = AsyncMock(side_effect=[None, expected]) if asynchronous else Mock(side_effect=[None, expected])
|
||
monkeypatch.setattr(chain, "async_recognize_music_from_source" if asynchronous else "recognize_music_from_source", callback)
|
||
kwargs = {"meta": MetaMusic(title="Song", artists=["Artist"]), "mtype": MediaType.MUSIC}
|
||
|
||
result = asyncio.run(chain._async_run_native_media_recognize(kwargs, True)) if asynchronous else chain._run_native_media_recognize(kwargs, True)
|
||
|
||
assert result.media_source == MediaSource.TheAudioDB and result.media_id == "track-42"
|
||
assert [call.kwargs["media_source"] for call in callback.call_args_list] == [MediaSource.MusicBrainz, MediaSource.TheAudioDB]
|
||
assert all(call.kwargs.get("media_id") is None for call in callback.call_args_list)
|
||
assert result.raw_data["recognition"]["status"] == "matched"
|
||
assert expected.raw_data == {} and result is not expected
|
||
|
||
|
||
@pytest.mark.parametrize("explicit", [False, True])
|
||
def test_bound_or_explicit_identity_does_not_cross_sources(monkeypatch, explicit):
|
||
"""
|
||
已绑定身份经过真实请求规划后仍固定来源，不用其它来源解释同一个ID。
|
||
"""
|
||
chain = object.__new__(MediaChain)
|
||
monkeypatch.setattr("app.chain.media.recognition.get_chain_runtime_config_snapshot",
|
||
lambda: SimpleNamespace(search_source="theaudiodb,musicbrainz,doubanmusic"))
|
||
meta = MetaMusic(title="Song", media_source=None if explicit else "musicbrainz", media_id=None if explicit else "recording")
|
||
plan = chain._build_recognition_plan(meta=meta, mtype=MediaType.MUSIC,
|
||
media_source=MediaSource.MusicBrainz if explicit else None,
|
||
media_id="recording" if explicit else None, episode_group=None,
|
||
cache=True, share_meta=None, music_type="recording")
|
||
expected = MusicInfo(media_source="musicbrainz", media_id="recording", title="Song")
|
||
callback = Mock(return_value=expected)
|
||
monkeypatch.setattr(chain, "recognize_music_from_source", callback)
|
||
|
||
assert chain._run_native_media_recognize(plan.module_kwargs(), True) is expected
|
||
callback.assert_called_once()
|
||
assert callback.call_args.kwargs["media_source"] == MediaSource.MusicBrainz
|
||
assert callback.call_args.kwargs["media_id"] == "recording"
|
||
|
||
|
||
def test_unbound_track_with_explicit_release_evidence_stays_in_musicbrainz(monkeypatch):
|
||
"""
|
||
只有发行MBID时也不能将其映射到其它来源的同名专辑。
|
||
"""
|
||
chain = object.__new__(MediaChain)
|
||
monkeypatch.setattr("app.chain.media.recognition.get_chain_runtime_config_snapshot",
|
||
lambda: SimpleNamespace(search_source="theaudiodb,musicbrainz"))
|
||
callback = Mock(return_value=None)
|
||
monkeypatch.setattr(chain, "recognize_music_from_source", callback)
|
||
chain._run_native_media_recognize({"meta": MetaMusic(title="Song", musicbrainz_release_id="known-release")}, True)
|
||
callback.assert_called_once()
|
||
assert callback.call_args.kwargs["media_source"] == MediaSource.MusicBrainz
|
||
|
||
|
||
@pytest.mark.parametrize("state", ["ambiguous", "conflict", "service_error", "budget_exhausted"])
|
||
def test_path_tiers_preserve_unresolved_diagnostics(monkeypatch, state):
|
||
"""
|
||
无远端ID的诊断不能在版本校验中丢失，再用文件名启动一轮弱查询。
|
||
"""
|
||
chain = object.__new__(MediaChain)
|
||
meta = MetaMusic(title="Song")
|
||
pending = MusicInfo(title="Song", raw_data={"recognition": {"status": state}})
|
||
monkeypatch.setattr(chain, "recognize_media", Mock(return_value=pending))
|
||
assert chain._recognize_music_meta_tier(meta, None, "标签") is pending
|
||
plan = _music_path_plan(meta, MetaMusic(title="filename"), None)
|
||
next(plan)
|
||
plan.send(None)
|
||
with pytest.raises(StopIteration) as finished:
|
||
plan.send(pending)
|
||
assert finished.value.value is pending
|
||
|
||
|
||
def test_ambiguity_cannot_be_overwritten_by_plugin_or_shared_identity():
|
||
"""
|
||
原生已判定歧义时，真实插件规划和共享状态机都必须保留待确认结果。
|
||
"""
|
||
chain = object.__new__(MediaChain)
|
||
chain.eventmanager = Mock()
|
||
meta = MetaMusic(title="Song")
|
||
pending = MusicInfo(title="Song", raw_data={"recognition": {"status": "ambiguous"}})
|
||
assert chain._supplement_media_recognize(meta, MediaType.MUSIC, None, None, pending) is pending
|
||
chain.eventmanager.check.assert_not_called()
|
||
plan = chain._build_recognition_plan(meta=meta, mtype=MediaType.MUSIC, media_source=None, media_id=None,
|
||
episode_group=None, cache=True, share_meta=None, music_type="recording")
|
||
steps = chain._recognition_steps(plan, True)
|
||
next(steps)
|
||
steps.send(pending)
|
||
with pytest.raises(StopIteration) as finished:
|
||
steps.send(pending)
|
||
assert finished.value.value is pending
|
||
|
||
|
||
class _Response:
|
||
"""
|
||
模拟HTTP响应及明确的资源释放，不允许测试访问真实服务。
|
||
"""
|
||
|
||
def __init__(self, payload, status=200):
|
||
"""
|
||
保存响应内容及关闭状态。
|
||
"""
|
||
self.payload, self.status_code, self.closed = payload, status, False
|
||
self.content, self.headers, self.text = b"json", {}, ""
|
||
|
||
def json(self):
|
||
"""
|
||
返回已设定的JSON结构。
|
||
"""
|
||
return self.payload
|
||
|
||
def close(self):
|
||
"""
|
||
标记同步响应释放。
|
||
"""
|
||
self.closed = True
|
||
|
||
async def aclose(self):
|
||
"""
|
||
标记异步响应释放。
|
||
"""
|
||
self.closed = True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||
@pytest.mark.parametrize("source", ["theaudiodb", "doubanmusic"])
|
||
async def test_secondary_provider_counts_requests_and_recovers_uncached_failure(monkeypatch, source, asynchronous):
|
||
"""
|
||
次级来源真实HTTP边界也服从预算，故障后同一请求可恢复而不会命中长负缓存。
|
||
"""
|
||
failed, recovered = _Response({}, 503), _Response({"album": [{"id": "ok"}]})
|
||
network = (AsyncMock if asynchronous else Mock)(side_effect=[failed, recovered])
|
||
client = SimpleNamespace(get_res=network)
|
||
if source == "theaudiodb":
|
||
getattr(TheAudioDbModule._request_json, "cache_clear")()
|
||
monkeypatch.setattr("app.modules.theaudiodb.AsyncRequestUtils" if asynchronous else "app.modules.theaudiodb.RequestUtils",
|
||
lambda **_kwargs: client)
|
||
|
||
async def request(query):
|
||
"""
|
||
通过实际同步或异步请求入口验证共享缓存。
|
||
"""
|
||
return await TheAudioDbModule._async_request_json("searchalbum.php", {"a": query}) if asynchronous else (
|
||
TheAudioDbModule._request_json("searchalbum.php", {"a": query}))
|
||
else:
|
||
api = object.__new__(DoubanApi)
|
||
api._request = client
|
||
api._user_agents = ["test"]
|
||
monkeypatch.setattr(api, "_prepare_get_request", lambda url, **kwargs: (url, kwargs))
|
||
getattr(api._music_get, "cache_clear")()
|
||
monkeypatch.setattr("app.modules.douban.apiv2.AsyncRequestUtils", lambda **_kwargs: client)
|
||
|
||
async def request(query):
|
||
"""
|
||
音乐API请求仍经过实际解析和关闭逻辑。
|
||
"""
|
||
return await api.async_music_search(query) if asynchronous else api.music_search(query)
|
||
try:
|
||
with capture_music_recognition(seconds=5, request_limit=1) as first:
|
||
assert await request("same") is None
|
||
assert first.requests == 1 and first.status == "service_error" and failed.closed
|
||
with capture_music_recognition(seconds=5, request_limit=1) as second:
|
||
assert await request("same") == recovered.payload
|
||
assert await request("other") is None
|
||
assert await request("same") == recovered.payload
|
||
assert second.requests == 1 and second.status == "budget_exhausted"
|
||
assert network.call_count == 2 and recovered.closed
|
||
finally:
|
||
if source == "theaudiodb":
|
||
getattr(TheAudioDbModule._request_json, "cache_clear")()
|
||
else:
|
||
getattr(api._music_get, "cache_clear")()
|
||
|
||
|
||
def _directory_setup(tmp_path, monkeypatch, sources, *, known_album=True):
|
||
"""
|
||
使用真实无标签FLAC和独立媒体链测试整专来源选择，不注册全局单例。
|
||
"""
|
||
directory = tmp_path / ("Artist - Album" if known_album else "Music")
|
||
directory.mkdir()
|
||
for index, title in enumerate(("First", "Second"), 1):
|
||
name = f"{index:02} - {title}.flac" if known_album else f"{index:02} - Artist - {title}.flac"
|
||
shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", directory / name)
|
||
config = {"sources": sources}
|
||
monkeypatch.setattr("app.chain.media.album.get_chain_runtime_config_snapshot", lambda: SimpleNamespace(
|
||
search_source=config["sources"], audio_extensions=(".flac",), music_cue_enable=True,
|
||
music_release_region_priority=("CN",), music_release_script_priority=("Hans",)))
|
||
chain = object.__new__(MediaChain)
|
||
chain._album_dir_cache = AlbumDirectoryCache(8)
|
||
monkeypatch.setattr(chain, "_finalize_recognition_result", lambda info, **_kwargs: info)
|
||
monkeypatch.setattr(chain, "_async_finalize_recognition_result", AsyncMock(side_effect=lambda info, **_kwargs: info))
|
||
primary = SimpleNamespace(match_music_album=Mock(return_value=None), async_match_music_album=AsyncMock(return_value=None))
|
||
monkeypatch.setattr("app.chain.media.album.MusicBrainzChain", lambda: primary)
|
||
return directory, chain, config, primary
|
||
|
||
|
||
def _album(source, identity="album-one", *, tracks=True):
|
||
"""
|
||
构造同一来源命名空间内的完整专辑和原生曲目身份。
|
||
"""
|
||
album = MusicAlbumInfo(media_source=source, media_id=identity, title="Album", artists=["Artist"])
|
||
if tracks:
|
||
album.tracks = [MusicInfo(media_source=source, media_id=f"{identity}-track-{index}", title=title,
|
||
artists=["Artist"], album="Album", album_id=identity,
|
||
track_number=index, disc_number=1, duration=3)
|
||
for index, title in enumerate(("First", "Second"), 1)]
|
||
return album
|
||
|
||
|
||
def _backend(albums):
|
||
"""
|
||
模拟来源链的读边界，真正的搜索计划、候选核验和文件对位继续运行。
|
||
"""
|
||
by_id = {album.media_id: album for album in albums}
|
||
cards = [album.to_music_info() for album in albums]
|
||
return SimpleNamespace(search_music=Mock(return_value=cards), async_search_music=AsyncMock(return_value=cards),
|
||
get_music_album=Mock(side_effect=by_id.get), async_get_music_album=AsyncMock(side_effect=by_id.get))
|
||
|
||
|
||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||
def test_directory_fallback_requires_complete_tracks_and_keeps_source_identity(tmp_path, monkeypatch, asynchronous):
|
||
"""
|
||
首源故障时可按完整次级曲目表整理；匹配身份始终属于该次级来源。
|
||
"""
|
||
directory, chain, _config, primary = _directory_setup(tmp_path, monkeypatch, "musicbrainz,theaudiodb")
|
||
|
||
def fail_primary(*_args, **_kwargs):
|
||
"""
|
||
复现主元数据源短暂不可用。
|
||
"""
|
||
report_music_recognition("service_error", "临时不可用")
|
||
return None
|
||
|
||
primary.match_music_album.side_effect = fail_primary
|
||
primary.async_match_music_album.side_effect = fail_primary
|
||
backend = _backend([_album(MediaSource.TheAudioDB)])
|
||
monkeypatch.setattr(chain, "_music_source_chain", lambda _source: backend)
|
||
|
||
result = asyncio.run(chain.async_recognize_music_album_directory(directory)) if asynchronous else chain.recognize_music_album_directory(directory)
|
||
|
||
assert len(result) == 2 and result.recognition["status"] == "matched"
|
||
assert [row["status"] for row in result.recognition["sources"]] == ["service_error", "matched"]
|
||
assert {info.media_source for info in result.values()} == {MediaSource.TheAudioDB}
|
||
assert {info.media_id for info in result.values()} == {"album-one-track-1", "album-one-track-2"}
|
||
assert all(info.musicbrainz_release_id is None and info.raw_data["recognition"]["release_verified"] is False for info in result.values())
|
||
|
||
|
||
@pytest.mark.parametrize("case", ["no_tracks", "ambiguous", "wrong_source", "wrong_duration"])
|
||
def test_directory_does_not_invent_alignment_or_pick_first_album(tmp_path, monkeypatch, case):
|
||
"""
|
||
没有曲目、等价多候选、混源ID及明显时长冲突都不能产生自动文件映射。
|
||
"""
|
||
directory, chain, _config, _primary = _directory_setup(tmp_path, monkeypatch, "theaudiodb")
|
||
album = _album(MediaSource.TheAudioDB, tracks=case != "no_tracks")
|
||
albums = [album, _album(MediaSource.TheAudioDB, "other")] if case == "ambiguous" else [album]
|
||
backend = _backend(albums)
|
||
if case == "wrong_source":
|
||
backend.get_music_album.side_effect = lambda _id: _album(MediaSource.DoubanMusic)
|
||
if case == "wrong_duration":
|
||
album.tracks[0].duration = 200
|
||
monkeypatch.setattr(chain, "_music_source_chain", lambda _source: backend)
|
||
|
||
result = chain.recognize_music_album_directory(directory)
|
||
|
||
assert not result
|
||
assert result.recognition["status"] == ("ambiguous" if case == "ambiguous" else "not_found")
|
||
|
||
|
||
def test_source_configuration_change_invalidates_directory_negative_cache(tmp_path, monkeypatch):
|
||
"""
|
||
切换来源后必须重新匹配，不能复用同一目录在旧来源中的无匹配结果。
|
||
"""
|
||
directory, chain, config, primary = _directory_setup(tmp_path, monkeypatch, "musicbrainz")
|
||
backend = _backend([_album(MediaSource.TheAudioDB)])
|
||
monkeypatch.setattr(chain, "_music_source_chain", lambda _source: backend)
|
||
assert not chain.recognize_music_album_directory(directory)
|
||
config["sources"] = "theaudiodb"
|
||
|
||
result = chain.recognize_music_album_directory(directory)
|
||
|
||
assert len(result) == 2
|
||
primary.match_music_album.assert_called_once()
|
||
backend.search_music.assert_called_once()
|
||
|
||
|
||
def test_directory_lookup_freezes_source_choice_with_cache_key(tmp_path, monkeypatch):
|
||
"""
|
||
配置在缓存键生成后改变，也不能将新来源的结果存入旧来源键。
|
||
"""
|
||
directory, chain, config, _primary = _directory_setup(tmp_path, monkeypatch, "theaudiodb")
|
||
backends = {source: _backend([_album(source)]) for source in (MediaSource.TheAudioDB, MediaSource.DoubanMusic)}
|
||
monkeypatch.setattr(chain, "_music_source_chain", backends.get)
|
||
original = chain._album_dir_cache.resolve
|
||
|
||
def change_config(key, signature, resolver):
|
||
"""
|
||
在真实缓存准备加载时切换下一次请求的来源。
|
||
"""
|
||
config["sources"] = "doubanmusic"
|
||
return original(key, signature, resolver)
|
||
|
||
monkeypatch.setattr(chain._album_dir_cache, "resolve", change_config)
|
||
first = chain.recognize_music_album_directory(directory)
|
||
second = chain.recognize_music_album_directory(directory)
|
||
|
||
assert {info.media_source for info in first.values()} == {MediaSource.TheAudioDB}
|
||
assert {info.media_source for info in second.values()} == {MediaSource.DoubanMusic}
|
||
|
||
|
||
def test_unknown_album_can_be_discovered_from_two_track_names(tmp_path, monkeypatch):
|
||
"""
|
||
专辑目录名无意义时，使用真实曲名返回的同来源album_id汇总候选再完整核验。
|
||
"""
|
||
directory, chain, _config, _primary = _directory_setup(tmp_path, monkeypatch, "theaudiodb", known_album=False)
|
||
album = _album(MediaSource.TheAudioDB)
|
||
backend = _backend([album])
|
||
backend.search_music.side_effect = lambda meta, **_kwargs: [track for track in album.tracks if track.title == meta.title]
|
||
monkeypatch.setattr(chain, "_music_source_chain", lambda _source: backend)
|
||
|
||
result = chain.recognize_music_album_directory(directory)
|
||
|
||
assert len(result) == 2
|
||
assert backend.search_music.call_count == 2
|
||
backend.get_music_album.assert_called_once_with("album-one")
|
||
|
||
|
||
def test_secondary_recording_candidates_are_not_selected_by_response_order():
|
||
"""
|
||
同名同艺人的不同原生ID没有额外证据时必须保持歧义。
|
||
"""
|
||
candidates = [MusicInfo(media_source="theaudiodb", media_id=identity, title="Song", artists=["Artist"])
|
||
for identity in ("one", "two")]
|
||
with capture_music_recognition() as observed:
|
||
assert TheAudioDbModule._select_track(MetaMusic(title="Song", artists=["Artist"]), candidates) is None
|
||
assert observed.status == "ambiguous" and len(observed.candidates) == 2
|
||
|
||
|
||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||
def test_secondary_catalog_preserves_local_multidisc_edition(tmp_path, monkeypatch, asynchronous):
|
||
"""
|
||
真实标签的再版年及双碟曲序不能被次级目录的单碟首版布局覆盖。
|
||
"""
|
||
from mutagen.flac import FLAC
|
||
|
||
directory, chain, _config, _primary = _directory_setup(tmp_path, monkeypatch, "theaudiodb")
|
||
files = sorted(directory.glob("*.flac"))
|
||
for disc, path in enumerate(files, 1):
|
||
tags = FLAC(path)
|
||
tags.update(title=["First" if disc == 1 else "Second"], album=["Album"], artist=["Unknown Artist"],
|
||
albumartist=["Artist"], discnumber=[str(disc)], disctotal=["2"], tracknumber=["1"],
|
||
tracktotal=["1"], date=["2020-09-01"], originaldate=["2001"])
|
||
tags.save()
|
||
album = _album(MediaSource.TheAudioDB)
|
||
album.release_date = "2001"
|
||
for track in album.tracks:
|
||
track.year, track.release_year, track.original_year = 2001, None, 2001
|
||
track.total_tracks, track.total_discs = 2, 1
|
||
backend = _backend([album])
|
||
monkeypatch.setattr(chain, "_music_source_chain", lambda _source: backend)
|
||
|
||
result = asyncio.run(chain.async_recognize_music_album_directory(directory)) if asynchronous else chain.recognize_music_album_directory(directory)
|
||
|
||
assert len(result) == 2
|
||
for disc, path in enumerate(files, 1):
|
||
info = result[str(path.resolve())]
|
||
assert (info.disc_number, info.track_number, info.total_tracks, info.total_discs) == (disc, 1, 1, 2)
|
||
assert (info.year, info.release_year, info.original_year) == (2020, 2020, 2001)
|
||
assert info.field_sources["release_year"] == "tag"
|
||
assert info.media_id == f"album-one-track-{disc}"
|
||
assert info.musicbrainz_release_id is None
|
||
assert [track.track_number for track in album.tracks] == [1, 2]
|
||
assert [track.year for track in album.tracks] == [2001, 2001]
|
||
|
||
|
||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||
@pytest.mark.parametrize("music_type", ["album", "recording"])
|
||
def test_douban_distinct_matching_identities_require_confirmation(asynchronous, music_type):
|
||
"""
|
||
同步和异步豆瓣搜索遇到同名同署名的不同身份时都不能取第一个。
|
||
"""
|
||
from app.modules.douban import DoubanModule
|
||
|
||
module = object.__new__(DoubanModule)
|
||
cards = [{"id": identity, "title": "Album", "artists": [{"name": "Artist"}]} for identity in ("100", "200")]
|
||
|
||
def detail(subject_id):
|
||
"""
|
||
提供同证据的两个专辑曲目表。
|
||
"""
|
||
return {"id": subject_id, "title": "Album", "singer": [{"name": "Artist"}],
|
||
"songs": [{"title": "First", "track_number": 1}]}
|
||
|
||
module.doubanapi = SimpleNamespace(music_search=Mock(return_value={"items": cards}),
|
||
async_music_search=AsyncMock(return_value={"items": cards}),
|
||
music_detail=Mock(side_effect=detail), async_music_detail=AsyncMock(side_effect=detail))
|
||
meta = MetaMusic(title="First", album="Album", artists=["Artist"], track_number=1, music_type=music_type)
|
||
with capture_music_recognition() as observation:
|
||
result = asyncio.run(module._async_recognize_music_media(meta, MediaSource.DoubanMusic, None, music_type)) if asynchronous else (
|
||
module._recognize_music_media(meta, MediaSource.DoubanMusic, None, music_type))
|
||
assert result is None and observation.status == "ambiguous"
|
||
assert len(observation.candidates) == 2
|
||
if music_type == "album":
|
||
module.doubanapi.music_detail.assert_not_called()
|
||
module.doubanapi.async_music_detail.assert_not_called()
|
||
|
||
|
||
@pytest.mark.parametrize("state", ["budget_exhausted", "service_error", "ambiguous", "conflict"])
|
||
def test_source_cache_cannot_turn_failed_parent_into_success(state):
|
||
"""
|
||
外层已中断的识别不能因后续来源命中缓存而返回成功身份。
|
||
"""
|
||
with capture_music_recognition() as observation:
|
||
report_music_recognition(state, "外层已中断")
|
||
result, diagnostic = recognize_music_sources((MediaSource.TheAudioDB,), lambda source: MusicInfo(
|
||
media_source=source, media_id="cached", title="First"), _accepts)
|
||
assert result is None
|
||
assert diagnostic["status"] == observation.status == state
|
||
|
||
|
||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||
def test_directory_does_not_confirm_truncated_candidates(tmp_path, monkeypatch, asynchronous):
|
||
"""
|
||
超过有界详情核验范围时保留歧义，不能仅凭前五个候选宣布唯一。
|
||
"""
|
||
directory, chain, _config, _primary = _directory_setup(tmp_path, monkeypatch, "theaudiodb")
|
||
backend = _backend([_album(MediaSource.TheAudioDB, str(index + 10)) for index in range(6)])
|
||
monkeypatch.setattr(chain, "_music_source_chain", lambda _source: backend)
|
||
result = asyncio.run(chain.async_recognize_music_album_directory(directory)) if asynchronous else chain.recognize_music_album_directory(directory)
|
||
assert not result and result.recognition["status"] == "ambiguous"
|
||
backend.get_music_album.assert_not_called()
|
||
backend.async_get_music_album.assert_not_called()
|
||
|
||
|
||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||
def test_empty_directory_diagnostics_survive_single_file_fallback(tmp_path, monkeypatch, asynchronous):
|
||
"""
|
||
空目录映射仍须把待确认原因送到单文件识别，并保留本地展示信息。
|
||
"""
|
||
from app.chain.media.path import _finalize_music_path_info
|
||
from app.domain.music import MusicDirectoryMatch
|
||
|
||
path = tmp_path / "First.flac"
|
||
path.write_bytes(b"test")
|
||
chain = object.__new__(MediaChain)
|
||
pending = MusicDirectoryMatch(recognition={"status": "ambiguous", "message": "多个专辑"})
|
||
monkeypatch.setattr(chain, "recognize_music_album_directory", Mock(return_value=pending))
|
||
monkeypatch.setattr(chain, "async_recognize_music_album_directory", AsyncMock(return_value=pending))
|
||
result = asyncio.run(chain._async_music_album_dir_fallback(path)) if asynchronous else chain._music_album_dir_fallback(path)
|
||
result = _finalize_music_path_info(MetaMusic(title="First", artists=["Artist"]), result)
|
||
assert result.title == "First" and result.artists == ["Artist"]
|
||
assert result.media_id is None and result.raw_data["recognition"] == pending.recognition
|