mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-01 21:07:01 +08:00
598d1663b1
* fix(download): preserve protocol failures and isolate site updates * test: cover tracker token lifecycle and record dependency facts --------- Co-authored-by: jxxghp <jxxghp@gmail.com>
2094 lines
72 KiB
Python
2094 lines
72 KiB
Python
"""
|
||
验证下载链路的提交、历史写入、剧集识别和后续处理。
|
||
"""
|
||
|
||
from collections.abc import Iterator
|
||
from dataclasses import asdict
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
|
||
import app.chain.download.batch as download_batch
|
||
import app.chain.download.existence as download_existence
|
||
import app.chain.download.failure as download_failure
|
||
import app.chain.download.processing as download_processing
|
||
import app.chain.download.selection as download_selection
|
||
import app.chain.download.submission as download_submission
|
||
import app.chain.download.subtitle as download_subtitle
|
||
from app.application.download.failures import (
|
||
DownloadFailureSnapshot,
|
||
DownloadFailureWrite,
|
||
)
|
||
from app.application.history import DownloadHistorySnapshot
|
||
from app.chain.download import DownloadChain
|
||
from app.domain.context import Context, MediaInfo, SubtitleInfo, TorrentInfo
|
||
from app.domain.metainfo import MetaInfo
|
||
from app.runtime.config import settings
|
||
from app.runtime.events import eventmanager
|
||
from app.schemas.file import FileItem
|
||
from app.schemas.mediaserver import NotExistMediaInfo
|
||
from app.schemas.system import TransferDirectoryConf
|
||
from app.schemas.transfer import DownloaderTorrent
|
||
from app.schemas.types import MediaSource, MediaType
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _restore_eventmanager_instance_override() -> Iterator[None]:
|
||
"""
|
||
每个用例后恢复事件单例实例属性，避免遮蔽后续类级 monkeypatch。
|
||
"""
|
||
instance = eventmanager
|
||
marker = object()
|
||
original = vars(instance).get("send_event", marker)
|
||
yield
|
||
if original is marker:
|
||
vars(instance).pop("send_event", None)
|
||
else:
|
||
vars(instance)["send_event"] = original
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _mock_tmdb_supplement(monkeypatch):
|
||
"""
|
||
隔离下载用例中的 TMDB 辅助识别外部边界。
|
||
"""
|
||
|
||
class _NoopMediaChain:
|
||
"""
|
||
保持原媒体对象不变的 TMDB 辅助识别替身。
|
||
"""
|
||
|
||
@staticmethod
|
||
def supplement_tmdb_info(media, _meta):
|
||
"""
|
||
返回原媒体对象。
|
||
"""
|
||
return media
|
||
|
||
monkeypatch.setattr(download_submission, "MediaChain", _NoopMediaChain)
|
||
monkeypatch.setattr(download_subtitle, "MediaChain", _NoopMediaChain)
|
||
monkeypatch.setattr(download_existence, "MediaChain", _NoopMediaChain)
|
||
|
||
|
||
class _FakeDownloadHistoryOper:
|
||
"""
|
||
避免单元测试写入真实下载历史，只验证下载链路的控制流。
|
||
"""
|
||
|
||
def add(self, _history, _files=()):
|
||
"""
|
||
忽略当前用例无需验证的类型化历史写入。
|
||
"""
|
||
pass
|
||
|
||
|
||
class _FakeTorrentHelper:
|
||
"""
|
||
避免解析真实种子内容，让测试聚焦下载成功后的后台处理。
|
||
"""
|
||
|
||
def get_fileinfo_from_torrent_content(self, _torrent_content):
|
||
"""
|
||
返回空种子名称和文件列表，隔离真实种子内容解析。
|
||
"""
|
||
return "", []
|
||
|
||
|
||
class _FakeThreadHelper:
|
||
"""
|
||
捕获提交到线程池的任务，测试中手动触发以避免真正启动后台线程。
|
||
"""
|
||
|
||
submitted = []
|
||
|
||
def submit(self, func, *args, **kwargs):
|
||
"""
|
||
记录待执行函数及参数，由测试决定何时运行。
|
||
"""
|
||
self.submitted.append((func, args, kwargs))
|
||
|
||
|
||
def _download_dirs():
|
||
"""
|
||
构造下载成功路径测试使用的允许下载目录配置。
|
||
"""
|
||
return [
|
||
TransferDirectoryConf(
|
||
name="本地下载",
|
||
priority=1,
|
||
storage="local",
|
||
download_path="/downloads",
|
||
),
|
||
]
|
||
|
||
|
||
class _FakeSubtitleStorageChain:
|
||
"""
|
||
模拟字幕 API 保存文件时使用的存储链。
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""
|
||
初始化上传记录。
|
||
"""
|
||
self.uploaded_files = []
|
||
|
||
def get_folder(self, storage, path):
|
||
"""
|
||
模拟目标目录存在或已创建。
|
||
"""
|
||
return FileItem(storage=storage, type="dir", path=path.as_posix(), name=path.name)
|
||
|
||
def get_file_item(self, _storage, _path):
|
||
"""
|
||
模拟目标字幕文件不存在。
|
||
"""
|
||
return None
|
||
|
||
def upload_file(self, fileitem, path):
|
||
"""
|
||
记录上传的临时字幕文件。
|
||
"""
|
||
self.uploaded_files.append(path)
|
||
return FileItem(
|
||
storage=fileitem.storage,
|
||
type="file",
|
||
path=(Path(fileitem.path) / path.name).as_posix(),
|
||
name=path.name,
|
||
)
|
||
|
||
|
||
class _FakeSubtitleResponse:
|
||
"""
|
||
模拟字幕 API 下载响应。
|
||
"""
|
||
|
||
content = b"subtitle-content"
|
||
headers = {}
|
||
|
||
|
||
class _FakeSubtitleResponseWithHeader:
|
||
"""
|
||
模拟带下载文件名响应头的字幕 API 响应。
|
||
"""
|
||
|
||
content = b"archive-content"
|
||
headers = {
|
||
"content-disposition": (
|
||
'attachment; filename="Hypnosis_AKA_Saimin_(1999)_480i_JAPANESE_NTSC_DVD_REMUX_MPEG-2_DD_2.0-MeeSta.rar"'
|
||
)
|
||
}
|
||
|
||
|
||
def test_download_single_submits_download_added_to_background(monkeypatch):
|
||
"""
|
||
添加下载成功后，站点字幕等后处理应提交到后台，不能阻塞下载接口返回。
|
||
"""
|
||
_FakeThreadHelper.submitted = []
|
||
monkeypatch.setattr(
|
||
"app.application.directory.DirectoryHelper.get_download_dirs",
|
||
lambda _self: _download_dirs(),
|
||
)
|
||
monkeypatch.setattr(download_processing, "ThreadHelper", _FakeThreadHelper)
|
||
monkeypatch.setattr(download_submission, "TorrentHelper", _FakeTorrentHelper)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.download_history_repository = _FakeDownloadHistoryOper()
|
||
chain.download = MagicMock(return_value=("qb", "hash123", "Original", "添加下载成功"))
|
||
chain.download_added = MagicMock()
|
||
chain.download_site_subtitles = MagicMock()
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.post_message = MagicMock()
|
||
|
||
context = Context(
|
||
meta_info=MetaInfo("Demo Movie 2024"),
|
||
media_info=MediaInfo(
|
||
type=MediaType.MOVIE,
|
||
title="Demo Movie",
|
||
year="2024",
|
||
tmdb_id=1,
|
||
genre_ids=[18],
|
||
),
|
||
torrent_info=TorrentInfo(
|
||
title="Demo Movie 2024",
|
||
enclosure="https://example.com/demo.torrent",
|
||
site_cookie="uid=1",
|
||
site_name="TestSite",
|
||
),
|
||
)
|
||
|
||
result = chain.download_single(
|
||
context=context,
|
||
torrent_content=b"torrent-content",
|
||
save_path="/downloads",
|
||
username="tester",
|
||
)
|
||
|
||
assert result == "hash123"
|
||
chain.download_added.assert_not_called()
|
||
assert len(_FakeThreadHelper.submitted) == 1
|
||
|
||
task, args, kwargs = _FakeThreadHelper.submitted[0]
|
||
assert args == ()
|
||
assert kwargs == {}
|
||
|
||
task()
|
||
|
||
chain.download_added.assert_called_once_with(
|
||
context=context,
|
||
download_dir=Path("/downloads"),
|
||
torrent_content=b"torrent-content",
|
||
)
|
||
chain.download_site_subtitles.assert_called_once_with(
|
||
context=context,
|
||
download_dir=Path("/downloads"),
|
||
torrent_content=b"torrent-content",
|
||
download_hash="hash123",
|
||
downloader="qb",
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("content_path", "expected_target_dir"),
|
||
[
|
||
pytest.param(
|
||
"/downloading/Demo.Movie",
|
||
Path("/downloading/Demo.Movie"),
|
||
id="multi-file-torrent-root",
|
||
),
|
||
pytest.param(
|
||
"/downloading/Demo.Movie/Demo.Movie.mkv",
|
||
Path("/downloading/Demo.Movie"),
|
||
id="single-file-inside-same-name-folder",
|
||
),
|
||
],
|
||
)
|
||
def test_download_site_subtitles_uses_downloader_content_path_without_creating_save_folder(
|
||
monkeypatch,
|
||
content_path,
|
||
expected_target_dir,
|
||
):
|
||
"""
|
||
TempPath 字幕应定位到种子根目录，避免同名目录内单文件重复拼接。
|
||
"""
|
||
accessed_paths = []
|
||
|
||
class _FakeTorrentHelper:
|
||
"""
|
||
提供固定的多文件种子目录名。
|
||
"""
|
||
|
||
def get_fileinfo_from_torrent_content(self, _content):
|
||
"""
|
||
返回多文件种子目录名。
|
||
"""
|
||
return "Demo.Movie", []
|
||
|
||
class _FakeStorageChain:
|
||
"""
|
||
只让 TempPath 下的实际内容目录可见。
|
||
"""
|
||
|
||
def get_file_item(self, storage, path):
|
||
"""
|
||
记录查询并返回 TempPath 内容目录。
|
||
"""
|
||
accessed_paths.append((storage, path))
|
||
if path == expected_target_dir:
|
||
return FileItem(
|
||
storage=storage,
|
||
type="dir",
|
||
path=path.as_posix(),
|
||
name=path.name,
|
||
)
|
||
return None
|
||
|
||
def create_folder(self, *_args, **_kwargs):
|
||
"""
|
||
若生产代码尝试预建目录则让测试失败。
|
||
"""
|
||
pytest.fail("TempPath 字幕处理不应在 save_path 预建目录")
|
||
|
||
monkeypatch.setattr(download_subtitle, "TorrentHelper", _FakeTorrentHelper)
|
||
monkeypatch.setattr(download_subtitle, "StorageChain", _FakeStorageChain)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.runtime_config = SimpleNamespace(
|
||
download_subtitle=True,
|
||
proxy=None,
|
||
temporary_path=Path("/tmp/moviepilot-test"),
|
||
subtitle_extensions=tuple(settings.RMT_SUBEXT),
|
||
user_agent="MoviePilotTest",
|
||
)
|
||
chain.list_torrents = MagicMock(return_value=[DownloaderTorrent(
|
||
hash="hash123",
|
||
downloader="qb",
|
||
content_path=content_path,
|
||
)])
|
||
chain._site_subtitle_links = MagicMock(return_value=[])
|
||
context = Context(
|
||
torrent_info=TorrentInfo(page_url="https://example.com/torrent/1"),
|
||
)
|
||
monkeypatch.setattr(download_subtitle.time, "sleep", lambda _seconds: None)
|
||
|
||
chain.download_site_subtitles(
|
||
context=context,
|
||
download_dir=Path("/downloads"),
|
||
torrent_content=b"torrent-content",
|
||
download_hash="hash123",
|
||
downloader="qb",
|
||
)
|
||
|
||
assert accessed_paths == [("local", expected_target_dir)]
|
||
chain.list_torrents.assert_called_once_with(
|
||
hashs=["hash123"],
|
||
downloader="qb",
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("download_hash", [None, "not-yet-visible"])
|
||
@pytest.mark.parametrize("storage, prefix", [("smb", "smb:"), ("local", "")])
|
||
def test_subtitle_fallback_separates_storage_identity_from_path(download_hash, storage, prefix):
|
||
"""
|
||
下载器尚未返回内容目录时，字幕存储查询也只能接收去掉协议前缀的路径。
|
||
"""
|
||
list_torrents = MagicMock(return_value=[])
|
||
|
||
result = download_subtitle._resolve_site_subtitle_target_dir(
|
||
download_dir=Path(f"{prefix}/downloads/TV"),
|
||
folder_name="Demo.Show", download_hash=download_hash,
|
||
downloader="qb", list_torrents=list_torrents,
|
||
)
|
||
|
||
assert result == (storage, Path("/downloads/TV/Demo.Show"))
|
||
|
||
|
||
def test_download_site_subtitles_does_not_create_missing_save_folder(monkeypatch):
|
||
"""
|
||
找不到实际内容目录时应放弃字幕写入，不能制造迁移冲突目录。
|
||
"""
|
||
class _FakeTorrentHelper:
|
||
"""
|
||
提供固定的多文件种子目录名。
|
||
"""
|
||
|
||
def get_fileinfo_from_torrent_content(self, _content):
|
||
"""
|
||
返回多文件种子目录名。
|
||
"""
|
||
return "Demo.Movie", []
|
||
|
||
class _FakeStorageChain:
|
||
"""
|
||
模拟最终 save_path 下尚未出现种子目录。
|
||
"""
|
||
|
||
def get_file_item(self, _storage, _path):
|
||
"""
|
||
始终报告目标目录不存在。
|
||
"""
|
||
return None
|
||
|
||
def create_folder(self, *_args, **_kwargs):
|
||
"""
|
||
若生产代码尝试预建目录则让测试失败。
|
||
"""
|
||
pytest.fail("缺失目录时不应预建 save_path 子目录")
|
||
|
||
monkeypatch.setattr(download_subtitle, "TorrentHelper", _FakeTorrentHelper)
|
||
monkeypatch.setattr(download_subtitle, "StorageChain", _FakeStorageChain)
|
||
monkeypatch.setattr(download_subtitle.time, "sleep", lambda _seconds: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.runtime_config = SimpleNamespace(
|
||
download_subtitle=True,
|
||
proxy=None,
|
||
temporary_path=Path("/tmp/moviepilot-test"),
|
||
subtitle_extensions=tuple(settings.RMT_SUBEXT),
|
||
user_agent="MoviePilotTest",
|
||
)
|
||
chain.list_torrents = MagicMock(return_value=[])
|
||
chain.transfer_history_repository = SimpleNamespace(
|
||
list_by_hash=MagicMock(return_value=[]),
|
||
)
|
||
chain._site_subtitle_links = MagicMock()
|
||
context = Context(
|
||
torrent_info=TorrentInfo(page_url="https://example.com/torrent/1"),
|
||
)
|
||
|
||
chain.download_site_subtitles(
|
||
context=context,
|
||
download_dir=Path("/downloads"),
|
||
torrent_content=b"torrent-content",
|
||
download_hash="hash123",
|
||
downloader="qb",
|
||
)
|
||
|
||
chain._site_subtitle_links.assert_not_called()
|
||
|
||
|
||
def test_download_site_subtitles_recreates_source_directory_and_queues_subtitle(monkeypatch):
|
||
"""
|
||
视频整理后源目录被移动清理时，应恢复目录并把新字幕交给整理队列。
|
||
"""
|
||
accessed_paths = []
|
||
created_dirs = []
|
||
transfer_calls = []
|
||
operation_order = []
|
||
target_dir = Path("/download/影视下载/下载中/Demo.Show")
|
||
subtitle_item = FileItem(
|
||
storage="local",
|
||
type="file",
|
||
path=(target_dir / "Demo.Show.S01E01.CHS.srt").as_posix(),
|
||
name="Demo.Show.S01E01.CHS.srt",
|
||
extension="srt",
|
||
)
|
||
transfer_histories = [SimpleNamespace(
|
||
status=True,
|
||
src="/download/影视下载/下载中/Demo.Show/Demo.Show.S01E01.1080p.WEB-DL.mkv",
|
||
dest="/media/电视剧/Example/Season 01/Example - S01E01.mkv",
|
||
)]
|
||
|
||
class _FakeTorrentHelper:
|
||
"""
|
||
提供固定的多文件种子目录名。
|
||
"""
|
||
|
||
def get_fileinfo_from_torrent_content(self, _content):
|
||
"""
|
||
返回多文件种子目录名。
|
||
"""
|
||
return "Demo.Show", []
|
||
|
||
class _FakeStorageChain:
|
||
"""
|
||
模拟源目录被整理清理并能重新创建。
|
||
"""
|
||
|
||
def get_file_item(self, storage, path):
|
||
"""
|
||
记录源目录检查并报告其不存在。
|
||
"""
|
||
accessed_paths.append((storage, path))
|
||
return None
|
||
|
||
def get_folder(self, storage, path):
|
||
"""
|
||
仅重建原下载目录。
|
||
"""
|
||
created_dirs.append((storage, path))
|
||
return FileItem(
|
||
storage=storage,
|
||
type="dir",
|
||
path=path.as_posix(),
|
||
name=path.name,
|
||
)
|
||
|
||
def list_files(self, _folder, recursion=False):
|
||
"""
|
||
提供字幕文件，并确认保存操作先于队列提交。
|
||
"""
|
||
assert recursion is False
|
||
assert "saved" in operation_order
|
||
operation_order.append("list")
|
||
return [subtitle_item]
|
||
|
||
class _FakeTransferChain:
|
||
"""
|
||
记录标准整理队列提交。
|
||
"""
|
||
|
||
def do_transfer(self, **kwargs):
|
||
"""
|
||
保存自动整理请求参数。
|
||
"""
|
||
transfer_calls.append(kwargs)
|
||
operation_order.append("queued")
|
||
return True, "已添加到整理队列"
|
||
|
||
monkeypatch.setattr(download_subtitle, "TorrentHelper", _FakeTorrentHelper)
|
||
monkeypatch.setattr(download_subtitle, "StorageChain", _FakeStorageChain)
|
||
monkeypatch.setattr(download_subtitle.time, "sleep", lambda _seconds: None)
|
||
from app.chain.transfer import facade as transfer_facade
|
||
monkeypatch.setattr(transfer_facade, "TransferChain", _FakeTransferChain)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.runtime_config = SimpleNamespace(
|
||
download_subtitle=True,
|
||
proxy=None,
|
||
temporary_path=Path("/tmp/moviepilot-test"),
|
||
subtitle_extensions=tuple(settings.RMT_SUBEXT),
|
||
media_extensions=tuple(settings.RMT_MEDIAEXT),
|
||
user_agent="MoviePilotTest",
|
||
)
|
||
chain.list_torrents = MagicMock(return_value=[DownloaderTorrent(
|
||
hash="hash123",
|
||
downloader="qb",
|
||
content_path="/download/影视下载/下载中/Demo.Show",
|
||
)])
|
||
chain.transfer_history_repository = SimpleNamespace(
|
||
list_by_hash=MagicMock(return_value=transfer_histories),
|
||
)
|
||
chain._site_subtitle_links = MagicMock(return_value=["https://example.com/subtitle.srt"])
|
||
|
||
class _FakeResponse:
|
||
"""
|
||
为字幕下载回调提供可关闭的成功响应。
|
||
"""
|
||
|
||
status_code = 200
|
||
content = b"subtitle-content"
|
||
|
||
def close(self):
|
||
"""
|
||
模拟释放网络响应。
|
||
"""
|
||
|
||
class _FakeHttp:
|
||
"""
|
||
返回固定字幕响应，避免真实网络请求。
|
||
"""
|
||
|
||
def get(self, *_args, **_kwargs):
|
||
"""
|
||
返回成功响应。
|
||
"""
|
||
return _FakeResponse()
|
||
|
||
monkeypatch.setattr(
|
||
download_subtitle,
|
||
"_download_ports_snapshot",
|
||
lambda: (_FakeHttp(), object()),
|
||
)
|
||
|
||
def _save_response(**kwargs):
|
||
"""
|
||
记录字幕已保存到重新创建的原下载目录。
|
||
"""
|
||
assert kwargs["working_dir_item"].path == target_dir.as_posix()
|
||
operation_order.append("saved")
|
||
|
||
chain._save_site_subtitle_response = MagicMock(side_effect=_save_response)
|
||
mediainfo = SimpleNamespace(type=MediaType.TV)
|
||
context = Context(
|
||
media_info=mediainfo,
|
||
torrent_info=TorrentInfo(page_url="https://example.com/torrent/1"),
|
||
)
|
||
|
||
chain.download_site_subtitles(
|
||
context=context,
|
||
download_dir=Path("/download/影视下载/下载中"),
|
||
torrent_content=b"torrent-content",
|
||
download_hash="hash123",
|
||
downloader="qb",
|
||
)
|
||
|
||
assert accessed_paths == [("local", target_dir)]
|
||
assert created_dirs == [("local", target_dir)]
|
||
chain.transfer_history_repository.list_by_hash.assert_called_once_with("hash123")
|
||
chain._site_subtitle_links.assert_called_once_with(context)
|
||
chain._save_site_subtitle_response.assert_called_once()
|
||
assert operation_order == ["saved", "list", "queued"]
|
||
assert len(transfer_calls) == 1
|
||
assert transfer_calls[0] == {
|
||
"fileitem": subtitle_item,
|
||
"mediainfo": mediainfo,
|
||
"mtype": MediaType.TV,
|
||
"downloader": "qb",
|
||
"download_hash": "hash123",
|
||
"background": True,
|
||
}
|
||
|
||
|
||
def test_download_single_supplements_category_before_download_event(monkeypatch):
|
||
"""
|
||
下载事件和目录选择前应已有 TMDB 分类，同时保留原识别源身份。
|
||
"""
|
||
captured = {}
|
||
|
||
class _FakeMediaChain:
|
||
"""
|
||
模拟 TMDB 辅助识别并记录调用。
|
||
"""
|
||
|
||
@staticmethod
|
||
def supplement_tmdb_info(media, _meta):
|
||
"""
|
||
给原媒体对象补充下载分类。
|
||
"""
|
||
media.tmdb_id = 12345
|
||
media.category = "日本动画"
|
||
return media
|
||
|
||
def cancel_download(_event_type, event_data):
|
||
"""
|
||
捕获下载事件后取消，避免进入真实下载流程。
|
||
"""
|
||
captured["event_data"] = event_data
|
||
event_data.cancel = True
|
||
return SimpleNamespace(event_data=event_data)
|
||
|
||
monkeypatch.setattr(download_submission, "MediaChain", _FakeMediaChain)
|
||
monkeypatch.setattr(eventmanager, "send_event", cancel_download)
|
||
media = MediaInfo(
|
||
media_source=MediaSource.Bangumi,
|
||
media_id="40000",
|
||
bangumi_id=40000,
|
||
type=MediaType.TV,
|
||
title="测试动画",
|
||
)
|
||
context = Context(
|
||
meta_info=MetaInfo("测试动画 S01"),
|
||
media_info=media,
|
||
torrent_info=TorrentInfo(title="测试动画 S01"),
|
||
)
|
||
|
||
result = DownloadChain.__new__(DownloadChain).download_single(
|
||
context=context,
|
||
torrent_content="magnet:?xt=urn:btih:test",
|
||
return_detail=True,
|
||
)
|
||
|
||
assert result == (None, "下载被事件取消")
|
||
assert captured["event_data"].options["media_category"] == "日本动画"
|
||
assert context.media_info.media_source == MediaSource.Bangumi
|
||
assert context.media_info.media_id == "40000"
|
||
assert context.media_info.tmdb_id == 12345
|
||
|
||
|
||
@pytest.mark.parametrize("save_path", ["", " "])
|
||
def test_resource_download_event_treats_blank_save_path_as_default(monkeypatch, save_path):
|
||
"""
|
||
旧订阅持久化的空白保存目录表示使用默认目录，不能被路径校验拒绝。
|
||
"""
|
||
captured = {}
|
||
|
||
def keep_event(_event_type, event_data):
|
||
"""
|
||
记录事件订阅方看到的保存目录。
|
||
"""
|
||
captured["save_path"] = event_data.options["save_path"]
|
||
return SimpleNamespace(event_data=event_data)
|
||
|
||
monkeypatch.setattr(eventmanager, "send_event", keep_event)
|
||
context = Context(
|
||
meta_info=MetaInfo("测试剧集 S01E01"),
|
||
media_info=MediaInfo(type=MediaType.TV, title="测试剧集"),
|
||
torrent_info=TorrentInfo(title="测试剧集 S01E01"),
|
||
)
|
||
|
||
result = DownloadChain._apply_resource_download_event(
|
||
context, None, None, "Subscribe|{}", None, save_path, None, None,
|
||
)
|
||
|
||
assert result == (None, None)
|
||
assert captured["save_path"] is None
|
||
|
||
|
||
def test_resource_download_event_treats_blank_override_as_default(monkeypatch):
|
||
"""
|
||
事件把保存目录改成空白时同样回到默认目录。
|
||
"""
|
||
|
||
def blank_override(_event_type, event_data):
|
||
"""
|
||
模拟插件清空保存目录。
|
||
"""
|
||
event_data.options["save_path"] = " "
|
||
return SimpleNamespace(event_data=event_data)
|
||
|
||
monkeypatch.setattr(eventmanager, "send_event", blank_override)
|
||
context = Context(
|
||
meta_info=MetaInfo("测试剧集 S01E01"),
|
||
media_info=MediaInfo(type=MediaType.TV, title="测试剧集"),
|
||
torrent_info=TorrentInfo(title="测试剧集 S01E01"),
|
||
)
|
||
|
||
result = DownloadChain._apply_resource_download_event(
|
||
context, None, None, None, None, "/downloads/tv", None, None,
|
||
)
|
||
|
||
assert result == (None, None)
|
||
|
||
|
||
def test_resource_download_event_still_rejects_invalid_save_path(monkeypatch):
|
||
"""
|
||
非空白的非法保存目录仍按白名单拒绝。
|
||
"""
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *_args: None)
|
||
context = Context(
|
||
meta_info=MetaInfo("测试剧集 S01E01"),
|
||
media_info=MediaInfo(type=MediaType.TV, title="测试剧集"),
|
||
torrent_info=TorrentInfo(title="测试剧集 S01E01"),
|
||
)
|
||
|
||
result = DownloadChain._apply_resource_download_event(
|
||
context, None, None, None, None, "relative/path", None, None,
|
||
)
|
||
|
||
assert result == ("relative/path", "保存路径必须是绝对路径")
|
||
|
||
|
||
def test_download_single_persists_custom_words_snapshot(monkeypatch):
|
||
"""
|
||
下载成功登记历史时，应把传入的订阅识别词原样存入快照，供整理时原样复现识别。
|
||
"""
|
||
captured = {}
|
||
|
||
class _CapturingDownloadHistoryOper:
|
||
"""
|
||
捕获写入下载历史的字段，验证识别词快照确实落库。
|
||
"""
|
||
|
||
def add(self, history, _files=()):
|
||
"""
|
||
捕获下载历史字段。
|
||
"""
|
||
captured.update(history.to_payload())
|
||
|
||
_FakeThreadHelper.submitted = []
|
||
monkeypatch.setattr(
|
||
"app.application.directory.DirectoryHelper.get_download_dirs",
|
||
lambda _self: _download_dirs(),
|
||
)
|
||
monkeypatch.setattr(download_processing, "ThreadHelper", _FakeThreadHelper)
|
||
monkeypatch.setattr(download_submission, "TorrentHelper", _FakeTorrentHelper)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.download_history_repository = _CapturingDownloadHistoryOper()
|
||
chain.download = MagicMock(return_value=("qb", "hash123", "Original", "添加下载成功"))
|
||
chain.download_added = MagicMock()
|
||
chain.download_site_subtitles = MagicMock()
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.post_message = MagicMock()
|
||
|
||
context = Context(
|
||
meta_info=MetaInfo("Demo Show 2024"),
|
||
media_info=MediaInfo(
|
||
type=MediaType.TV,
|
||
title="Demo Show",
|
||
year="2024",
|
||
tmdb_id=1,
|
||
genre_ids=[18],
|
||
poster_path="https://images.example.com/original/poster.jpg",
|
||
backdrop_path="https://images.example.com/original/backdrop.jpg",
|
||
),
|
||
torrent_info=TorrentInfo(
|
||
title="Demo Show 2024",
|
||
enclosure="https://example.com/demo.torrent",
|
||
site_cookie="uid=1",
|
||
site_name="TestSite",
|
||
),
|
||
)
|
||
|
||
custom_words = "S04 => S01\n第 <> 集 >> EP+66"
|
||
result = chain.download_single(
|
||
context=context,
|
||
torrent_content=b"torrent-content",
|
||
save_path="/downloads",
|
||
username="tester",
|
||
custom_words=custom_words,
|
||
)
|
||
|
||
assert result == "hash123"
|
||
assert chain.download.call_args.kwargs["custom_words"] == custom_words
|
||
assert captured["custom_words"] == custom_words
|
||
assert captured["poster"] == "https://images.example.com/w500/poster.jpg"
|
||
assert captured["image"] == "https://images.example.com/w500/backdrop.jpg"
|
||
|
||
|
||
def test_save_subtitle_response_creates_missing_temp_directory(monkeypatch, tmp_path):
|
||
"""
|
||
下载字幕 API 保存响应前应自动创建缺失的临时目录。
|
||
"""
|
||
storage_chain = _FakeSubtitleStorageChain()
|
||
temp_path = tmp_path / "missing-temp"
|
||
assert not temp_path.exists()
|
||
|
||
monkeypatch.setattr(download_subtitle, "StorageChain", lambda: storage_chain)
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.runtime_config = SimpleNamespace(
|
||
temporary_path=temp_path,
|
||
subtitle_extensions=tuple(settings.RMT_SUBEXT),
|
||
)
|
||
subtitle = SubtitleInfo(
|
||
title="Demo Movie",
|
||
enclosure="https://example.test/subtitle.srt",
|
||
file_name="Demo.Movie.zh-cn.srt",
|
||
)
|
||
|
||
success, message, saved_files = chain._save_subtitle_response(
|
||
subtitle=subtitle,
|
||
response=_FakeSubtitleResponse(),
|
||
storage="local",
|
||
target_dir=Path("/downloads"),
|
||
)
|
||
|
||
assert temp_path.exists()
|
||
assert success
|
||
assert message == "字幕文件保存成功"
|
||
assert saved_files == ["/downloads/Demo.Movie.zh-cn.srt"]
|
||
assert storage_chain.uploaded_files
|
||
|
||
|
||
def test_save_subtitle_response_accepts_rar_filename_from_header(monkeypatch, tmp_path):
|
||
"""
|
||
PHP 下载链接应按响应头文件名识别 RAR 字幕压缩包，而不是按 URL 后缀误拒绝。
|
||
"""
|
||
storage_chain = _FakeSubtitleStorageChain()
|
||
temp_path = tmp_path / "temp"
|
||
extracted_files = []
|
||
|
||
def fake_unpack_archive(archive_file, extract_dir, archive_format=None):
|
||
"""
|
||
按真实端口传入的独占目录构造RAR解压结果。
|
||
"""
|
||
assert archive_format == "rar"
|
||
assert archive_file.suffix == ".rar"
|
||
assert extract_dir.is_relative_to(temp_path)
|
||
extract_dir.mkdir(parents=True, exist_ok=True)
|
||
extracted_subtitle = extract_dir / "Hypnosis_AKA_Saimin_(1999).srt"
|
||
extracted_subtitle.write_text("subtitle", encoding="utf-8")
|
||
extracted_files.append(extracted_subtitle)
|
||
|
||
monkeypatch.setattr(download_subtitle, "StorageChain", lambda: storage_chain)
|
||
monkeypatch.setattr(
|
||
"app.adapters.system.host.SystemUtils.unpack_archive",
|
||
fake_unpack_archive,
|
||
)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.runtime_config = SimpleNamespace(
|
||
temporary_path=temp_path,
|
||
subtitle_extensions=tuple(settings.RMT_SUBEXT),
|
||
)
|
||
subtitle = SubtitleInfo(
|
||
title="Hypnosis",
|
||
enclosure="https://audiences.me/downloadsubs.php?torrentid=666519&subid=2195",
|
||
)
|
||
|
||
success, message, saved_files = chain._save_subtitle_response(
|
||
subtitle=subtitle,
|
||
response=_FakeSubtitleResponseWithHeader(),
|
||
storage="local",
|
||
target_dir=Path("/downloads"),
|
||
)
|
||
|
||
assert success
|
||
assert message == "字幕文件保存成功"
|
||
assert saved_files == ["/downloads/Hypnosis_AKA_Saimin_(1999).srt"]
|
||
assert storage_chain.uploaded_files == extracted_files
|
||
assert not list(temp_path.iterdir())
|
||
|
||
|
||
def test_save_subtitle_response_rejects_unsupported_filename_from_header(monkeypatch, tmp_path):
|
||
"""
|
||
响应头文件名不是字幕或支持的压缩包时，应继续拒绝保存。
|
||
"""
|
||
storage_chain = _FakeSubtitleStorageChain()
|
||
temp_path = tmp_path / "temp"
|
||
response = SimpleNamespace(
|
||
content=b"<html>error</html>",
|
||
headers={"content-disposition": 'attachment; filename="error.html"'},
|
||
)
|
||
|
||
monkeypatch.setattr(download_subtitle, "StorageChain", lambda: storage_chain)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.runtime_config = SimpleNamespace(
|
||
temporary_path=temp_path,
|
||
subtitle_extensions=tuple(settings.RMT_SUBEXT),
|
||
)
|
||
subtitle = SubtitleInfo(
|
||
title="Hypnosis",
|
||
enclosure="https://audiences.me/downloadsubs.php?torrentid=666519&subid=2195",
|
||
)
|
||
|
||
success, message, saved_files = chain._save_subtitle_response(
|
||
subtitle=subtitle,
|
||
response=response,
|
||
storage="local",
|
||
target_dir=Path("/downloads"),
|
||
)
|
||
|
||
assert not success
|
||
assert message == "下载链接不是支持的字幕文件：error.html"
|
||
assert saved_files == []
|
||
assert storage_chain.uploaded_files == []
|
||
|
||
|
||
class _FakeBatchTorrentHelper:
|
||
"""
|
||
为批量下载测试提供稳定排序和种子文件集数解析。
|
||
"""
|
||
|
||
episodes = []
|
||
received_custom_words = None
|
||
|
||
def sort_torrents(self, contexts):
|
||
"""
|
||
保持测试输入顺序，避免依赖真实站点优先级配置。
|
||
"""
|
||
return contexts
|
||
|
||
def sort_group_torrents(self, contexts):
|
||
"""
|
||
模拟真实提前控重行为，回归时会丢掉同一媒体季集的后续候选。
|
||
"""
|
||
results = []
|
||
added = set()
|
||
for context in contexts:
|
||
media = context.media_info
|
||
meta = context.meta_info
|
||
if media.type == MediaType.TV:
|
||
media_name = f"{media.title_year}{meta.season_episode}"
|
||
else:
|
||
media_name = media.title_year
|
||
if media_name in added:
|
||
continue
|
||
added.add(media_name)
|
||
results.append(context)
|
||
return results
|
||
|
||
def get_torrent_episodes(self, _files, custom_words=None):
|
||
"""
|
||
记录自定义识别词配置，并返回预设剧集编号的副本。
|
||
"""
|
||
type(self).received_custom_words = custom_words
|
||
return list(self.episodes)
|
||
|
||
|
||
def _build_tv_context(episode_list=None):
|
||
"""
|
||
构造标题未显式标集数的单季电视剧候选。
|
||
"""
|
||
episodes = episode_list or []
|
||
return SimpleNamespace(
|
||
media_info=SimpleNamespace(
|
||
type=MediaType.TV,
|
||
title_year="Test Show (2026)",
|
||
media_source=MediaSource.TMDB,
|
||
media_id="1",
|
||
tmdb_id=1,
|
||
douban_id=None,
|
||
bangumi_id=None,
|
||
anilist_id=None,
|
||
),
|
||
meta_info=SimpleNamespace(
|
||
season_list=[1],
|
||
season_episode="S01E01",
|
||
episode_list=episodes,
|
||
title="Test Show",
|
||
org_string="Test Show S01 2160p",
|
||
set_episodes=lambda begin, end: None,
|
||
),
|
||
torrent_info=SimpleNamespace(title="Test Show S01 2160p", site_name="TestSite"),
|
||
allowed_episodes=None,
|
||
confirmed_full_coverage=False,
|
||
)
|
||
|
||
|
||
def _tv_no_exists(season: int, info: NotExistMediaInfo) -> dict:
|
||
"""
|
||
以统一媒体身份键构造单季缺集测试数据。
|
||
"""
|
||
return {"tmdb:1": {season: info}}
|
||
|
||
|
||
def test_download_identity_matching_uses_only_media_source_and_media_id():
|
||
"""
|
||
通用下载匹配不得再把来源特定辅助 ID 当作主身份键。
|
||
"""
|
||
media = SimpleNamespace(
|
||
media_source=MediaSource.Douban,
|
||
media_id="douban-1",
|
||
tmdb_id=1,
|
||
douban_id="douban-1",
|
||
bangumi_id=2,
|
||
anilist_id=3,
|
||
)
|
||
|
||
assert DownloadChain._media_identity_keys(media) == {"douban:douban-1"}
|
||
assert DownloadChain._matches_media_identity(media, "douban:douban-1") is True
|
||
assert DownloadChain._matches_media_identity(media, 1) is False
|
||
|
||
|
||
def test_download_failure_fingerprint_ignores_auxiliary_provider_ids():
|
||
"""
|
||
缺少主身份时失败指纹应按标题年份降级，不得依赖 IMDb 或 TVDB 辅助 ID。
|
||
"""
|
||
torrent = SimpleNamespace(site=1, torrent_id="torrent-1")
|
||
|
||
def build_context(imdb_id: str, tvdb_id: int):
|
||
"""
|
||
构造只有辅助 Provider ID 的失败上下文。
|
||
"""
|
||
return SimpleNamespace(
|
||
media_info=SimpleNamespace(
|
||
type=MediaType.MOVIE,
|
||
media_source=None,
|
||
media_id=None,
|
||
title="Demo Movie",
|
||
year="2026",
|
||
imdb_id=imdb_id,
|
||
tvdb_id=tvdb_id,
|
||
season=None,
|
||
),
|
||
meta_info=SimpleNamespace(season=None, episode=None, episode_list=[]),
|
||
torrent_info=torrent,
|
||
)
|
||
|
||
assert DownloadChain._build_download_failure_fingerprint(
|
||
build_context("tt0000001", 1)
|
||
) == DownloadChain._build_download_failure_fingerprint(
|
||
build_context("tt9999999", 999)
|
||
)
|
||
|
||
|
||
def test_batch_download_rejects_complete_coverage_when_files_do_not_cover_target(monkeypatch):
|
||
"""
|
||
完整覆盖要求不能让 1-13 这种局部包冒充 1-143 的目标范围。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = list(range(1, 14))
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[],
|
||
total_episode=143,
|
||
start_episode=1,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == []
|
||
assert lefts == no_exists
|
||
assert context.confirmed_full_coverage is False
|
||
chain.download_single.assert_not_called()
|
||
|
||
|
||
def test_batch_download_preserves_special_season_zero(monkeypatch):
|
||
"""
|
||
特别季整季需求必须以季 0 匹配候选，不能回退成第 1 季。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = list(range(1, 7))
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
context.meta_info.season_list = [0]
|
||
context.meta_info.season_episode = "S00"
|
||
context.meta_info.org_string = "Test Show S00 2160p"
|
||
context.torrent_info.title = "Test Show S00 2160p"
|
||
no_exists = _tv_no_exists(
|
||
0,
|
||
NotExistMediaInfo(
|
||
season=0,
|
||
episodes=[],
|
||
total_episode=6,
|
||
start_episode=1,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == [context]
|
||
assert lefts == {}
|
||
chain.download_single.assert_called_once()
|
||
|
||
|
||
def test_batch_download_rejects_complete_coverage_when_only_missing_episodes_match(monkeypatch):
|
||
"""
|
||
完整覆盖要求目标范围全集，不能只覆盖当前缺口集。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = [4, 5]
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[4, 5],
|
||
total_episode=5,
|
||
start_episode=1,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == []
|
||
assert lefts == no_exists
|
||
assert context.confirmed_full_coverage is False
|
||
chain.download_single.assert_not_called()
|
||
|
||
|
||
def test_batch_download_tries_next_episode_candidate_when_first_download_fails(monkeypatch):
|
||
"""
|
||
同一季集的首个候选下载失败时，应继续尝试排序后的下一个候选资源。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = []
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_single = MagicMock(side_effect=[None, "hash"])
|
||
|
||
first_context = _build_tv_context(episode_list=[1])
|
||
first_context.torrent_info.title = "Test Show S01E01 First"
|
||
second_context = _build_tv_context(episode_list=[1])
|
||
second_context.torrent_info.title = "Test Show S01E01 Second"
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[1],
|
||
total_episode=1,
|
||
start_episode=1,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(
|
||
contexts=[first_context, second_context],
|
||
no_exists=no_exists,
|
||
)
|
||
|
||
assert downloads == [second_context]
|
||
assert lefts == {}
|
||
assert chain.download_single.call_count == 2
|
||
assert chain.download_single.call_args_list[0].args[0] is first_context
|
||
assert chain.download_single.call_args_list[1].args[0] is second_context
|
||
|
||
|
||
def test_batch_download_does_not_download_duplicate_movie_after_success(monkeypatch):
|
||
"""
|
||
电影保留失败重试能力，但同一影片成功一次后不应继续添加后续候选。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = []
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
first_context = SimpleNamespace(
|
||
media_info=SimpleNamespace(type=MediaType.MOVIE, title_year="Demo Movie (2026)"),
|
||
meta_info=SimpleNamespace(season_episode=""),
|
||
torrent_info=SimpleNamespace(title="Demo Movie First"),
|
||
)
|
||
second_context = SimpleNamespace(
|
||
media_info=SimpleNamespace(type=MediaType.MOVIE, title_year="Demo Movie (2026)"),
|
||
meta_info=SimpleNamespace(season_episode=""),
|
||
torrent_info=SimpleNamespace(title="Demo Movie Second"),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[first_context, second_context])
|
||
|
||
assert downloads == [first_context]
|
||
assert lefts is None
|
||
chain.download_single.assert_called_once()
|
||
assert chain.download_single.call_args.args[0] is first_context
|
||
|
||
|
||
def test_batch_download_threads_custom_words_to_download_single(monkeypatch):
|
||
"""
|
||
订阅识别词须经 batch_download 透传到 download_single，作为整理快照随下载存档。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = []
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = SimpleNamespace(
|
||
media_info=SimpleNamespace(type=MediaType.MOVIE, title_year="Demo Movie (2026)"),
|
||
meta_info=SimpleNamespace(season_episode=""),
|
||
torrent_info=SimpleNamespace(title="Demo Movie"),
|
||
)
|
||
|
||
custom_words = "S04 => S01\n第 <> 集 >> EP+66"
|
||
downloads, _lefts = chain.batch_download(contexts=[context], custom_words=custom_words)
|
||
|
||
assert downloads == [context]
|
||
chain.download_single.assert_called_once()
|
||
assert chain.download_single.call_args.kwargs["custom_words"] == custom_words
|
||
|
||
|
||
def test_batch_download_applies_custom_words_to_torrent_file_episodes(monkeypatch):
|
||
"""
|
||
订阅识别词须用于种子文件集数解析，确保跨季映射后能选中缺失集。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = [170]
|
||
_FakeBatchTorrentHelper.received_custom_words = None
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(
|
||
return_value=(b"torrent-content", "", ["A.Will.Eternal.S04E05.mkv"]),
|
||
)
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[170],
|
||
total_episode=200,
|
||
start_episode=155,
|
||
),
|
||
)
|
||
custom_words = (
|
||
"A.Will.Eternal.S04 => 一念永恒{[tmdbid=107371;type=tv]}S01 "
|
||
"&& S01 <> 2160p >> EP+165"
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(
|
||
contexts=[context],
|
||
no_exists=no_exists,
|
||
custom_words=custom_words,
|
||
)
|
||
|
||
assert downloads == [context]
|
||
assert lefts == {}
|
||
assert _FakeBatchTorrentHelper.received_custom_words == [custom_words]
|
||
assert chain.download_single.call_args.kwargs["episodes"] == {170}
|
||
|
||
|
||
def test_download_single_records_failure_cooldown_when_downloader_rejects(monkeypatch):
|
||
"""
|
||
下载器拒绝种子且没有返回 hash 时，应记录资源级失败冷却。
|
||
"""
|
||
captured = {}
|
||
|
||
class _CapturingDownloadFailureOper:
|
||
"""
|
||
捕获下载失败冷却记录，避免测试写入数据库。
|
||
"""
|
||
|
||
def record_failure(self, failure: DownloadFailureWrite) -> None:
|
||
"""
|
||
保存写入字段供断言使用。
|
||
"""
|
||
captured.update(asdict(failure))
|
||
|
||
monkeypatch.setattr(
|
||
"app.application.directory.DirectoryHelper.get_download_dirs",
|
||
lambda _self: _download_dirs(),
|
||
)
|
||
monkeypatch.setattr(download_submission, "TorrentHelper", _FakeTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.download_failure_repository = _CapturingDownloadFailureOper()
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
error_msg = "添加种子任务失败：无法读取种子文件"
|
||
chain.download = MagicMock(return_value=("qb", None, "Original", error_msg))
|
||
chain.post_message = MagicMock()
|
||
|
||
context = Context(
|
||
meta_info=MetaInfo("Demo Movie 2026"),
|
||
media_info=MediaInfo(
|
||
type=MediaType.MOVIE,
|
||
title="Demo Movie",
|
||
year="2026",
|
||
tmdb_id=1,
|
||
genre_ids=[18],
|
||
),
|
||
torrent_info=TorrentInfo(
|
||
site=12,
|
||
site_name="AGSVPT",
|
||
title="Demo Movie 2026 1080p",
|
||
enclosure="https://example.com/download.php?id=484660",
|
||
size=1024,
|
||
),
|
||
)
|
||
|
||
download_id, returned_error = chain.download_single(
|
||
context=context,
|
||
torrent_content=b"torrent-content",
|
||
save_path="/downloads",
|
||
source="Subscribe|{}",
|
||
return_detail=True,
|
||
)
|
||
|
||
assert download_id is None
|
||
assert returned_error == error_msg
|
||
assert captured["fingerprint"] == DownloadChain._build_download_failure_fingerprint(
|
||
context
|
||
)
|
||
assert captured["torrent_id"] == "example.com:id=484660"
|
||
assert captured["site"] == 12
|
||
assert captured["error_message"] == error_msg
|
||
assert captured["next_retry_at"] > captured["failed_at"]
|
||
|
||
|
||
def test_download_failure_query_skips_non_subscribe_sources(monkeypatch):
|
||
"""
|
||
非订阅下载不得读取失败冷却，避免改变手工下载行为。
|
||
"""
|
||
query = MagicMock(side_effect=AssertionError("unexpected query"))
|
||
repository = SimpleNamespace(
|
||
get_active_by_fingerprints=query,
|
||
)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.download_failure_repository = repository
|
||
|
||
assert chain._active_download_failure_fingerprints(
|
||
contexts=[_build_tv_context()],
|
||
source="Manual",
|
||
) == {}
|
||
query.assert_not_called()
|
||
|
||
|
||
def test_download_failure_query_failure_is_fail_open(monkeypatch):
|
||
"""
|
||
冷却查询失败时继续下载候选，并记录一次明确错误。
|
||
"""
|
||
|
||
class _FailingDownloadFailureRepository:
|
||
"""
|
||
模拟失败的下载冷却查询端口。
|
||
"""
|
||
|
||
def get_active_by_fingerprints(
|
||
self,
|
||
fingerprints: list[str],
|
||
now_time: str,
|
||
) -> dict[str, DownloadFailureSnapshot]:
|
||
"""
|
||
拒绝查询以验证 Chain 的 fail-open 语义。
|
||
"""
|
||
assert fingerprints
|
||
assert now_time
|
||
raise RuntimeError("query failed")
|
||
|
||
error = MagicMock()
|
||
monkeypatch.setattr(download_failure.logger, "error", error)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.download_failure_repository = _FailingDownloadFailureRepository()
|
||
|
||
assert chain._active_download_failure_fingerprints(
|
||
contexts=[_build_tv_context()],
|
||
source="Subscribe|{}",
|
||
) == {}
|
||
error.assert_called_once_with("查询下载失败冷却失败：query failed")
|
||
|
||
|
||
def test_download_failure_write_failure_is_fail_open(monkeypatch):
|
||
"""
|
||
冷却写入失败不得覆盖原下载结果，并记录一次明确错误。
|
||
"""
|
||
|
||
class _FailingDownloadFailureRepository:
|
||
"""
|
||
模拟失败的下载冷却写入端口。
|
||
"""
|
||
|
||
def record_failure(self, failure: DownloadFailureWrite) -> None:
|
||
"""
|
||
拒绝写入以验证 Chain 的 fail-open 语义。
|
||
"""
|
||
assert failure.fingerprint
|
||
raise RuntimeError("write failed")
|
||
|
||
error = MagicMock()
|
||
monkeypatch.setattr(download_failure.logger, "error", error)
|
||
context = _build_tv_context()
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.download_failure_repository = _FailingDownloadFailureRepository()
|
||
|
||
fingerprint = chain._record_download_failure(
|
||
context=context,
|
||
error_msg="下载失败",
|
||
source="Subscribe|{}",
|
||
)
|
||
|
||
assert fingerprint == DownloadChain._build_download_failure_fingerprint(context)
|
||
error.assert_called_once_with("记录下载失败冷却失败：write failed")
|
||
|
||
|
||
def test_download_failure_fingerprint_distinguishes_special_season_zero():
|
||
"""
|
||
失败冷却指纹应区分特别季与未指定季，避免错误共享冷却状态。
|
||
"""
|
||
def build_context(season):
|
||
return SimpleNamespace(
|
||
media_info=SimpleNamespace(
|
||
type=MediaType.TV,
|
||
title="Demo Show",
|
||
year="2026",
|
||
tmdb_id=1,
|
||
season=None,
|
||
),
|
||
meta_info=SimpleNamespace(season=season, episode=None, episode_list=[]),
|
||
torrent_info=SimpleNamespace(
|
||
site=12,
|
||
title="Demo Show Specials",
|
||
torrent_id="484660",
|
||
),
|
||
)
|
||
|
||
special_fingerprint = DownloadChain._build_download_failure_fingerprint(build_context(0))
|
||
unspecified_fingerprint = DownloadChain._build_download_failure_fingerprint(build_context(None))
|
||
|
||
assert special_fingerprint
|
||
assert unspecified_fingerprint
|
||
assert special_fingerprint != unspecified_fingerprint
|
||
|
||
|
||
def test_batch_download_skips_failed_subscription_resource_and_tries_next(monkeypatch):
|
||
"""
|
||
订阅自动下载应跳过冷却中的失败资源，但继续尝试同媒体的后续候选。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = []
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
first_context = SimpleNamespace(
|
||
media_info=SimpleNamespace(
|
||
type=MediaType.MOVIE,
|
||
title="Demo Movie",
|
||
year="2026",
|
||
title_year="Demo Movie (2026)",
|
||
tmdb_id=1,
|
||
douban_id=None,
|
||
),
|
||
meta_info=SimpleNamespace(season=None, episode=None, episode_list=[], season_episode=""),
|
||
torrent_info=SimpleNamespace(
|
||
site=12,
|
||
site_name="AGSVPT",
|
||
title="Demo Movie Bad",
|
||
torrent_id="484660",
|
||
size=1024,
|
||
),
|
||
)
|
||
second_context = SimpleNamespace(
|
||
media_info=SimpleNamespace(
|
||
type=MediaType.MOVIE,
|
||
title="Demo Movie",
|
||
year="2026",
|
||
title_year="Demo Movie (2026)",
|
||
tmdb_id=1,
|
||
douban_id=None,
|
||
),
|
||
meta_info=SimpleNamespace(season=None, episode=None, episode_list=[], season_episode=""),
|
||
torrent_info=SimpleNamespace(
|
||
site=13,
|
||
site_name="OtherSite",
|
||
title="Demo Movie Good",
|
||
torrent_id="999999",
|
||
size=2048,
|
||
),
|
||
)
|
||
failed_fingerprint = DownloadChain._build_download_failure_fingerprint(first_context)
|
||
|
||
class _ActiveDownloadFailureOper:
|
||
"""
|
||
返回第一个候选的活跃失败冷却记录。
|
||
"""
|
||
|
||
def get_active_by_fingerprints(
|
||
self,
|
||
fingerprints: list[str],
|
||
now_time: str,
|
||
) -> dict[str, DownloadFailureSnapshot]:
|
||
"""
|
||
模拟数据库批量查询活跃失败记录。
|
||
"""
|
||
assert now_time
|
||
assert failed_fingerprint in fingerprints
|
||
return {
|
||
failed_fingerprint: DownloadFailureSnapshot(
|
||
fingerprint=failed_fingerprint,
|
||
error_message="无法读取种子文件",
|
||
next_retry_at="2026-01-02 03:04:05",
|
||
)
|
||
}
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.download_failure_repository = _ActiveDownloadFailureOper()
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
downloads, lefts = chain.batch_download(
|
||
contexts=[first_context, second_context],
|
||
source="Subscribe|{}",
|
||
)
|
||
|
||
assert downloads == [second_context]
|
||
assert lefts is None
|
||
chain.download_single.assert_called_once()
|
||
assert chain.download_single.call_args.args[0] is second_context
|
||
|
||
|
||
def test_batch_download_accepts_complete_coverage_when_files_cover_target_range(monkeypatch):
|
||
"""
|
||
自定义起始集场景按目标范围覆盖判断，100-143 可满足 start=100、total=143。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = list(range(100, 144))
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[],
|
||
total_episode=143,
|
||
start_episode=100,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == [context]
|
||
assert lefts == {}
|
||
assert context.confirmed_full_coverage is True
|
||
chain.download_single.assert_called_once()
|
||
|
||
|
||
def test_batch_download_rejects_complete_coverage_when_files_have_same_count_but_wrong_range(monkeypatch):
|
||
"""
|
||
完整覆盖按目标集号集合判断，不能让同数量的偏移局部包通过。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = list(range(1, 45))
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[],
|
||
total_episode=143,
|
||
start_episode=100,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == []
|
||
assert lefts == no_exists
|
||
assert context.confirmed_full_coverage is False
|
||
chain.download_single.assert_not_called()
|
||
|
||
|
||
def test_batch_download_accepts_complete_coverage_when_title_episodes_cover_target(monkeypatch):
|
||
"""
|
||
显式标出完整范围的候选也可满足完整覆盖任务，并在整季阶段核对种子文件后复用内容提交。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = list(range(1, 144))
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent", None, ["files"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context(episode_list=list(range(1, 144)))
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[],
|
||
total_episode=143,
|
||
start_episode=1,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == [context]
|
||
assert lefts == {}
|
||
assert context.confirmed_full_coverage is True
|
||
chain.download_torrent.assert_called_once()
|
||
chain.download_single.assert_called_once()
|
||
assert chain.download_single.call_args.kwargs["torrent_content"] == b"torrent"
|
||
|
||
|
||
def test_batch_download_rejects_complete_coverage_when_title_episodes_are_partial(monkeypatch):
|
||
"""
|
||
显式标出局部范围的候选不能满足完整覆盖任务。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = []
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock()
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context(episode_list=list(range(1, 14)))
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[],
|
||
total_episode=143,
|
||
start_episode=1,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == []
|
||
assert lefts == no_exists
|
||
assert context.confirmed_full_coverage is False
|
||
chain.download_torrent.assert_not_called()
|
||
chain.download_single.assert_not_called()
|
||
|
||
|
||
def test_batch_download_complete_coverage_ignores_allowed_episode_narrowing(monkeypatch):
|
||
"""
|
||
完整覆盖任务不能因候选允许集裁剪而把局部包误判为覆盖目标范围。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = []
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock()
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context(episode_list=[1, 2])
|
||
context.allowed_episodes = {1, 2}
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[],
|
||
total_episode=12,
|
||
start_episode=1,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == []
|
||
assert lefts == no_exists
|
||
assert context.confirmed_full_coverage is False
|
||
chain.download_torrent.assert_not_called()
|
||
chain.download_single.assert_not_called()
|
||
|
||
|
||
def test_batch_download_skips_multi_episode_pack_outside_upgrade_allowlist(monkeypatch):
|
||
"""
|
||
分集洗版只允许 E20/E21 时，已达上限的 E01-E03 整包不得进入下载器。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = [1, 2, 3]
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context(episode_list=[1, 2, 3])
|
||
context.allowed_episodes = {20, 21}
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(season=1, episodes=list(range(1, 22)), total_episode=21),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == []
|
||
assert lefts == no_exists
|
||
chain.download_single.assert_not_called()
|
||
|
||
|
||
def test_batch_download_complete_coverage_pack_respects_upgrade_allowlist(monkeypatch):
|
||
"""
|
||
完整覆盖任务不能让部分剧集获准的显式整包绕过分集洗版限制。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = []
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context(episode_list=[1, 2, 3])
|
||
context.allowed_episodes = {3}
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[],
|
||
total_episode=3,
|
||
require_complete_coverage=True,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == []
|
||
assert lefts == no_exists
|
||
chain.download_single.assert_not_called()
|
||
|
||
|
||
def test_batch_download_skips_unlabelled_full_pack_outside_upgrade_allowlist(monkeypatch):
|
||
"""
|
||
标题无集号的整季包也不能绕过分集洗版的允许集下载已达上限剧集。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = [1, 2, 3]
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
context.allowed_episodes = {3}
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(season=1, episodes=[], total_episode=3),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == []
|
||
assert lefts == no_exists
|
||
chain.download_torrent.assert_not_called()
|
||
chain.download_single.assert_not_called()
|
||
|
||
|
||
@pytest.mark.parametrize("torrent_episodes, downloaded", [([1, 2, 3], True), ([1, 2, 3, 4], False)])
|
||
def test_batch_download_unlabelled_full_pack_checks_actual_episode_allowlist(
|
||
monkeypatch, torrent_episodes, downloaded
|
||
):
|
||
"""
|
||
整季目标均获准时可下载，但种子文件多出未获准剧集时须跳过。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = torrent_episodes
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
context.allowed_episodes = {1, 2, 3}
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(season=1, episodes=[], total_episode=3),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == ([context] if downloaded else [])
|
||
assert lefts == ({} if downloaded else no_exists)
|
||
assert chain.download_single.call_count == int(downloaded)
|
||
|
||
|
||
def test_batch_download_keeps_count_check_without_complete_coverage(monkeypatch):
|
||
"""
|
||
普通整季缺失仍沿用数量判断，避免完整覆盖语义影响非严格场景。
|
||
"""
|
||
_FakeBatchTorrentHelper.episodes = list(range(2, 145))
|
||
monkeypatch.setattr(download_batch, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(download_selection, "TorrentHelper", _FakeBatchTorrentHelper)
|
||
monkeypatch.setattr(eventmanager, "send_event", lambda *args, **kwargs: None)
|
||
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.eventmanager = MagicMock()
|
||
chain.eventmanager.send_event.return_value = None
|
||
chain.download_torrent = MagicMock(return_value=(b"torrent-content", "", ["demo.mkv"]))
|
||
chain.download_single = MagicMock(return_value="hash")
|
||
|
||
context = _build_tv_context()
|
||
no_exists = _tv_no_exists(
|
||
1,
|
||
NotExistMediaInfo(
|
||
season=1,
|
||
episodes=[],
|
||
total_episode=143,
|
||
start_episode=1,
|
||
),
|
||
)
|
||
|
||
downloads, lefts = chain.batch_download(contexts=[context], no_exists=no_exists)
|
||
|
||
assert downloads == [context]
|
||
assert lefts == {}
|
||
assert context.confirmed_full_coverage is False
|
||
chain.download_single.assert_called_once()
|
||
|
||
|
||
def test_downloading_includes_media_type_and_source_site(monkeypatch):
|
||
"""
|
||
正在下载任务应从下载历史回填媒体类型和来源站点。
|
||
"""
|
||
torrent = DownloaderTorrent(hash="download-hash", title="Demo.Release")
|
||
history = DownloadHistorySnapshot(
|
||
id=1,
|
||
path="/downloads/Demo.Release.mkv",
|
||
episodes="E02",
|
||
image="https://images.example.com/backdrop.jpg",
|
||
poster="https://images.example.com/poster.jpg",
|
||
seasons="S01",
|
||
title="示例剧集",
|
||
media_source=MediaSource.TMDB,
|
||
media_id="1001",
|
||
torrent_site="示例站点",
|
||
type="电视剧",
|
||
userid="user-1",
|
||
username="tester",
|
||
)
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
monkeypatch.setattr(chain, "list_torrents", lambda **_kwargs: [torrent])
|
||
chain.download_history_repository = SimpleNamespace(
|
||
get_by_hashes=lambda _hashes: {torrent.hash: history}
|
||
)
|
||
|
||
result = chain.downloading(name="qb-main")
|
||
|
||
assert result == [torrent]
|
||
assert torrent.media is not None
|
||
assert torrent.media.type == "电视剧"
|
||
assert torrent.media.image == "https://images.example.com/poster.jpg"
|
||
assert torrent.media.poster == "https://images.example.com/poster.jpg"
|
||
assert torrent.media.backdrop == "https://images.example.com/backdrop.jpg"
|
||
assert torrent.site_name == "示例站点"
|
||
assert torrent.userid == "user-1"
|
||
assert torrent.username == "tester"
|
||
|
||
|
||
def _prepared_download_for_settle(download_dir: Path) -> download_submission._PreparedDownload:
|
||
"""
|
||
构造仅用于下载结算路径断言的准备事实。
|
||
"""
|
||
file_name = "The.Ordinary.Jackpot.S01E01.1080p.mkv"
|
||
return download_submission._PreparedDownload(
|
||
torrent=TorrentInfo(title=file_name),
|
||
media=MediaInfo(type=MediaType.TV, title="中头奖还是要上班"),
|
||
meta=MetaInfo(file_name),
|
||
torrent_content=b"torrent",
|
||
folder_name="",
|
||
file_list=[file_name],
|
||
download_dir=download_dir,
|
||
download_uri=download_dir.as_posix(),
|
||
download_episodes=None,
|
||
site_downloader="qb",
|
||
)
|
||
|
||
|
||
def _settle_accepted_download(chain, download_dir: Path, already_exists: bool = False) -> None:
|
||
"""
|
||
以固定下载上下文调用一次成功结算。
|
||
|
||
:param chain: 待结算的下载链实例
|
||
:param download_dir: 提交前解析出的预期下载目录
|
||
:param already_exists: 是否按下载器已存在同一任务结算
|
||
"""
|
||
chain._settle_accepted_download(
|
||
prepared=_prepared_download_for_settle(download_dir),
|
||
context=Context(),
|
||
episodes=None,
|
||
channel=None,
|
||
source="subscribe",
|
||
userid=None,
|
||
username="admin",
|
||
custom_words=None,
|
||
actual_downloader="qb",
|
||
download_hash="hash123",
|
||
layout="NoSubfolder",
|
||
already_exists=already_exists,
|
||
)
|
||
|
||
|
||
def test_settle_accepted_download_uses_downloader_actual_save_dir():
|
||
"""
|
||
下载器中已存在同 Hash 任务时，下载历史须落到下载器实际保存目录。
|
||
"""
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.list_torrents = MagicMock(return_value=[DownloaderTorrent(
|
||
hash="hash123",
|
||
downloader="qb",
|
||
save_path="/volume4/flow",
|
||
content_path="/volume4/flow/The.Ordinary.Jackpot.S01E01.1080p.mkv",
|
||
)])
|
||
chain._settle_download_success = MagicMock()
|
||
|
||
_settle_accepted_download(chain, Path("/volume5/PT/downloads/TV/日韩剧"))
|
||
|
||
assert chain._settle_download_success.call_args.kwargs["download_dir"] == Path("/volume4/flow")
|
||
chain.list_torrents.assert_called_once_with(hashs=["hash123"], downloader="qb")
|
||
|
||
|
||
def test_settle_accepted_download_keeps_expected_dir_without_downloader_task():
|
||
"""
|
||
下载器查询不到任务时保持预期目录，不改变既有结算行为。
|
||
"""
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.list_torrents = MagicMock(return_value=[])
|
||
chain._settle_download_success = MagicMock()
|
||
|
||
_settle_accepted_download(chain, Path("/downloads/TV"))
|
||
|
||
assert chain._settle_download_success.call_args.kwargs["download_dir"] == Path("/downloads/TV")
|
||
|
||
|
||
def test_settle_accepted_download_falls_back_when_downloader_lookup_fails():
|
||
"""
|
||
下载器查询失败时回退预期目录，不能因回读失败阻断下载结算。
|
||
"""
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.list_torrents = MagicMock(side_effect=RuntimeError("downloader unavailable"))
|
||
chain._settle_download_success = MagicMock()
|
||
|
||
_settle_accepted_download(chain, Path("/downloads/TV"))
|
||
|
||
assert chain._settle_download_success.call_args.kwargs["download_dir"] == Path("/downloads/TV")
|
||
|
||
|
||
def test_settle_accepted_download_skips_lookup_for_remote_dir():
|
||
"""
|
||
远程存储保存目录不由本地回读，保持预期值且不查询下载器。
|
||
"""
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.list_torrents = MagicMock(return_value=[])
|
||
chain._settle_download_success = MagicMock()
|
||
|
||
_settle_accepted_download(chain, Path("smb:/downloads/TV"))
|
||
|
||
assert chain._settle_download_success.call_args.kwargs["download_dir"] == Path("smb:/downloads/TV")
|
||
chain.list_torrents.assert_not_called()
|
||
|
||
|
||
def test_settle_accepted_download_marks_reuse_when_task_exists():
|
||
"""
|
||
下载器已存在同一任务时，复用事实须传递到结算，供历史跳过重复通知。
|
||
"""
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.list_torrents = MagicMock(return_value=[])
|
||
chain._settle_download_success = MagicMock()
|
||
|
||
_settle_accepted_download(chain, Path("/downloads/TV"), already_exists=True)
|
||
|
||
assert chain._settle_download_success.call_args.kwargs["already_exists"] is True
|
||
|
||
|
||
def test_settle_accepted_download_records_actual_path_for_existing_torrent():
|
||
"""
|
||
复现刷流先下载场景：下载历史与文件明细须记录下载器实际路径。
|
||
"""
|
||
chain = DownloadChain.__new__(DownloadChain)
|
||
chain.runtime_config = SimpleNamespace(media_extensions=(".mkv",))
|
||
chain.durable_event_writer = None
|
||
chain.download_history_repository = MagicMock()
|
||
chain._build_download_notification = MagicMock(return_value=None)
|
||
chain._submit_download_added_task = MagicMock()
|
||
chain.eventmanager = MagicMock()
|
||
chain.list_torrents = MagicMock(return_value=[DownloaderTorrent(
|
||
hash="hash123",
|
||
downloader="qb",
|
||
save_path="/volume4/flow",
|
||
content_path="/volume4/flow/The.Ordinary.Jackpot.S01E01.1080p.mkv",
|
||
)])
|
||
|
||
_settle_accepted_download(chain, Path("/volume5/PT/downloads/TV/日韩剧"))
|
||
|
||
history, files = chain.download_history_repository.add.call_args.args
|
||
assert history.path == "/volume4/flow/The.Ordinary.Jackpot.S01E01.1080p.mkv"
|
||
assert files[0].fullpath == "/volume4/flow/The.Ordinary.Jackpot.S01E01.1080p.mkv"
|
||
assert files[0].savepath == "/volume4/flow"
|