mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-01 21:07:01 +08:00
fix(scraper): align built-in NFO results with text contract (#7049)
* fix(scraper): return NFO text from built-in providers * test: cover type migration boundaries * fix: silence dynamic import lint diagnostics * test: keep coverage policy stable for nfo repair * fix: preserve merged scraper documentation during conflict repair --------- Co-authored-by: Derrors <derrors@local> Co-authored-by: jxxghp <jxxghp@gmail.com>
This commit is contained in:
@@ -140,14 +140,14 @@ class MediaScraperHelper:
|
||||
mediainfo: MediaInfo,
|
||||
season: Optional[int] = None,
|
||||
episode: Optional[int] = None,
|
||||
) -> Optional[bytes]:
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
根据统一媒体信息生成电影、剧集、季或单集 NFO。
|
||||
|
||||
:param mediainfo: 统一媒体信息
|
||||
:param season: 季号
|
||||
:param episode: 集号
|
||||
:return: UTF-8 编码的 NFO XML 字节
|
||||
:return: UTF-8 编码声明的 NFO XML 文本;媒体信息为空时返回 None
|
||||
"""
|
||||
if not mediainfo:
|
||||
return None
|
||||
@@ -177,7 +177,7 @@ class MediaScraperHelper:
|
||||
DomUtils.add_node(doc, root, "season", "-1")
|
||||
DomUtils.add_node(doc, root, "episode", "-1")
|
||||
|
||||
return doc.toprettyxml(indent=" ", encoding="utf-8")
|
||||
return doc.toprettyxml(indent=" ", encoding="utf-8").decode("utf-8")
|
||||
|
||||
@classmethod
|
||||
def get_metadata_img(
|
||||
|
||||
@@ -697,7 +697,7 @@ class AniListModule(MediaAuxiliaryProviderMixin, _ModuleBase):
|
||||
season: Optional[int] = None,
|
||||
episode: Optional[int] = None,
|
||||
**kwargs,
|
||||
) -> Optional[bytes]:
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
为选择 AniList 刮削来源的媒体生成 NFO。
|
||||
|
||||
@@ -705,7 +705,7 @@ class AniListModule(MediaAuxiliaryProviderMixin, _ModuleBase):
|
||||
:param season: 可选季号,省略时生成媒体级内容
|
||||
:param episode: 可选集号,与季号共同决定内容范围
|
||||
:param kwargs: 统一刮削接口的额外参数,当前不读取
|
||||
:return: UTF-8 XML 字节;刮削来源不符或无法生成时为 None
|
||||
:return: 包含 UTF-8 编码声明的 XML 文本;刮削来源不符或无法生成时为 None
|
||||
"""
|
||||
scrape_source = mediainfo.scrape_source or get_runtime_setting('SCRAP_SOURCE')
|
||||
if scrape_source != "anilist":
|
||||
|
||||
@@ -456,7 +456,7 @@ class BangumiModule(MediaAuxiliaryProviderMixin, _ModuleBase):
|
||||
season: Optional[int] = None,
|
||||
episode: Optional[int] = None,
|
||||
**kwargs,
|
||||
) -> Optional[bytes]:
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
为选择 Bangumi 刮削来源的媒体生成 NFO 内容。
|
||||
|
||||
@@ -464,7 +464,7 @@ class BangumiModule(MediaAuxiliaryProviderMixin, _ModuleBase):
|
||||
:param season: 目标季号,省略时生成媒体级元数据
|
||||
:param episode: 目标集号,省略时由季号决定元数据范围
|
||||
:param kwargs: 统一刮削接口的额外参数,当前不使用
|
||||
:return: UTF-8 XML 字节;刮削来源不是 Bangumi 或无法生成时为 None
|
||||
:return: 包含 UTF-8 编码声明的 XML 文本;刮削来源不是 Bangumi 或无法生成时为 None
|
||||
"""
|
||||
scrape_source = mediainfo.scrape_source or get_runtime_setting('SCRAP_SOURCE')
|
||||
if scrape_source != "bangumi":
|
||||
|
||||
@@ -1875,14 +1875,14 @@ class DoubanModule(MediaAuxiliaryProviderMixin, _ModuleBase):
|
||||
count=count)
|
||||
return self._project_collection(infos)
|
||||
|
||||
def metadata_nfo(self, mediainfo: MediaInfo, season: int = None, **kwargs) -> Optional[bytes]:
|
||||
def metadata_nfo(self, mediainfo: MediaInfo, season: Optional[int] = None, **kwargs) -> Optional[str]:
|
||||
"""
|
||||
获取 UTF-8 编码的 NFO XML 字节。
|
||||
获取包含 UTF-8 编码声明的 NFO XML 文本。
|
||||
|
||||
:param mediainfo: 已识别媒体信息,可带自定义刮削来源
|
||||
:param season: 目标季号;None 表示整剧,0 可表示特别篇
|
||||
:param kwargs: 兼容统一模块入口的扩展参数;按识别入口需要转交,刮削入口当前忽略
|
||||
:return: UTF-8 编码的 NFO XML 字节;未选择豆瓣刮削时返回 None
|
||||
:return: 包含 UTF-8 编码声明的 NFO XML 文本;未选择豆瓣刮削时返回 None
|
||||
"""
|
||||
if (mediainfo.scrape_source or get_runtime_setting('SCRAP_SOURCE')) != "douban":
|
||||
return None
|
||||
|
||||
@@ -19,13 +19,13 @@ class DoubanScraper:
|
||||
_force_nfo = False
|
||||
_force_img = False
|
||||
|
||||
def get_metadata_nfo(self, mediainfo: MediaInfo, season: Optional[int] = None) -> Optional[bytes]:
|
||||
def get_metadata_nfo(self, mediainfo: MediaInfo, season: Optional[int] = None) -> Optional[str]:
|
||||
"""
|
||||
生成电影、整剧或季级别的 NFO XML。
|
||||
|
||||
:param mediainfo: 包含豆瓣简介、演职员和评分的媒体对象
|
||||
:param season: 季号;非电影且提供季号时生成季文档
|
||||
:return: UTF-8 编码的 XML 字节;无法生成时返回 None
|
||||
:return: 包含 UTF-8 编码声明的 XML 文本;无法生成时返回 None
|
||||
"""
|
||||
if mediainfo.type == MediaType.MOVIE:
|
||||
doc = self.__gen_movie_nfo_file(mediainfo=mediainfo)
|
||||
@@ -35,7 +35,7 @@ class DoubanScraper:
|
||||
else:
|
||||
doc = self.__gen_tv_nfo_file(mediainfo=mediainfo)
|
||||
if doc:
|
||||
return doc.toprettyxml(indent=" ", encoding="utf-8")
|
||||
return doc.toprettyxml(indent=" ", encoding="utf-8").decode("utf-8")
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@@ -846,7 +846,7 @@ class ImdbModule(MediaAuxiliaryProviderMixin, _ModuleBase):
|
||||
season: Optional[int] = None,
|
||||
episode: Optional[int] = None,
|
||||
**kwargs,
|
||||
) -> Optional[bytes]:
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
为选择 IMDb 刮削来源的媒体生成 NFO。
|
||||
|
||||
@@ -854,7 +854,7 @@ class ImdbModule(MediaAuxiliaryProviderMixin, _ModuleBase):
|
||||
:param season: 可选季号
|
||||
:param episode: 可选集号
|
||||
:param kwargs: 统一刮削接口的额外参数,当前不读取
|
||||
:return: UTF-8 XML 字节;来源不符、刮削器未初始化或无法生成时为 None
|
||||
:return: 包含 UTF-8 编码声明的 XML 文本;来源不符、刮削器未初始化或无法生成时为 None
|
||||
"""
|
||||
del kwargs
|
||||
if (mediainfo.scrape_source or get_runtime_setting('SCRAP_SOURCE')) != MediaSource.IMDb.value:
|
||||
|
||||
@@ -61,7 +61,7 @@ class TmdbScraper:
|
||||
:param mediainfo: 包含 TMDB 标识、媒体类型及可选分集组的媒体信息
|
||||
:param season: 季号;None 时生成电影或整剧文档
|
||||
:param episode: 输出文档的单集编号;非零且提供季号时生成单集文档
|
||||
:return: UTF-8 编码的 XML 字节;未生成文档时返回 None
|
||||
:return: 包含 UTF-8 编码声明的 XML 文本;未生成文档时返回 None
|
||||
"""
|
||||
if mediainfo.type == MediaType.MOVIE:
|
||||
doc = self.__gen_movie_nfo_file(mediainfo=mediainfo)
|
||||
@@ -92,7 +92,7 @@ class TmdbScraper:
|
||||
else:
|
||||
doc = self.__gen_tv_nfo_file(mediainfo=mediainfo)
|
||||
if doc:
|
||||
return doc.toprettyxml(indent=" ", encoding="utf-8")
|
||||
return doc.toprettyxml(indent=" ", encoding="utf-8").decode("utf-8")
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@@ -108,7 +108,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
|
||||
| Event Contract | 53 | 均已有 payload model,但当前全部是 diagnostic enforcement |
|
||||
| Python 源码量 | 373,625 行 | 本次重新扫描物理行数,排除 `app/plugins/**`;74 个文件超过 1,000 行,12 个超过 2,000 行;源码尺寸仅作观察 |
|
||||
| 长方法 | 309 个超过 80 行 | 本次 AST 扫描排除 `app/plugins/**`;61 个超过 150 行,17 个超过 250 行;新增函数仍受 C901/嵌套深度门禁约束 |
|
||||
| 全量 mypy 历史债务 | 4,445 / 252 文件 | 本分支为 Pydantic 字段补充描述并保留默认值,减少 3 项 assignment 错误;本轮在消息、站点、诊断、备份、用户配置、订阅、AcoustID、TMDB、仪表板和命令生命周期边界补齐类型,历史债务再减少 53 项;TMDB v3 API 显式导出并保留集合 parts 的列表合同,逐文件逐错误码无增长;此前工作流筛选、动态插件路由和 stdio 流补齐边界类型,历史债务减少 3 项;Bitpt 与 CookieCloud 补齐边界类型,历史债务减少 16 项;知行站点解析器补齐数值属性与基类合同,历史债务减少 18 项;通知排版补齐 Slack 消息块类型,历史债务减少 2 项;工作流资源过滤改用领域对象调用过滤链,恢复的节点输出在消费前归一化,并收紧可空资源列表的类型低水位;消息媒体交互类型债务已清零,canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降;微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后,类型错误减少 4 项;站点解析器分享率改用浮点初值,类型错误减少 20 项;本次下载与文件安全修复再减少 3 项;Transmission 进度回退值归一化再减少 2 项;本批清理 Servarr、认证、存储、订阅策略和搜索/规则/通知 Schema 的 18 项类型错误,并修复系统配置中的 Python 2 异常语法;本批再清理订阅元数据、订阅写入、交互宿主、整理路径、删除统计和工作流动作的 14 项类型错误,逐文件逐错误码无增长;本批补齐 OCR 请求参数、工具 Schema 摘要和宿主模块配置快照的 5 项边界类型,逐文件逐错误码无增长;本批为 Rousi 用户消息分页解析补齐返回类型与数据结构兜底,减少 2 项类型错误;本批优先处理 TMDB v3 电影与电视剧对象模型,补齐 98 个同步/异步方法注解并减少 122 项历史类型错误;随后补齐剧集、季度、人物、搜索和趋势对象的 142 个方法注解,再减少 142 项类型错误;绿联 API 响应边界补齐具体泛型与运行时字典收窄,减少 31 项类型错误;运行时限流器补齐状态方法、装饰器边界和时间队列类型,减少 31 项类型错误;QQ Bot API 响应边界补齐具体泛型与 JSON 形状校验,减少 17 项类型错误;rTorrent 客户端清理 182 项错误,调用端同步明确字典合同,累计再减少 190 项错误;豆瓣 API、中心服务接口及订阅消费端补齐返回边界和参数收窄,结合 Context 方法签名与缓存代理类型清理,累计再减少 455 项错误;共享 HTTP 客户端清理 110 项错误,TMDB API 清理 84 项错误,索引器爬虫清理 87 项错误,并补充 U115、AliPan、OpenList 对可空流响应的防守;U115 其余存储基类契约债务暂留待测试完备后处理;缓存同步/异步继承冲突保留为设计债务,未纳入严格门禁 |
|
||||
| 全量 mypy 历史债务 | 4,443 / 252 文件 | 本分支为 Pydantic 字段补充描述并保留默认值,减少 3 项 assignment 错误;本次 NFO 返回类型与契约对齐,减少 2 项历史类型错误;本轮在消息、站点、诊断、备份、用户配置、订阅、AcoustID、TMDB、仪表板和命令生命周期边界补齐类型,历史债务再减少 53 项;TMDB v3 API 显式导出并保留集合 parts 的列表合同,逐文件逐错误码无增长;此前工作流筛选、动态插件路由和 stdio 流补齐边界类型,历史债务减少 3 项;Bitpt 与 CookieCloud 补齐边界类型,历史债务减少 16 项;知行站点解析器补齐数值属性与基类合同,历史债务减少 18 项;通知排版补齐 Slack 消息块类型,历史债务减少 2 项;工作流资源过滤改用领域对象调用过滤链,恢复的节点输出在消费前归一化,并收紧可空资源列表的类型低水位;消息媒体交互类型债务已清零,canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降;微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后,类型错误减少 4 项;站点解析器分享率改用浮点初值,类型错误减少 20 项;本次下载与文件安全修复再减少 3 项;Transmission 进度回退值归一化再减少 2 项;本批清理 Servarr、认证、存储、订阅策略和搜索/规则/通知 Schema 的 18 项类型错误,并修复系统配置中的 Python 2 异常语法;本批再清理订阅元数据、订阅写入、交互宿主、整理路径、删除统计和工作流动作的 14 项类型错误,逐文件逐错误码无增长;本批补齐 OCR 请求参数、工具 Schema 摘要和宿主模块配置快照的 5 项边界类型,逐文件逐错误码无增长;本批为 Rousi 用户消息分页解析补齐返回类型与数据结构兜底,减少 2 项类型错误;本批优先处理 TMDB v3 电影与电视剧对象模型,补齐 98 个同步/异步方法注解并减少 122 项历史类型错误;随后补齐剧集、季度、人物、搜索和趋势对象的 142 个方法注解,再减少 142 项类型错误;绿联 API 响应边界补齐具体泛型与运行时字典收窄,减少 31 项类型错误;运行时限流器补齐状态方法、装饰器边界和时间队列类型,减少 31 项类型错误;QQ Bot API 响应边界补齐具体泛型与 JSON 形状校验,减少 17 项类型错误;rTorrent 客户端清理 182 项错误,调用端同步明确字典合同,累计再减少 190 项错误;豆瓣 API、中心服务接口及订阅消费端补齐返回边界和参数收窄,结合 Context 方法签名与缓存代理类型清理,累计再减少 455 项错误;共享 HTTP 客户端清理 110 项错误,TMDB API 清理 84 项错误,索引器爬虫清理 87 项错误,并补充 U115、AliPan、OpenList 对可空流响应的防守;U115 其余存储基类契约债务暂留待测试完备后处理;缓存同步/异步继承冲突保留为设计债务,未纳入严格门禁 |
|
||||
| Ruff 历史诊断 | 73 | 低水位门禁通过,但规则集只覆盖 `E4/E7/E9/F/I` |
|
||||
| 覆盖率逐文件低水位 | 全 app(排除 app/plugins)当前聚合 77.68% | 存量文件比例不降;语句减少且未覆盖行数不增时允许比例变化,新文件至少 80% |
|
||||
|
||||
|
||||
@@ -817,7 +817,7 @@
|
||||
"union-attr": 18
|
||||
},
|
||||
"app/modules/douban/__init__.py": {
|
||||
"assignment": 24,
|
||||
"assignment": 23,
|
||||
"empty-body": 1,
|
||||
"no-any-return": 1,
|
||||
"no-untyped-def": 7,
|
||||
@@ -1220,7 +1220,6 @@
|
||||
"index": 2,
|
||||
"no-any-return": 4,
|
||||
"no-untyped-def": 3,
|
||||
"return-value": 1,
|
||||
"type-arg": 6
|
||||
},
|
||||
"app/modules/themoviedb/tmdbv3api/as_obj.py": {
|
||||
|
||||
67
tests/test_metadata_nfo_contract.py
Normal file
67
tests/test_metadata_nfo_contract.py
Normal file
@@ -0,0 +1,67 @@
|
||||
"""
|
||||
验证内置 NFO 生成器的返回值符合模块契约。
|
||||
"""
|
||||
|
||||
import xml.etree.ElementTree as ET
|
||||
from typing import Optional
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from app.domain.context import MediaInfo
|
||||
from app.domain.meta.metabase import MetaBase
|
||||
from app.domain.scraper import MediaScraperHelper
|
||||
from app.modules.douban.scraper import DoubanScraper
|
||||
from app.modules.themoviedb.scraper import TmdbScraper
|
||||
from app.runtime.extensions.module.contracts import diagnose_module_result
|
||||
from app.schemas.types import MediaType
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"scraper_type",
|
||||
(TmdbScraper, DoubanScraper, MediaScraperHelper),
|
||||
)
|
||||
def test_builtin_movie_nfo_returns_utf8_text(scraper_type: type) -> None:
|
||||
"""
|
||||
内置生成器应返回可解析的中文 XML 文本,且通过模块结果契约检查。
|
||||
"""
|
||||
mediainfo = MediaInfo(type=MediaType.MOVIE, title="测试电影")
|
||||
scraper = scraper_type()
|
||||
if isinstance(scraper, TmdbScraper):
|
||||
nfo = scraper.get_metadata_nfo(meta=Mock(spec=MetaBase), mediainfo=mediainfo)
|
||||
else:
|
||||
nfo = scraper.get_metadata_nfo(mediainfo=mediainfo)
|
||||
|
||||
assert isinstance(nfo, str)
|
||||
assert diagnose_module_result("metadata_nfo", nfo) == ()
|
||||
assert nfo.startswith('<?xml version="1.0" encoding="utf-8"?>')
|
||||
document = ET.fromstring(nfo)
|
||||
assert document.tag == "movie"
|
||||
assert document.findtext("title") == "测试电影"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("season", "episode", "expected_root"),
|
||||
(
|
||||
(None, None, "tvshow"),
|
||||
(1, None, "season"),
|
||||
(1, 2, "episodedetails"),
|
||||
),
|
||||
)
|
||||
def test_shared_tv_nfo_returns_text_for_each_shape(
|
||||
season: Optional[int], episode: Optional[int], expected_root: str
|
||||
) -> None:
|
||||
"""
|
||||
统一生成器生成剧集、季和单集时都应保持文本返回值。
|
||||
"""
|
||||
mediainfo = MediaInfo(type=MediaType.TV, title="测试剧集")
|
||||
|
||||
nfo = MediaScraperHelper.get_metadata_nfo(
|
||||
mediainfo, season=season, episode=episode
|
||||
)
|
||||
|
||||
assert isinstance(nfo, str)
|
||||
assert diagnose_module_result("metadata_nfo", nfo) == ()
|
||||
document = ET.fromstring(nfo)
|
||||
assert document.tag == expected_root
|
||||
assert document.findtext("title") == "测试剧集"
|
||||
@@ -179,6 +179,17 @@ def test_literal_template_skips_music_fields_for_video() -> None:
|
||||
assert "类别:国产剧" in rendered["text"]
|
||||
|
||||
|
||||
def test_template_helper_supports_string_cache_keys_and_rejects_invalid_literal() -> None:
|
||||
"""
|
||||
字符串模板缓存应可读回;显式 literal 模板解析失败时应给出明确错误。
|
||||
"""
|
||||
helper = TemplateHelper()
|
||||
helper.set_cache_context("template-key", {"title": "Demo"})
|
||||
|
||||
assert helper.get_cache_context("template-key") == {"title": "Demo"}
|
||||
assert helper.parse_template_content("{'title':", "literal") is None
|
||||
|
||||
|
||||
def test_message_renders_from_db_config(notification_templates: SystemConfigOper) -> None:
|
||||
"""
|
||||
MessageTemplateHelper 应使用数据库配置的模板渲染消息,
|
||||
|
||||
@@ -11,7 +11,7 @@ TemplateContextBuilder 的并发安全单元测试。
|
||||
import threading
|
||||
|
||||
from app.application.messaging.message import TemplateContextBuilder, TemplateHelper
|
||||
from app.domain.context import MediaInfo
|
||||
from app.domain.context import MediaInfo, TorrentInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
from app.domain.metainfo import MetaInfo
|
||||
from app.modules.filemanager.transhandler import TransHandler
|
||||
@@ -143,6 +143,41 @@ def test_build_exposes_total_episodes_from_current_season() -> None:
|
||||
assert context.get("total_episodes") == 3
|
||||
|
||||
|
||||
def test_build_exposes_matching_episode_metadata_when_name_is_empty() -> None:
|
||||
"""
|
||||
季集元数据命中当前集时,即使 TMDB 没有集名也应保留空字符串和日期字段。
|
||||
"""
|
||||
meta = MetaInfo("Demo S01E01")
|
||||
context = TemplateContextBuilder().build(
|
||||
meta=meta,
|
||||
episodes_info=[TmdbEpisode(episode_number=1, name=None, air_date="2025-01-02")],
|
||||
)
|
||||
|
||||
assert context["episode_title"] == ""
|
||||
assert context["episode_date"] == "2025-01-02"
|
||||
|
||||
|
||||
def test_build_projects_torrent_size_and_strips_html_description() -> None:
|
||||
"""
|
||||
种子上下文应格式化数字大小,并清理描述中的 HTML 标签。
|
||||
"""
|
||||
torrent = TorrentInfo(
|
||||
title="Demo",
|
||||
size=1024 * 1024 * 2,
|
||||
description="<p>简介 <b>内容</b></p>",
|
||||
labels=["免费"],
|
||||
hit_and_run=True,
|
||||
)
|
||||
context = {}
|
||||
|
||||
TemplateContextBuilder._add_torrent_info(context, torrent)
|
||||
|
||||
assert context["size"] == "2.0M"
|
||||
assert context["description"] == "简介 内容"
|
||||
assert context["hit_and_run"] == "是"
|
||||
assert context["labels"] == "免费"
|
||||
|
||||
|
||||
def test_build_preserves_special_season_context() -> None:
|
||||
"""
|
||||
显式 S00 必须优先于媒体回退季,并使用特别季年份。
|
||||
|
||||
159
tests/test_tmdb_module_coverage.py
Normal file
159
tests/test_tmdb_module_coverage.py
Normal file
@@ -0,0 +1,159 @@
|
||||
"""
|
||||
覆盖 TMDB 模块的边界分支和同步异步入口。
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
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
|
||||
from app.schemas.types import MediaSource, MediaType
|
||||
|
||||
|
||||
def _module() -> TheMovieDbModule:
|
||||
"""
|
||||
构造注入测试客户端的 TMDB 模块。
|
||||
"""
|
||||
module = object.__new__(TheMovieDbModule)
|
||||
module.tmdb = Mock()
|
||||
module.cache = Mock()
|
||||
module.category = Mock()
|
||||
return module
|
||||
|
||||
|
||||
def _request(media_type: str = "电影", missing_fields: Optional[list[str]] = None) -> ClassificationEnrichmentRequest:
|
||||
"""
|
||||
构造分类事实补充请求。
|
||||
"""
|
||||
return ClassificationEnrichmentRequest(
|
||||
identity={"media_source": "themoviedb", "media_id": "1"},
|
||||
media_type=media_type,
|
||||
missing_fields=missing_fields or ["genres"],
|
||||
external_ids={"themoviedb": "1"},
|
||||
policy_revision=1,
|
||||
timeout_seconds=1.0,
|
||||
)
|
||||
|
||||
|
||||
def test_module_metadata_and_classification_boundaries() -> None:
|
||||
"""
|
||||
模块元数据和分类补充应覆盖无效请求、缺失事实和成功响应。
|
||||
"""
|
||||
module = _module()
|
||||
assert module.get_name() == "TheMovieDb"
|
||||
assert module.get_type().value
|
||||
assert module.get_subtype().value
|
||||
assert module.get_priority() == 1
|
||||
assert module.init_setting() is None
|
||||
assert module.get_media_classification_facts(_request(media_type="音乐")) is None
|
||||
module.tmdb.get_info.return_value = None
|
||||
assert module.get_media_classification_facts(_request(missing_fields=["genres"])) is None
|
||||
module.tmdb.get_info.return_value = {
|
||||
"id": 1,
|
||||
"media_type": MediaType.MOVIE,
|
||||
"title": "测试",
|
||||
"genres": [{"id": 18, "name": "剧情"}],
|
||||
}
|
||||
result = module.get_media_classification_facts(_request(missing_fields=["media.title", "unknown"]))
|
||||
assert result is not None
|
||||
assert result.match.media_id == "1"
|
||||
assert "media.title" in result.facts
|
||||
assert "unknown" not in result.facts
|
||||
module.tmdb.get_info.return_value = None
|
||||
assert module.get_media_classification_facts(_request()) is None
|
||||
|
||||
|
||||
def test_module_connection_test_and_plan_boundaries(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""
|
||||
连接测试和识别计划应分别返回成功、状态码失败、网络失败及拒绝原因。
|
||||
"""
|
||||
module = _module()
|
||||
monkeypatch.setattr("app.modules.themoviedb.RequestUtils", lambda **_: Mock(get_res=Mock(return_value=Mock(status_code=200))))
|
||||
assert module.test() == (True, "")
|
||||
monkeypatch.setattr("app.modules.themoviedb.RequestUtils", lambda **_: Mock(get_res=Mock(return_value=Mock(status_code=503))))
|
||||
assert module.test()[0] is False
|
||||
monkeypatch.setattr("app.modules.themoviedb.RequestUtils", lambda **_: Mock(get_res=Mock(return_value=None)))
|
||||
assert module.test()[0] is False
|
||||
meta = MetaBase("测试 2024")
|
||||
assert module._build_recognize_plan(meta, MediaType.MUSIC, None, None, None, True) is None
|
||||
assert module._build_recognize_plan(meta, MediaType.MOVIE, MediaSource.Douban, None, None, True) is None
|
||||
assert module._build_recognize_plan(meta, MediaType.MOVIE, MediaSource.TMDB, "bad", None, True) is None
|
||||
assert module._build_recognize_plan(None, None, None, None, None, True) is None
|
||||
|
||||
|
||||
def test_module_search_and_episode_group_helpers() -> None:
|
||||
"""
|
||||
搜索计划、剧集组年份和媒体结果应覆盖空值与有效值分支。
|
||||
"""
|
||||
meta = MetaBase("测试 2024")
|
||||
assert TheMovieDbModule._build_media_search_plan(meta, None) is not None
|
||||
meta.name = ""
|
||||
plan = TheMovieDbModule._build_media_search_plan(meta, None)
|
||||
assert plan is not None and not plan.search_multi
|
||||
meta.name = "测试"
|
||||
meta.type = MediaType.UNKNOWN
|
||||
meta.year = None
|
||||
plan = TheMovieDbModule._build_media_search_plan(meta, None)
|
||||
assert plan is not None and plan.search_multi
|
||||
assert TheMovieDbModule._merge_media_search_results(plan, multi_results=[{"id": 1}]) == [{"id": 1}]
|
||||
tv = MediaInfo(tmdb_info={"id": 1, "media_type": MediaType.TV, "name": "测试", "first_air_date": "2024-01-01"})
|
||||
group = [{"order": 1, "episodes": [{"episode_number": 1, "air_date": "2024-02-03"}]}]
|
||||
TheMovieDbModule._fill_group_season_info(tv, "group", group)
|
||||
assert tv.seasons == {1: [1]}
|
||||
assert tv.season_years == {1: "2024"}
|
||||
assert TheMovieDbModule._collect_group_season_years(group) == {1: "2024"}
|
||||
search_meta = MetaBase("测试")
|
||||
search_meta.begin_season = 2
|
||||
result = TheMovieDbModule._build_search_medias_result(search_meta, [{"id": 1, "media_type": MediaType.TV, "name": "测试"}])
|
||||
assert result[0].season == 2
|
||||
|
||||
|
||||
def test_module_identity_disambiguation_and_cache_management() -> None:
|
||||
"""
|
||||
身份消歧、缓存回填和缓存管理应保留成功及拒绝分支。
|
||||
"""
|
||||
module = _module()
|
||||
meta = MetaBase("同名 2024")
|
||||
meta.type = MediaType.TV
|
||||
tv = {"id": 1, "name": "同名", "first_air_date": "2024-01-01"}
|
||||
movie = {"id": 2, "title": "其它", "release_date": "2020-01-01"}
|
||||
assert module._resolve_tmdbid_candidates(1, meta, tv, False, movie, False) == tv
|
||||
assert TheMovieDbModule._disambiguate_by_meta(tv, movie, meta) == tv
|
||||
assert TheMovieDbModule._disambiguate_by_meta(tv, movie, None) is None
|
||||
with pytest.raises(TMDbConnectionError):
|
||||
module._resolve_tmdbid_candidates(1, None, None, True, None, False)
|
||||
info = MediaInfo(tmdb_info={"id": 1, "media_type": MediaType.MOVIE, "title": "测试"})
|
||||
assert module.update_recognize_cache(meta, info) is True
|
||||
info.media_source = "douban"
|
||||
assert module.update_recognize_cache(meta, info) is None
|
||||
assert module.tmdb_cache_items() == module.cache.list_items.return_value
|
||||
module.tmdb_cache_delete("key")
|
||||
module.tmdb_cache_clear()
|
||||
module.clear_cache = Mock()
|
||||
|
||||
|
||||
def test_module_match_and_image_sync_async_boundaries() -> None:
|
||||
"""
|
||||
匹配详情、图片补全和异步入口应覆盖有效及空响应。
|
||||
"""
|
||||
module = _module()
|
||||
module.tmdb.match.return_value = {"id": 1, "media_type": MediaType.MOVIE, "title": "测试"}
|
||||
assert module.match_tmdbinfo("测试") is not None
|
||||
module.tmdb.match.return_value = {"id": 1, "media_type": MediaType.MOVIE}
|
||||
module.tmdb.get_info.return_value = {"id": 1, "media_type": MediaType.MOVIE, "title": "测试"}
|
||||
assert module.match_tmdbinfo("测试") is not None
|
||||
module.tmdb.async_match = AsyncMock(return_value=None)
|
||||
assert asyncio.run(module.async_match_tmdbinfo("测试")) is None
|
||||
module.tmdb.get_info.return_value = {"id": 1, "media_type": MediaType.MOVIE}
|
||||
assert module.tmdb_info(1, MediaType.MOVIE, 0) is not None
|
||||
module.tmdb.get_tv_season_detail.return_value = {"id": 1}
|
||||
assert module.tmdb_info(1, MediaType.TV, 1) == {"id": 1}
|
||||
module.tmdb.async_get_info = AsyncMock(return_value=None)
|
||||
assert asyncio.run(module.async_tmdb_info(1, MediaType.MOVIE)) is None
|
||||
assert TheMovieDbModule._pick_best_tmdb_image(None) is None
|
||||
137
tests/test_trimemedia_coverage.py
Normal file
137
tests/test_trimemedia_coverage.py
Normal file
@@ -0,0 +1,137 @@
|
||||
"""
|
||||
飞牛媒体服务器类型迁移后的边界分支测试。
|
||||
|
||||
测试只使用本地替身,不连接真实飞牛服务。
|
||||
"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from app.modules.trimemedia import api as trimemedia_api
|
||||
from app.modules.trimemedia.trimemedia import TrimeMedia
|
||||
from app.schemas.types import MediaSource
|
||||
|
||||
|
||||
def _authenticated_client() -> TrimeMedia:
|
||||
"""
|
||||
构造一个带有最小认证状态的飞牛客户端替身。
|
||||
"""
|
||||
client = TrimeMedia.__new__(TrimeMedia)
|
||||
client._host = "https://trimemedia.example"
|
||||
client._username = "admin"
|
||||
client._password = "password"
|
||||
client._access_code = "access"
|
||||
client._playhost = None
|
||||
client._sync_libraries = []
|
||||
client._userinfo = trimemedia_api.User("user", "admin", 1)
|
||||
client._api = SimpleNamespace(
|
||||
host="https://trimemedia.example",
|
||||
token="token",
|
||||
cookies={"Trim-MC-access": "access"},
|
||||
)
|
||||
client.is_authenticated = Mock(return_value=True)
|
||||
return client
|
||||
|
||||
|
||||
def test_authenticate_returns_none_when_probe_fails(monkeypatch):
|
||||
"""
|
||||
临时认证 API 探测失败时返回 None。
|
||||
"""
|
||||
client = TrimeMedia.__new__(TrimeMedia)
|
||||
client._host = "https://trimemedia.example"
|
||||
client._username = "admin"
|
||||
client._password = "password"
|
||||
monkeypatch.setattr(
|
||||
TrimeMedia,
|
||||
"_TrimeMedia__create_api",
|
||||
staticmethod(lambda *_args: None),
|
||||
)
|
||||
|
||||
assert client.authenticate("admin", "password") is None
|
||||
|
||||
|
||||
def test_get_items_yields_nested_directory_items():
|
||||
"""
|
||||
媒体库目录应递归展开,并只产出电影条目。
|
||||
"""
|
||||
client = _authenticated_client()
|
||||
root = trimemedia_api.Item(guid="root", type=trimemedia_api.Type.DIRECTORY)
|
||||
movie = trimemedia_api.Item(guid="movie", type=trimemedia_api.Type.MOVIE)
|
||||
client._api.item_list = Mock(side_effect=[[root], [movie]])
|
||||
client._TrimeMedia__build_media_server_item = Mock(return_value="movie-result")
|
||||
|
||||
assert list(client.get_items("library")) == ["movie-result"]
|
||||
assert client._api.item_list.call_args_list[1].kwargs["guid"] == "root"
|
||||
|
||||
|
||||
def test_get_tv_episodes_skips_incomplete_episode_numbers():
|
||||
"""
|
||||
飞牛返回缺少季号或集号的条目时,只保留可用于统一契约的完整编号。
|
||||
"""
|
||||
client = _authenticated_client()
|
||||
client.get_iteminfo = Mock(
|
||||
return_value=SimpleNamespace(media_source=MediaSource.TMDB, media_id="1")
|
||||
)
|
||||
client._api.season_list = Mock(
|
||||
return_value=[SimpleNamespace(guid="season", season_number=1)]
|
||||
)
|
||||
client._api.episode_list = Mock(
|
||||
return_value=[
|
||||
SimpleNamespace(season_number=None, episode_number=1),
|
||||
SimpleNamespace(season_number=1, episode_number=None),
|
||||
SimpleNamespace(season_number=1, episode_number=2),
|
||||
]
|
||||
)
|
||||
|
||||
assert client.get_tv_episodes(item_id="series", media_source=MediaSource.TMDB, media_id="1") == (
|
||||
"series",
|
||||
{1: [2]},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["get_latest", "get_latest_backdrops"])
|
||||
def test_latest_rejects_none_count_like_previous_arithmetic(method):
|
||||
"""
|
||||
最新列表的 None 数量仍在原有乘法位置报告 TypeError。
|
||||
"""
|
||||
client = _authenticated_client()
|
||||
|
||||
with pytest.raises(TypeError, match="NoneType"):
|
||||
getattr(client, method)(num=None)
|
||||
|
||||
|
||||
def test_get_items_rejects_none_start_index_like_previous_arithmetic():
|
||||
"""
|
||||
媒体列表的 None 起始位置仍保持原有加法失败语义。
|
||||
"""
|
||||
client = _authenticated_client()
|
||||
|
||||
with pytest.raises(TypeError, match="NoneType"):
|
||||
list(client.get_items("library", start_index=None))
|
||||
|
||||
|
||||
def test_get_resume_builds_play_item_and_get_image_cookies(monkeypatch):
|
||||
"""
|
||||
继续观看条目应生成播放卡片,图片请求应返回带访问码的 Cookie。
|
||||
"""
|
||||
client = _authenticated_client()
|
||||
client._api.play_list = Mock(
|
||||
return_value=[
|
||||
trimemedia_api.Item(
|
||||
guid="movie",
|
||||
type=trimemedia_api.Type.MOVIE,
|
||||
title="电影",
|
||||
poster="/poster.jpg",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
resume = client.get_resume(num=1)
|
||||
assert resume and resume[0].id == "movie"
|
||||
monkeypatch.setattr("app.modules.trimemedia.trimemedia.SecurityUtils.is_safe_url", lambda *_args, **_kwargs: True)
|
||||
assert client.get_image_cookies("https://trimemedia.example/poster.jpg") == {
|
||||
"Trim-MC-token": "token",
|
||||
"Trim-MC-access": "access",
|
||||
}
|
||||
Reference in New Issue
Block a user