mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-28 19:06:36 +08:00
437 lines
15 KiB
Python
437 lines
15 KiB
Python
"""
|
||
TMDB 模块同步与异步公开能力的业务决策一致性测试。
|
||
"""
|
||
|
||
import asyncio
|
||
from typing import Optional
|
||
from unittest.mock import AsyncMock, Mock, call
|
||
|
||
import pytest
|
||
|
||
from app.domain.context import MediaInfo
|
||
from app.domain.meta.metabase import MetaBase
|
||
from app.modules.themoviedb import TheMovieDbModule
|
||
from app.modules.themoviedb.tmdbv3api.exceptions import TMDbConnectionError
|
||
from app.schemas.category import ClassificationEnrichmentRequest, ClassificationIdentityFacts
|
||
from app.schemas.types import MediaSource, MediaType
|
||
|
||
|
||
def _module_with_clients() -> TheMovieDbModule:
|
||
"""
|
||
构造不触发模块生命周期、只注入测试客户端的 TMDB 模块。
|
||
"""
|
||
module = object.__new__(TheMovieDbModule)
|
||
module.tmdb = Mock()
|
||
module.cache = Mock()
|
||
module.category = Mock()
|
||
return module
|
||
|
||
|
||
def _movie_info(tmdbid: int = 10) -> dict:
|
||
"""
|
||
构造可投影为 MediaInfo 的最小电影详情。
|
||
"""
|
||
return {
|
||
"id": tmdbid,
|
||
"media_type": MediaType.MOVIE,
|
||
"title": "测试电影",
|
||
"release_date": "2024-02-03",
|
||
"genres": [{"id": 18, "name": "剧情"}],
|
||
}
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("status_code", "expected"),
|
||
[
|
||
(200, (True, "")),
|
||
(503, (False, "无法连接 tmdb.example，错误码：503")),
|
||
(None, (False, "tmdb.example 网络连接失败")),
|
||
],
|
||
)
|
||
def test_tmdb_connection_check_reports_http_and_network_failures(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
status_code: Optional[int],
|
||
expected: tuple[bool, str],
|
||
) -> None:
|
||
"""
|
||
连接检查应区分认证成功、服务端错误和没有 HTTP 响应。
|
||
"""
|
||
response = Mock(status_code=status_code) if status_code is not None else None
|
||
request = Mock()
|
||
request.get_res.return_value = response
|
||
request_factory = Mock(return_value=request)
|
||
settings = {
|
||
"NORMAL_USER_AGENT": "test-agent",
|
||
"PROXY": None,
|
||
"TMDB_API_DOMAIN": "tmdb.example",
|
||
"TMDB_API_KEY": "test-key",
|
||
}
|
||
monkeypatch.setattr("app.modules.themoviedb.RequestUtils", request_factory)
|
||
monkeypatch.setattr("app.modules.themoviedb.get_runtime_setting", settings.__getitem__)
|
||
|
||
assert _module_with_clients().test() == expected
|
||
request_factory.assert_called_once_with(ua="test-agent", proxies=None)
|
||
request.get_res.assert_called_once_with(
|
||
"https://tmdb.example/3/movie/550?api_key=test-key"
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("tmdb_id", "expected_calls"),
|
||
[
|
||
("invalid-id", 0),
|
||
("129", 1),
|
||
],
|
||
)
|
||
def test_tmdb_enrichment_rejects_invalid_or_missing_details(
|
||
tmdb_id: str,
|
||
expected_calls: int,
|
||
) -> None:
|
||
"""
|
||
非数字身份或未取得详情时不能向分类规则提供推测的事实。
|
||
"""
|
||
module = _module_with_clients()
|
||
module.tmdb_info = Mock(return_value=None)
|
||
request = ClassificationEnrichmentRequest(
|
||
identity=ClassificationIdentityFacts(
|
||
media_source=MediaSource.Douban.value,
|
||
media_id="1291561",
|
||
),
|
||
media_type=MediaType.MOVIE.value,
|
||
missing_fields=["media.countries"],
|
||
external_ids={MediaSource.TMDB.value: tmdb_id},
|
||
policy_revision=3,
|
||
timeout_seconds=2.5,
|
||
)
|
||
|
||
assert module.get_media_classification_facts(request) is None
|
||
assert module.tmdb_info.call_count == expected_calls
|
||
if expected_calls:
|
||
module.tmdb_info.assert_called_once_with(tmdbid=129, mtype=MediaType.MOVIE)
|
||
|
||
|
||
def test_recognize_sync_async_share_identity_cache_and_result_decisions() -> None:
|
||
"""
|
||
显式 TMDB 身份在双入口应产生相同查询、缓存写入和媒体结果。
|
||
"""
|
||
module = _module_with_clients()
|
||
info = _movie_info()
|
||
module.tmdb.get_info.return_value = info
|
||
module.tmdb.async_get_info = AsyncMock(return_value=info)
|
||
module.category.get_movie_category.return_value = "剧情片"
|
||
sync_meta = MetaBase("测试电影 2024")
|
||
async_meta = MetaBase("测试电影 2024")
|
||
|
||
sync_result = module.recognize_media(
|
||
meta=sync_meta,
|
||
mtype=MediaType.MOVIE,
|
||
media_source=MediaSource.TMDB,
|
||
media_id="10",
|
||
cache=False,
|
||
)
|
||
async_result = asyncio.run(
|
||
module.async_recognize_media(
|
||
meta=async_meta,
|
||
mtype=MediaType.MOVIE,
|
||
media_source=MediaSource.TMDB,
|
||
media_id="10",
|
||
cache=False,
|
||
)
|
||
)
|
||
|
||
assert sync_result is not None
|
||
assert async_result is not None
|
||
assert (
|
||
sync_result.media_source,
|
||
sync_result.media_id,
|
||
sync_result.title,
|
||
sync_result.year,
|
||
sync_result.category,
|
||
sync_result.recognize_cache_hit,
|
||
) == (
|
||
async_result.media_source,
|
||
async_result.media_id,
|
||
async_result.title,
|
||
async_result.year,
|
||
async_result.category,
|
||
async_result.recognize_cache_hit,
|
||
)
|
||
module.tmdb.get_info.assert_called_once_with(
|
||
mtype=MediaType.MOVIE, tmdbid=10, raise_on_connection_error=True
|
||
)
|
||
module.tmdb.async_get_info.assert_awaited_once_with(
|
||
mtype=MediaType.MOVIE, tmdbid=10, raise_on_connection_error=True
|
||
)
|
||
assert sync_meta.media_source == async_meta.media_source == MediaSource.TMDB
|
||
assert sync_meta.media_id == async_meta.media_id == "10"
|
||
assert module.cache.update.call_args_list == [
|
||
call(sync_meta, info),
|
||
call(async_meta, info),
|
||
]
|
||
|
||
|
||
def test_recognize_sync_async_pass_episode_group_as_group_id() -> None:
|
||
"""
|
||
剧集组 ID 应按 TmdbApi 的 group_id 参数在双入口转发。
|
||
"""
|
||
module = _module_with_clients()
|
||
info = _movie_info()
|
||
group_id = "5ad0ec240e0a26303f00d84d"
|
||
module.tmdb.get_info.return_value = info
|
||
module.tmdb.async_get_info = AsyncMock(return_value=info)
|
||
module.tmdb.get_tv_group_seasons.return_value = []
|
||
module.tmdb.async_get_tv_group_seasons = AsyncMock(return_value=[])
|
||
module.category.get_movie_category.return_value = "剧情片"
|
||
|
||
module.recognize_media(
|
||
mtype=MediaType.MOVIE,
|
||
media_source=MediaSource.TMDB,
|
||
media_id="10",
|
||
episode_group=group_id,
|
||
cache=False,
|
||
)
|
||
asyncio.run(
|
||
module.async_recognize_media(
|
||
mtype=MediaType.MOVIE,
|
||
media_source=MediaSource.TMDB,
|
||
media_id="10",
|
||
episode_group=group_id,
|
||
cache=False,
|
||
)
|
||
)
|
||
|
||
module.tmdb.get_tv_group_seasons.assert_called_once_with(group_id=group_id)
|
||
module.tmdb.async_get_tv_group_seasons.assert_awaited_once_with(
|
||
group_id=group_id
|
||
)
|
||
|
||
|
||
def test_recognize_sync_async_share_negative_cache_short_circuit() -> None:
|
||
"""
|
||
负缓存命中时双入口都不得继续访问 TMDB 客户端。
|
||
"""
|
||
module = _module_with_clients()
|
||
module.tmdb.async_get_info = AsyncMock()
|
||
module.tmdb.async_match = AsyncMock()
|
||
module.cache.get.side_effect = [
|
||
{"title": None},
|
||
{"title": None},
|
||
]
|
||
sync_meta = MetaBase("负缓存电影 2026")
|
||
async_meta = MetaBase("负缓存电影 2026")
|
||
sync_meta.name = "负缓存电影"
|
||
async_meta.name = "负缓存电影"
|
||
|
||
sync_result = module.recognize_media(
|
||
meta=sync_meta, media_source=MediaSource.TMDB, cache=True
|
||
)
|
||
async_result = asyncio.run(
|
||
module.async_recognize_media(
|
||
meta=async_meta, media_source=MediaSource.TMDB, cache=True
|
||
)
|
||
)
|
||
|
||
assert sync_result is None
|
||
assert async_result is None
|
||
assert module.cache.get.call_args_list == [call(sync_meta), call(async_meta)]
|
||
module.tmdb.get_info.assert_not_called()
|
||
module.tmdb.async_get_info.assert_not_awaited()
|
||
module.tmdb.match.assert_not_called()
|
||
module.tmdb.async_match.assert_not_awaited()
|
||
module.cache.update.assert_not_called()
|
||
|
||
|
||
def test_recognize_sync_async_share_identity_connection_failure() -> None:
|
||
"""
|
||
显式身份查询连接失败时双入口都应失败且不得写入负缓存。
|
||
"""
|
||
module = _module_with_clients()
|
||
sync_lookup = Mock(side_effect=TMDbConnectionError("offline"))
|
||
async_lookup = AsyncMock(side_effect=TMDbConnectionError("offline"))
|
||
module._get_info_by_tmdbid = sync_lookup
|
||
module._async_get_info_by_tmdbid = async_lookup
|
||
sync_meta = MetaBase("连接失败电影 2026")
|
||
async_meta = MetaBase("连接失败电影 2026")
|
||
|
||
sync_result = module.recognize_media(
|
||
meta=sync_meta,
|
||
mtype=MediaType.MOVIE,
|
||
media_source=MediaSource.TMDB,
|
||
media_id="10",
|
||
cache=False,
|
||
)
|
||
async_result = asyncio.run(
|
||
module.async_recognize_media(
|
||
meta=async_meta,
|
||
mtype=MediaType.MOVIE,
|
||
media_source=MediaSource.TMDB,
|
||
media_id="10",
|
||
cache=False,
|
||
)
|
||
)
|
||
|
||
assert sync_result is None
|
||
assert async_result is None
|
||
sync_lookup.assert_called_once_with(
|
||
tmdbid=10, mtype=MediaType.MOVIE, meta=sync_meta
|
||
)
|
||
async_lookup.assert_awaited_once_with(
|
||
tmdbid=10, mtype=MediaType.MOVIE, meta=async_meta
|
||
)
|
||
module.cache.update.assert_not_called()
|
||
|
||
|
||
def test_name_match_sync_async_share_ordered_fallback_plan() -> None:
|
||
"""
|
||
电视剧名称识别的带季查询与无年份回退在双入口应完全同序。
|
||
"""
|
||
module = _module_with_clients()
|
||
meta = MetaBase("测试剧 S02 2024")
|
||
meta.name = "测试剧"
|
||
meta.type = MediaType.TV
|
||
meta.year = "2024"
|
||
meta.begin_season = 2
|
||
result = {"id": 20}
|
||
module.tmdb.match.side_effect = [None, result]
|
||
module.tmdb.async_match = AsyncMock(side_effect=[None, result])
|
||
group_seasons = [{"order": 2, "episodes": []}]
|
||
|
||
sync_result = module._search_by_name(meta.name, meta, group_seasons)
|
||
async_result = asyncio.run(
|
||
module._async_search_by_name(meta.name, meta, group_seasons)
|
||
)
|
||
|
||
assert sync_result == async_result == result
|
||
expected_calls = [
|
||
call(
|
||
name="测试剧",
|
||
year="2024",
|
||
mtype=MediaType.TV,
|
||
season_year="2024",
|
||
season_number=2,
|
||
group_seasons=group_seasons,
|
||
),
|
||
call(name="测试剧", mtype=MediaType.TV),
|
||
]
|
||
assert module.tmdb.match.call_args_list == expected_calls
|
||
assert module.tmdb.async_match.call_args_list == expected_calls
|
||
|
||
|
||
def test_media_search_sync_async_share_combination_and_sorting() -> None:
|
||
"""
|
||
未知类型且有年份时，双入口应合并电影和电视剧并按日期倒序。
|
||
"""
|
||
module = _module_with_clients()
|
||
meta = MetaBase("同名作品 2024")
|
||
meta.name = "同名作品"
|
||
meta.type = MediaType.UNKNOWN
|
||
meta.year = "2024"
|
||
movie = _movie_info(30)
|
||
tv = {
|
||
"id": 31,
|
||
"media_type": MediaType.TV,
|
||
"name": "同名作品",
|
||
"first_air_date": "2024-08-01",
|
||
}
|
||
module.tmdb.search_movies.return_value = [movie]
|
||
module.tmdb.search_tvs.return_value = [tv]
|
||
module.tmdb.async_search_movies = AsyncMock(return_value=[movie])
|
||
module.tmdb.async_search_tvs = AsyncMock(return_value=[tv])
|
||
|
||
sync_result = module.search_medias(meta, media_source=MediaSource.TMDB)
|
||
async_result = asyncio.run(
|
||
module.async_search_medias(meta, media_source=MediaSource.TMDB)
|
||
)
|
||
|
||
assert sync_result is not None
|
||
assert async_result is not None
|
||
assert [item.tmdb_id for item in sync_result] == [31, 30]
|
||
assert [item.tmdb_id for item in async_result] == [31, 30]
|
||
module.tmdb.search_movies.assert_called_once_with("同名作品", "2024")
|
||
module.tmdb.search_tvs.assert_called_once_with("同名作品", "2024")
|
||
module.tmdb.async_search_movies.assert_awaited_once_with("同名作品", "2024")
|
||
module.tmdb.async_search_tvs.assert_awaited_once_with("同名作品", "2024")
|
||
|
||
|
||
@pytest.mark.parametrize("media_source,recognize_source", [
|
||
("themoviedb", "themoviedb"),
|
||
("themoviedb", "douban"),
|
||
("douban", "themoviedb"),
|
||
])
|
||
@pytest.mark.parametrize("media_type", [MediaType.MOVIE, MediaType.TV])
|
||
def test_obtain_images_sync_async_share_query_and_mapping(
|
||
monkeypatch: pytest.MonkeyPatch, media_source: str, recognize_source: str, media_type: MediaType,
|
||
) -> None:
|
||
"""
|
||
图片补全双入口应选择同一接口、语言参数和最佳图片映射。
|
||
"""
|
||
module = _module_with_clients()
|
||
monkeypatch.setattr("app.modules.themoviedb.get_runtime_setting", lambda key: (
|
||
recognize_source if key == "RECOGNIZE_SOURCE" else lambda path: f"https://images.test{path}"
|
||
))
|
||
images = {
|
||
"posters": [{"file_path": "/poster.jpg", "vote_average": 8}],
|
||
"backdrops": [{"file_path": "/backdrop.jpg", "vote_average": 7}],
|
||
"logos": [{"file_path": "/logo.png", "vote_average": 6}],
|
||
}
|
||
module.tmdb.get_movie_images.return_value = images
|
||
module.tmdb.async_get_movie_images = AsyncMock(return_value=images)
|
||
module.tmdb.get_tv_images.return_value = images
|
||
module.tmdb.async_get_tv_images = AsyncMock(return_value=images)
|
||
sync_media = MediaInfo(tmdb_info=_movie_info(40))
|
||
async_media = MediaInfo(tmdb_info=_movie_info(40))
|
||
sync_media.original_language = "ja"
|
||
async_media.original_language = "ja"
|
||
for media in (sync_media, async_media):
|
||
media.media_source = media_source
|
||
media.type = media_type
|
||
|
||
sync_result = module.obtain_images(sync_media)
|
||
async_result = asyncio.run(module.async_obtain_images(async_media))
|
||
|
||
assert sync_result is sync_media
|
||
assert async_result is async_media
|
||
assert sync_media.poster_path == async_media.poster_path
|
||
assert sync_media.backdrop_path == async_media.backdrop_path
|
||
assert sync_media.logo_path == async_media.logo_path
|
||
assert sync_media.poster_path == "https://images.test/poster.jpg"
|
||
assert sync_media.backdrop_path == "https://images.test/backdrop.jpg"
|
||
assert sync_media.logo_path == "https://images.test/logo.png"
|
||
method = "get_movie_images" if media_type == MediaType.MOVIE else "get_tv_images"
|
||
getattr(module.tmdb, method).assert_called_once_with(40, original_language="ja")
|
||
getattr(module.tmdb, f"async_{method}").assert_awaited_once_with(40, original_language="ja")
|
||
|
||
|
||
@pytest.mark.parametrize("media_source,recognize_source,tmdb_id,complete_images", [
|
||
("douban", "douban", None, False),
|
||
("douban", "douban", 40, False),
|
||
("douban", "douban", None, True),
|
||
("douban", "themoviedb", None, False),
|
||
("themoviedb", "douban", None, False),
|
||
("themoviedb", "themoviedb", None, False),
|
||
("themoviedb", "themoviedb", 40, True),
|
||
])
|
||
def test_obtain_images_sync_async_skip_ineligible_media(
|
||
monkeypatch: pytest.MonkeyPatch, media_source: str, recognize_source: str,
|
||
tmdb_id: Optional[int], complete_images: bool,
|
||
) -> None:
|
||
"""
|
||
不适用来源、缺少 TMDB ID 或图片齐全时，双入口原样返回且不查询图片。
|
||
"""
|
||
monkeypatch.setattr("app.modules.themoviedb.get_runtime_setting", lambda _key: recognize_source)
|
||
module = _module_with_clients()
|
||
media = MediaInfo()
|
||
media.media_source = media_source
|
||
media.tmdb_id = tmdb_id
|
||
media.type = MediaType.TV
|
||
media.poster_path = "/existing-poster.jpg"
|
||
if complete_images:
|
||
media.logo_path = "/existing-logo.png"
|
||
media.backdrop_path = "/existing-backdrop.jpg"
|
||
original = media.to_dict()
|
||
|
||
assert module.obtain_images(media) is media
|
||
assert asyncio.run(module.async_obtain_images(media)) is media
|
||
assert media.to_dict() == original
|
||
assert module.tmdb.mock_calls == []
|