mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-03 22:04:57 +08:00
244 lines
8.1 KiB
Python
244 lines
8.1 KiB
Python
"""
|
||
验证电影合集整理时的媒体识别与目录规划。
|
||
"""
|
||
|
||
from dataclasses import replace
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from app.application.history import DownloadHistorySnapshot
|
||
from app.application.transfer.workflow import TransferTask
|
||
from app.chain.transfer.facade import TransferChain
|
||
from app.domain.context import MediaInfo
|
||
from app.domain.meta.metabase import MetaBase
|
||
from app.runtime.config import settings
|
||
from app.schemas.file import FileItem
|
||
from app.schemas.types import MediaSource, MediaType
|
||
|
||
|
||
def _make_chain() -> TransferChain:
|
||
"""
|
||
构造不启动后台线程的整理链测试实例。
|
||
"""
|
||
chain = object.__new__(TransferChain)
|
||
chain._scrape_batches = {}
|
||
chain._media_exts = settings.RMT_MEDIAEXT
|
||
chain._subtitle_exts = settings.RMT_SUBEXT
|
||
chain._audio_exts = settings.RMT_AUDIOEXT
|
||
chain._allowed_exts = (
|
||
chain._media_exts + chain._subtitle_exts + chain._audio_exts
|
||
)
|
||
chain.jobview = SimpleNamespace(
|
||
finish_task=lambda task: None,
|
||
try_remove_job=lambda task: None,
|
||
)
|
||
chain._TransferChain__get_trans_fileitems = lambda fileitem, predicate: [
|
||
(fileitem, False)
|
||
]
|
||
chain._TransferChain__put_to_jobview = lambda task: True
|
||
chain._register_scrape_batch_task = lambda task: None
|
||
chain._close_scrape_batch = lambda batch_id: None
|
||
return chain
|
||
|
||
|
||
class _FileMeta(MetaBase):
|
||
"""
|
||
电影合集文件的元数据。
|
||
|
||
继承 MetaBase 而非 SimpleNamespace：TransferTask.meta 已标注为 MetaBase，
|
||
pydantic 会做 isinstance 校验。name 是 MetaBase 上的 property，用类级属性遮蔽。
|
||
"""
|
||
|
||
name = None
|
||
|
||
def __init__(self, year: str = "2013"):
|
||
"""
|
||
初始化用于检验合集年份冲突的单部电影元数据。
|
||
"""
|
||
super().__init__(title="The Hunger Games Catching Fire")
|
||
self.name = "The Hunger Games Catching Fire"
|
||
self.year = year
|
||
self.type = MediaType.UNKNOWN
|
||
self.begin_season = None
|
||
self.begin_episode = None
|
||
self.part = None
|
||
|
||
|
||
def _make_file_meta(year: str = "2013") -> _FileMeta:
|
||
"""
|
||
构造电影合集文件的元数据。
|
||
"""
|
||
return _FileMeta(year=year)
|
||
|
||
|
||
def _make_history() -> DownloadHistorySnapshot:
|
||
"""
|
||
构造被合集首部电影占用的下载历史。
|
||
"""
|
||
return DownloadHistorySnapshot(
|
||
id=1,
|
||
path="/downloads/The.Hunger.Games.Complete.4-Film.Collection",
|
||
download_hash="collection-hash",
|
||
downloader="qbittorrent",
|
||
type=MediaType.MOVIE.value,
|
||
title="饥饿游戏",
|
||
year="2012",
|
||
media_source=MediaSource.TMDB,
|
||
media_id="70160",
|
||
episode_group=None,
|
||
media_category=None,
|
||
username=None,
|
||
custom_words=None,
|
||
note=None,
|
||
)
|
||
|
||
|
||
def test_movie_year_conflict_only_applies_to_movies():
|
||
"""
|
||
仅电影年份冲突应触发逐文件识别，电视剧季包仍复用下载历史。
|
||
"""
|
||
file_meta = _make_file_meta()
|
||
movie_history = _make_history()
|
||
tv_history = replace(movie_history, type=MediaType.TV.value)
|
||
|
||
assert TransferChain._is_movie_year_conflict(file_meta, movie_history)
|
||
assert not TransferChain._is_movie_year_conflict(file_meta, tv_history)
|
||
same_year_history = replace(movie_history, year="2013")
|
||
assert not TransferChain._is_movie_year_conflict(file_meta, same_year_history)
|
||
|
||
|
||
def test_conflicting_download_history_recognizes_movie_by_file_meta(monkeypatch):
|
||
"""
|
||
手动整理未指定媒体时，冲突的合集历史应回退到文件元数据识别。
|
||
"""
|
||
chain = object.__new__(TransferChain)
|
||
fallback_media = MediaInfo(
|
||
type=MediaType.MOVIE,
|
||
title="饥饿游戏2：星火燎原",
|
||
year="2013",
|
||
tmdb_id=101299,
|
||
)
|
||
recognized_meta = []
|
||
chain.jobview = SimpleNamespace(
|
||
migrate_task=lambda task: False,
|
||
try_remove_job=lambda task: None,
|
||
)
|
||
chain.transfer_history_repository = SimpleNamespace(
|
||
get_by_type_tmdbid=lambda **kwargs: None
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.chain.transfer.execution.MediaChain",
|
||
lambda: SimpleNamespace(
|
||
recognize_media=lambda **kwargs: pytest.fail("不应按合集历史 ID 识别"),
|
||
recognize_by_meta=lambda meta, obtain_images: (
|
||
recognized_meta.append(meta) or fallback_media
|
||
),
|
||
supplement_tmdb_info=lambda media, _meta: media,
|
||
),
|
||
)
|
||
monkeypatch.setattr("app.chain.transfer.filter.MediaChain", lambda: SimpleNamespace(
|
||
recognize_media=lambda **kwargs: pytest.fail("不应按合集历史 ID 识别"),
|
||
recognize_by_meta=lambda meta, obtain_images: (
|
||
recognized_meta.append(meta) or fallback_media
|
||
),
|
||
supplement_tmdb_info=lambda media, _meta: media,
|
||
))
|
||
task = TransferTask(
|
||
fileitem=FileItem(
|
||
storage="local",
|
||
path="/downloads/collection/The.Hunger.Games.Catching.Fire.2013.mkv",
|
||
type="file",
|
||
name="The.Hunger.Games.Catching.Fire.2013.mkv",
|
||
extension="mkv",
|
||
size=1024,
|
||
),
|
||
meta=_make_file_meta(),
|
||
download_history=_make_history(),
|
||
preview=True,
|
||
)
|
||
|
||
state, message = chain._TransferChain__handle_transfer(task)
|
||
|
||
assert not state
|
||
assert "已在整理队列中" in message
|
||
assert recognized_meta == [task.meta]
|
||
assert task.mediainfo.tmdb_id == 101299
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("manual", "expected_tmdb_id"),
|
||
[
|
||
(False, None),
|
||
(True, 70160),
|
||
],
|
||
)
|
||
def test_movie_collection_conflict_only_drops_automatic_media(
|
||
monkeypatch, manual: bool, expected_tmdb_id: int
|
||
):
|
||
"""
|
||
自动整理应丢弃冲突的合集媒体，手动明确指定的媒体仍应保留。
|
||
"""
|
||
chain = _make_chain()
|
||
source_file = FileItem(
|
||
storage="local",
|
||
path=(
|
||
"/downloads/The.Hunger.Games.Complete.4-Film.Collection/"
|
||
"The.Hunger.Games.Catching.Fire.2013.mkv"
|
||
),
|
||
type="file",
|
||
name="The.Hunger.Games.Catching.Fire.2013.mkv",
|
||
extension="mkv",
|
||
size=1024,
|
||
)
|
||
file_meta = _make_file_meta()
|
||
history = _make_history()
|
||
history_oper = SimpleNamespace(
|
||
get_by_hash=lambda download_hash: history,
|
||
get_file_by_fullpath=lambda fullpath: None,
|
||
get_files_by_savepath=lambda savepath: [],
|
||
get_by_path=lambda path: None,
|
||
)
|
||
captured_tasks = []
|
||
|
||
def fake_handle_transfer(task, callback=None):
|
||
"""
|
||
记录整理任务，避免执行真实文件操作。
|
||
"""
|
||
captured_tasks.append(task)
|
||
return True, ""
|
||
|
||
chain._TransferChain__handle_transfer = fake_handle_transfer
|
||
chain.transfer_history_repository = SimpleNamespace(
|
||
get_by_src=lambda src, storage=None: None
|
||
)
|
||
chain.download_history_repository = history_oper
|
||
monkeypatch.setattr(
|
||
"app.chain.transfer.workflow.get_configured_system_config",
|
||
lambda: SimpleNamespace(get=lambda key: None),
|
||
)
|
||
monkeypatch.setattr("app.chain.transfer.format.get_configured_system_config", lambda: SimpleNamespace(get=lambda key: None))
|
||
monkeypatch.setattr("app.chain.transfer.request.StorageChain", lambda: SimpleNamespace())
|
||
monkeypatch.setattr("app.chain.transfer.records.StorageChain", lambda: SimpleNamespace())
|
||
monkeypatch.setattr("app.chain.transfer.request.MetaInfoPath", lambda *args, **kwargs: file_meta)
|
||
|
||
# 用真 MediaInfo 而非 SimpleNamespace：它会被装进 TransferTask.mediainfo，
|
||
# 那个字段已标注为 MusicInfo | MediaInfo，pydantic 会做 isinstance 校验
|
||
conflicting_media = MediaInfo()
|
||
conflicting_media.tmdb_id = 70160
|
||
conflicting_media.type = MediaType.MOVIE
|
||
conflicting_media.year = "2012"
|
||
|
||
chain.do_transfer(
|
||
fileitem=source_file,
|
||
mediainfo=conflicting_media,
|
||
download_hash=history.download_hash,
|
||
background=False,
|
||
manual=manual,
|
||
preview=True,
|
||
)
|
||
|
||
assert len(captured_tasks) == 1
|
||
task_media = captured_tasks[0].mediainfo
|
||
assert getattr(task_media, "tmdb_id", None) == expected_tmdb_id
|