mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-02 21:34:32 +08:00
584 lines
20 KiB
Python
584 lines
20 KiB
Python
"""
|
|
验证文件监控事件转换、写入稳定等待和异常恢复。
|
|
"""
|
|
|
|
import threading
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from watchfiles import Change
|
|
|
|
from app.application.history import TransferHistorySnapshot
|
|
from app.monitor.dispatcher import TransferDispatcher
|
|
from app.monitor.monitor import Monitor
|
|
from app.monitor.watcher import DirectoryChangeEvent, LocalDirectoryWatcher
|
|
from app.schemas.system import TransferDirectoryConf
|
|
from app.schemas.types import MediaType
|
|
|
|
|
|
class CallbackRecorder:
|
|
"""
|
|
测试用目录监控回调记录器。
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""
|
|
初始化事件记录列表。
|
|
"""
|
|
self.events = []
|
|
self.stability_events = []
|
|
|
|
def event_handler(self, event, text: str, event_path: str, file_size: int = None):
|
|
"""
|
|
记录目录监控分发出来的事件。
|
|
:param event: 目录监控事件
|
|
:param text: 事件描述
|
|
:param event_path: 事件路径
|
|
:param file_size: 文件大小
|
|
"""
|
|
self.events.append((event, text, event_path, file_size))
|
|
|
|
def event_unstable(self, event_path: Path):
|
|
"""
|
|
记录文件长时间未稳定告警。
|
|
"""
|
|
self.stability_events.append(("unstable", event_path.as_posix()))
|
|
|
|
def event_stable(self, event_path: Path):
|
|
"""
|
|
记录文件恢复稳定通知。
|
|
"""
|
|
self.stability_events.append(("stable", event_path.as_posix()))
|
|
|
|
|
|
def _build_monitor_with_dispatcher(handle_file: MagicMock = None):
|
|
"""
|
|
构造带分发器的测试用 Monitor 骨架。
|
|
:param handle_file: 替换分发器 handle_file 的替身
|
|
:return: (Monitor 骨架, 分发器)
|
|
"""
|
|
monitor = object.__new__(Monitor)
|
|
dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={})
|
|
if handle_file is not None:
|
|
dispatcher.handle_file = handle_file
|
|
monitor._dispatcher = dispatcher
|
|
monitor._lifecycle_lock = threading.RLock()
|
|
monitor._owner_lock = threading.Lock()
|
|
monitor._work_stop_event = threading.Event()
|
|
monitor._shutdown_event = threading.Event()
|
|
monitor._closed = False
|
|
monitor._compensation_threads = {}
|
|
monitor._scheduler_shutdown_thread = None
|
|
monitor._scheduler_shutdown_succeeded = False
|
|
monitor._scheduler = None
|
|
return monitor, dispatcher
|
|
|
|
|
|
def test_handle_changes_dispatches_added_and_modified_files(tmp_path):
|
|
"""
|
|
新增和修改文件应转换成目录监控整理回调。
|
|
"""
|
|
added_file = tmp_path / "a_added.mkv"
|
|
modified_file = tmp_path / "b_modified.mkv"
|
|
skipped_dir = tmp_path / "c_dir"
|
|
added_file.write_bytes(b"added")
|
|
modified_file.write_bytes(b"modified")
|
|
skipped_dir.mkdir()
|
|
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=True)
|
|
watcher._handle_changes({
|
|
(Change.added, added_file.as_posix()),
|
|
(Change.modified, modified_file.as_posix()),
|
|
(Change.deleted, added_file.as_posix()),
|
|
(Change.added, skipped_dir.as_posix()),
|
|
})
|
|
|
|
assert len(callback.events) == 2
|
|
assert (
|
|
Change.added,
|
|
"新增",
|
|
added_file.as_posix(),
|
|
5,
|
|
) == (
|
|
callback.events[0][0].change_type,
|
|
callback.events[0][1],
|
|
callback.events[0][2],
|
|
callback.events[0][3],
|
|
)
|
|
assert (
|
|
Change.modified,
|
|
"修改",
|
|
modified_file.as_posix(),
|
|
8,
|
|
) == (
|
|
callback.events[1][0].change_type,
|
|
callback.events[1][1],
|
|
callback.events[1][2],
|
|
callback.events[1][3],
|
|
)
|
|
|
|
|
|
def test_running_watcher_waits_for_file_stability_before_dispatch(tmp_path):
|
|
"""
|
|
实际监控线程收到新增事件时，应等待文件大小和 mtime 连续稳定后再整理。
|
|
"""
|
|
movie_file = tmp_path / "movie.mkv"
|
|
movie_file.write_bytes(b"part")
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=True)
|
|
# 直接调用内部循环时没有真实线程，用替身模拟运行中的监控线程，覆盖稳定性分支。
|
|
watcher.is_alive = MagicMock(return_value=True)
|
|
|
|
watcher._handle_changes({(Change.added, movie_file.as_posix())})
|
|
assert callback.events == []
|
|
assert movie_file.as_posix() in watcher._pending_stability
|
|
|
|
watcher._pending_stability[movie_file.as_posix()]["due"] = 0
|
|
watcher._process_pending_stability()
|
|
assert callback.events == []
|
|
|
|
watcher._pending_stability[movie_file.as_posix()]["due"] = 0
|
|
watcher._process_pending_stability()
|
|
assert len(callback.events) == 1
|
|
assert callback.events[0][2] == movie_file.as_posix()
|
|
assert callback.events[0][3] == 4
|
|
|
|
|
|
def test_running_watcher_reports_long_unstable_file_and_recovery(tmp_path, monkeypatch):
|
|
"""
|
|
文件长时间写入应告警一次，并在稳定后发送恢复状态再继续整理。
|
|
"""
|
|
movie_file = tmp_path / "long-writing.mkv"
|
|
movie_file.write_bytes(b"part")
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=True)
|
|
watcher.is_alive = MagicMock(return_value=True)
|
|
|
|
watcher._handle_changes({(Change.added, movie_file.as_posix())})
|
|
pending = watcher._pending_stability[movie_file.as_posix()]
|
|
# 固定单调时钟，避免新启动的 CI 机器尚未运行到告警阈值。
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 1000.0)
|
|
pending["first_seen"] = 1000.0 - watcher.FILE_STABILITY_WARNING_SECONDS
|
|
pending["due"] = 0
|
|
watcher._process_pending_stability()
|
|
assert callback.stability_events == [("unstable", movie_file.as_posix())]
|
|
|
|
watcher._pending_stability[movie_file.as_posix()]["due"] = 0
|
|
watcher._process_pending_stability()
|
|
|
|
assert callback.stability_events == [
|
|
("unstable", movie_file.as_posix()),
|
|
("stable", movie_file.as_posix()),
|
|
]
|
|
assert len(callback.events) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("suffix", [".!qB", ".!QB", ".part", ".PART", ".aria2"])
|
|
def test_running_watcher_ignores_download_temporary_files(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, suffix: str
|
|
) -> None:
|
|
"""
|
|
配置的临时下载后缀不进入稳定性队列，长期写入也不应告警。
|
|
|
|
:param tmp_path: 隔离的临时下载目录
|
|
:param monkeypatch: 替换运行配置、监控线程状态和单调时钟
|
|
:param suffix: 下载器使用的临时后缀，包含大小写和自定义后缀
|
|
"""
|
|
monkeypatch.setattr("app.monitor.watcher.get_runtime_setting", lambda _: [".!qb", ".part", ".aria2"])
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 1000.0)
|
|
temporary_file = tmp_path / f"episode.mp4{suffix}"
|
|
temporary_file.write_bytes(b"downloading")
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=True)
|
|
monkeypatch.setattr(watcher, "is_alive", lambda: True)
|
|
|
|
watcher._handle_changes({(Change.added, temporary_file.as_posix())})
|
|
temporary_file.write_bytes(b"still downloading")
|
|
watcher._handle_changes({(Change.modified, temporary_file.as_posix())})
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 2801.0)
|
|
watcher._process_pending_stability()
|
|
|
|
assert watcher._pending_stability == {}
|
|
assert callback.events == []
|
|
assert callback.stability_events == []
|
|
|
|
|
|
def test_running_watcher_dispatches_completed_download_after_rename(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""
|
|
临时文件下载完成并改名后，正式视频仍需稳定两次再分发一次。
|
|
|
|
:param tmp_path: 隔离的临时下载目录
|
|
:param monkeypatch: 替换临时后缀配置、监控线程状态和单调时钟
|
|
"""
|
|
monkeypatch.setattr("app.monitor.watcher.get_runtime_setting", lambda _: [".!qb"])
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 1000.0)
|
|
temporary_file = tmp_path / "episode.mp4.!qB"
|
|
completed_file = tmp_path / "episode.mp4"
|
|
temporary_file.write_bytes(b"complete")
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=True)
|
|
monkeypatch.setattr(watcher, "is_alive", lambda: True)
|
|
watcher._handle_changes({(Change.added, temporary_file.as_posix())})
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 2801.0)
|
|
watcher._process_pending_stability()
|
|
|
|
temporary_file.rename(completed_file)
|
|
watcher._handle_changes({
|
|
(Change.deleted, temporary_file.as_posix()),
|
|
(Change.added, completed_file.as_posix()),
|
|
})
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 2803.0)
|
|
watcher._process_pending_stability()
|
|
assert callback.events == []
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 2805.0)
|
|
watcher._process_pending_stability()
|
|
|
|
assert [event[2] for event in callback.events] == [completed_file.as_posix()]
|
|
assert watcher._pending_stability == {}
|
|
assert callback.stability_events == []
|
|
|
|
|
|
def test_running_watcher_rechecks_temporary_suffix_for_pending_file(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""
|
|
临时后缀配置热更新后，已入队的文件不应继续产生超时告警。
|
|
|
|
:param tmp_path: 隔离的临时下载目录
|
|
:param monkeypatch: 替换运行配置、监控线程状态和单调时钟
|
|
"""
|
|
temporary_extensions: list[str] = []
|
|
monkeypatch.setattr("app.monitor.watcher.get_runtime_setting", lambda _: temporary_extensions)
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 1000.0)
|
|
temporary_file = tmp_path / "episode.mp4.!qB"
|
|
temporary_file.write_bytes(b"downloading")
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=True)
|
|
monkeypatch.setattr(watcher, "is_alive", lambda: True)
|
|
watcher._handle_changes({(Change.added, temporary_file.as_posix())})
|
|
assert temporary_file.as_posix() in watcher._pending_stability
|
|
|
|
temporary_extensions.append(".!qb")
|
|
monkeypatch.setattr("app.monitor.watcher.time.monotonic", lambda: 2801.0)
|
|
watcher._process_pending_stability()
|
|
|
|
assert watcher._pending_stability == {}
|
|
assert callback.events == []
|
|
assert callback.stability_events == []
|
|
|
|
|
|
def test_running_watcher_filters_temporary_files_in_added_directory(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""
|
|
新增目录递归展开时跳过临时文件，并保留正式视频的稳定性检测。
|
|
|
|
:param tmp_path: 隔离的目录树
|
|
:param monkeypatch: 替换监控运行配置和线程状态
|
|
"""
|
|
runtime_settings = {"DOWNLOAD_TMPEXT": [".!qb"], "MONITOR_RESCAN_DELAYS": "30,120"}
|
|
monkeypatch.setattr("app.monitor.watcher.get_runtime_setting", runtime_settings.__getitem__)
|
|
added_dir = tmp_path / "season"
|
|
added_dir.mkdir()
|
|
temporary_file = added_dir / "episode.mp4.!qB"
|
|
completed_file = added_dir / "complete.mp4"
|
|
temporary_file.write_bytes(b"downloading")
|
|
completed_file.write_bytes(b"complete")
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=True)
|
|
monkeypatch.setattr(watcher, "is_alive", lambda: True)
|
|
|
|
watcher._handle_changes({(Change.added, added_dir.as_posix())})
|
|
|
|
assert set(watcher._pending_stability) == {completed_file.as_posix()}
|
|
assert callback.events == []
|
|
assert callback.stability_events == []
|
|
|
|
|
|
def test_handle_changes_skips_missing_paths(tmp_path):
|
|
"""
|
|
事件到达时已经消失的路径不应触发整理。
|
|
"""
|
|
missing_file = tmp_path / "missing.mkv"
|
|
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=True)
|
|
watcher._handle_changes({(Change.added, missing_file.as_posix())})
|
|
|
|
assert callback.events == []
|
|
|
|
|
|
def test_handle_changes_expands_added_directory_files(tmp_path):
|
|
"""
|
|
整体移入的新增目录应递归转换成内部文件事件且不重复分发。
|
|
"""
|
|
added_dir = tmp_path / "task"
|
|
nested_dir = added_dir / "season"
|
|
nested_dir.mkdir(parents=True)
|
|
movie_file = added_dir / "movie.mkv"
|
|
episode_file = nested_dir / "episode.mkv"
|
|
ignored_file = added_dir / ".DS_Store"
|
|
movie_file.write_bytes(b"movie")
|
|
episode_file.write_bytes(b"episode")
|
|
ignored_file.write_bytes(b"ignored")
|
|
|
|
callback = CallbackRecorder()
|
|
watcher = LocalDirectoryWatcher(tmp_path, callback=callback, force_polling=False)
|
|
watcher._handle_changes({
|
|
(Change.added, added_dir.as_posix()),
|
|
(Change.added, movie_file.as_posix()),
|
|
})
|
|
|
|
assert [
|
|
(event.change_type, event_path, file_size)
|
|
for event, _, event_path, file_size in callback.events
|
|
] == [
|
|
(Change.added, movie_file.as_posix(), 5),
|
|
(Change.added, episode_file.as_posix(), 7),
|
|
]
|
|
|
|
|
|
def test_event_handler_routes_file_events_to_transfer_handler():
|
|
"""
|
|
文件事件应继续按 local 存储交给整理流程。
|
|
"""
|
|
handle_file = MagicMock()
|
|
monitor, _ = _build_monitor_with_dispatcher(handle_file)
|
|
event_path = Path("/downloads/movie.mkv")
|
|
event = DirectoryChangeEvent(
|
|
change_type=Change.added,
|
|
src_path=event_path.as_posix(),
|
|
is_directory=False
|
|
)
|
|
|
|
monitor.event_handler(
|
|
event=event,
|
|
text="新增",
|
|
event_path=event_path.as_posix(),
|
|
file_size=1024
|
|
)
|
|
|
|
handle_file.assert_called_once_with(
|
|
storage="local",
|
|
event_path=event_path,
|
|
file_size=1024
|
|
)
|
|
|
|
|
|
def test_event_handler_ignores_directory_events():
|
|
"""
|
|
目录事件不应进入文件整理流程。
|
|
"""
|
|
handle_file = MagicMock()
|
|
monitor, _ = _build_monitor_with_dispatcher(handle_file)
|
|
event_path = Path("/downloads/folder")
|
|
event = DirectoryChangeEvent(
|
|
change_type=Change.added,
|
|
src_path=event_path.as_posix(),
|
|
is_directory=True
|
|
)
|
|
|
|
monitor.event_handler(
|
|
event=event,
|
|
text="新增",
|
|
event_path=event_path.as_posix()
|
|
)
|
|
|
|
handle_file.assert_not_called()
|
|
|
|
|
|
def test_event_handler_ignores_download_temp_files():
|
|
"""
|
|
下载器临时文件不应进入整理流程。
|
|
"""
|
|
handle_file = MagicMock()
|
|
monitor, _ = _build_monitor_with_dispatcher(handle_file)
|
|
event_path = Path("/downloads/movie.mkv.!qB")
|
|
event = DirectoryChangeEvent(
|
|
change_type=Change.modified,
|
|
src_path=event_path.as_posix(),
|
|
is_directory=False
|
|
)
|
|
|
|
monitor.event_handler(
|
|
event=event,
|
|
text="修改",
|
|
event_path=event_path.as_posix(),
|
|
file_size=1024
|
|
)
|
|
|
|
handle_file.assert_not_called()
|
|
|
|
|
|
def test_event_handler_ignores_non_transferable_files():
|
|
"""
|
|
非可整理后缀文件不应进入整理流程。
|
|
"""
|
|
handle_file = MagicMock()
|
|
monitor, _ = _build_monitor_with_dispatcher(handle_file)
|
|
event_path = Path("/downloads/movie.nfo")
|
|
event = DirectoryChangeEvent(
|
|
change_type=Change.added,
|
|
src_path=event_path.as_posix(),
|
|
is_directory=False
|
|
)
|
|
|
|
monitor.event_handler(
|
|
event=event,
|
|
text="新增",
|
|
event_path=event_path.as_posix(),
|
|
file_size=1024
|
|
)
|
|
|
|
handle_file.assert_not_called()
|
|
|
|
|
|
def test_handle_file_skips_transfer_when_history_exists(monkeypatch):
|
|
"""
|
|
已有整理记录的源文件不应再次进入整理链。
|
|
"""
|
|
dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={})
|
|
event_path = Path("/downloads/movie.mkv")
|
|
lookups = []
|
|
|
|
class FakeTransferHistoryOper:
|
|
"""
|
|
测试用整理历史查询。
|
|
"""
|
|
|
|
def get_by_src(self, src: str, storage: str = None):
|
|
"""
|
|
记录查询参数并返回已有记录。
|
|
"""
|
|
lookups.append((src, storage))
|
|
return TransferHistorySnapshot(
|
|
id=1,
|
|
src=src,
|
|
src_storage=storage,
|
|
src_fileitem={"size": 1024},
|
|
status=True,
|
|
)
|
|
|
|
def get_success_by_src(self, src: str, storage: str = None):
|
|
"""
|
|
成功记录已由首次查询返回，无需二次回退。
|
|
"""
|
|
return None
|
|
|
|
transfer_chain = MagicMock()
|
|
logger_info = MagicMock()
|
|
logger_debug = MagicMock()
|
|
monkeypatch.setattr(
|
|
"app.monitor.dispatcher.get_transfer_history_repository",
|
|
FakeTransferHistoryOper,
|
|
)
|
|
monkeypatch.setattr("app.monitor.dispatcher.TransferChain", transfer_chain)
|
|
monkeypatch.setattr("app.monitor.dispatcher.logger.info", logger_info)
|
|
monkeypatch.setattr("app.monitor.dispatcher.logger.debug", logger_debug)
|
|
|
|
handled = dispatcher.handle_file(
|
|
storage="local",
|
|
event_path=event_path,
|
|
file_size=1024,
|
|
)
|
|
|
|
assert not handled
|
|
assert lookups == [(event_path.as_posix(), "local")]
|
|
transfer_chain.assert_not_called()
|
|
logger_info.assert_not_called()
|
|
assert "已整理过且文件未变化" in logger_debug.call_args.args[0]
|
|
|
|
|
|
def test_handle_file_invokes_transfer_when_history_missing(monkeypatch):
|
|
"""
|
|
没有整理记录的源文件应继续进入整理链。
|
|
"""
|
|
dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={})
|
|
event_path = Path("/downloads/movie.mkv")
|
|
|
|
class FakeTransferHistoryOper:
|
|
"""
|
|
测试用空整理历史查询。
|
|
"""
|
|
|
|
def get_by_src(self, src: str, storage: str = None):
|
|
"""
|
|
返回空整理记录。
|
|
"""
|
|
return None
|
|
|
|
def get_success_by_src(self, src: str, storage: str = None):
|
|
"""
|
|
空仓储没有成功历史。
|
|
"""
|
|
return None
|
|
|
|
transfer_chain_instance = MagicMock()
|
|
transfer_chain = MagicMock(return_value=transfer_chain_instance)
|
|
monkeypatch.setattr(
|
|
"app.monitor.dispatcher.get_transfer_history_repository",
|
|
FakeTransferHistoryOper,
|
|
)
|
|
monkeypatch.setattr("app.monitor.dispatcher.TransferChain", transfer_chain)
|
|
|
|
handled = dispatcher.handle_file(
|
|
storage="local",
|
|
event_path=event_path,
|
|
file_size=1024,
|
|
)
|
|
|
|
assert handled
|
|
transfer_chain_instance.do_transfer.assert_called_once()
|
|
fileitem = transfer_chain_instance.do_transfer.call_args.kwargs["fileitem"]
|
|
assert fileitem.storage == "local"
|
|
assert fileitem.path == event_path.as_posix()
|
|
assert fileitem.size == 1024
|
|
|
|
|
|
def test_handle_file_prefers_music_type_from_monitor_directory(monkeypatch):
|
|
"""
|
|
音乐目录监控触发整理时应透传音乐类型，避免音频按影视名称识别。
|
|
"""
|
|
dispatcher = TransferDispatcher(all_exts=[".flac"], cache={})
|
|
event_path = Path("/downloads/music/album/track.flac")
|
|
directories = [
|
|
TransferDirectoryConf(
|
|
storage="local",
|
|
download_path="/downloads",
|
|
media_type=MediaType.MOVIE.value,
|
|
monitor_type="monitor",
|
|
),
|
|
TransferDirectoryConf(
|
|
storage="local",
|
|
download_path="/downloads/music",
|
|
media_type=MediaType.MUSIC.value,
|
|
monitor_type="monitor",
|
|
),
|
|
]
|
|
transfer_chain_instance = MagicMock()
|
|
|
|
# 历史查重已由 _should_skip_by_history 统一承担（含失败重试预算与版本变化判定），
|
|
# 这里放行以便验证 mtype 的传递
|
|
monkeypatch.setattr(dispatcher, "_should_skip_by_history", MagicMock(return_value=False))
|
|
monkeypatch.setattr(
|
|
"app.monitor.dispatcher.DirectoryHelper",
|
|
MagicMock(return_value=MagicMock(get_download_dirs=MagicMock(return_value=directories))),
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.monitor.dispatcher.TransferChain",
|
|
MagicMock(return_value=transfer_chain_instance),
|
|
)
|
|
|
|
handled = dispatcher.handle_file(
|
|
storage="local",
|
|
event_path=event_path,
|
|
file_size=1024,
|
|
)
|
|
|
|
assert handled
|
|
assert transfer_chain_instance.do_transfer.call_args.kwargs["mtype"] == MediaType.MUSIC
|