mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
368 lines
12 KiB
Python
368 lines
12 KiB
Python
"""
|
|
验证目录监控异常后的退避、模式降级与停止行为。
|
|
"""
|
|
|
|
import threading
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
from app.monitor.monitor import Monitor
|
|
from app.monitor.recovery import RecoveryExecutor
|
|
from app.monitor.watcher import LocalDirectoryWatcher
|
|
|
|
|
|
def _build_watcher(tmp_path, force_polling):
|
|
"""
|
|
构造测试用目录监控。
|
|
:param tmp_path: 监控目录
|
|
:param force_polling: 是否强制轮询
|
|
:return: 目录监控
|
|
"""
|
|
return LocalDirectoryWatcher(tmp_path, callback=MagicMock(), force_polling=force_polling)
|
|
|
|
|
|
def test_run_retries_with_backoff_in_compatibility_mode(tmp_path, monkeypatch):
|
|
"""
|
|
兼容模式下监控循环抛异常后应退避重启，而不是直接结束线程。
|
|
"""
|
|
monkeypatch.setattr(LocalDirectoryWatcher, "RESTART_BACKOFF", (0,))
|
|
watcher = _build_watcher(tmp_path, force_polling=True)
|
|
calls = []
|
|
|
|
def fake_run_watch(force_polling):
|
|
"""
|
|
模拟底层监控循环持续抛出 FUSE 错误。
|
|
"""
|
|
calls.append(force_polling)
|
|
if len(calls) >= 3:
|
|
watcher.stop()
|
|
raise OSError(131, "State not recoverable")
|
|
|
|
monkeypatch.setattr(watcher, "_run_watch", fake_run_watch)
|
|
|
|
watcher._run()
|
|
|
|
assert calls == [True, True, True]
|
|
assert watcher.restart_count == 2
|
|
|
|
|
|
def test_run_falls_back_to_polling_before_backoff(tmp_path, monkeypatch):
|
|
"""
|
|
快速模式失败应先降级为兼容模式重试，且降级不计入退避重启次数。
|
|
"""
|
|
monkeypatch.setattr(LocalDirectoryWatcher, "RESTART_BACKOFF", (0,))
|
|
watcher = _build_watcher(tmp_path, force_polling=None)
|
|
calls = []
|
|
|
|
def fake_run_watch(force_polling):
|
|
"""
|
|
模拟快速模式与兼容模式先后失败。
|
|
"""
|
|
calls.append(force_polling)
|
|
if len(calls) >= 2:
|
|
watcher.stop()
|
|
raise OSError("inotify watch limit reached")
|
|
|
|
monkeypatch.setattr(watcher, "_run_watch", fake_run_watch)
|
|
|
|
watcher._run()
|
|
|
|
assert calls == [None, True]
|
|
assert watcher.restart_count == 0
|
|
|
|
|
|
def test_run_returns_when_stop_requested(tmp_path, monkeypatch):
|
|
"""
|
|
收到停止信号后监控循环正常返回，不应触发重启。
|
|
"""
|
|
watcher = _build_watcher(tmp_path, force_polling=True)
|
|
calls = []
|
|
|
|
def fake_run_watch(force_polling):
|
|
"""
|
|
模拟收到停止信号后正常退出的监控循环。
|
|
"""
|
|
calls.append(force_polling)
|
|
|
|
monkeypatch.setattr(watcher, "_run_watch", fake_run_watch)
|
|
|
|
watcher._run()
|
|
|
|
assert calls == [True]
|
|
assert watcher.restart_count == 0
|
|
|
|
|
|
def test_is_stalled_detects_silent_failure(tmp_path):
|
|
"""
|
|
监控线程存活但长时间无活动时应判定为静默失效。
|
|
"""
|
|
watcher = _build_watcher(tmp_path, force_polling=True)
|
|
|
|
# 线程未启动时不做判定
|
|
assert watcher.is_stalled() is False
|
|
|
|
thread = MagicMock()
|
|
thread.is_alive.return_value = True
|
|
watcher._thread = thread
|
|
watcher._mark_activity()
|
|
assert watcher.is_stalled() is False
|
|
|
|
watcher._last_activity -= LocalDirectoryWatcher.STALL_TIMEOUT + 1
|
|
assert watcher.is_stalled() is True
|
|
|
|
|
|
def test_is_stalled_ignores_stopped_watcher(tmp_path):
|
|
"""
|
|
已请求停止的监控不应再被判定为静默失效。
|
|
"""
|
|
watcher = _build_watcher(tmp_path, force_polling=True)
|
|
thread = MagicMock()
|
|
thread.is_alive.return_value = True
|
|
watcher._thread = thread
|
|
watcher._mark_activity()
|
|
watcher._last_activity -= LocalDirectoryWatcher.STALL_TIMEOUT + 1
|
|
|
|
watcher.stop()
|
|
|
|
assert watcher.is_stalled() is False
|
|
|
|
|
|
def _build_monitor(monkeypatch, put_recorder):
|
|
"""
|
|
构造测试用 Monitor 骨架，绕过单例初始化。
|
|
:param monkeypatch: pytest monkeypatch
|
|
:param put_recorder: 消息推送记录器
|
|
:return: Monitor 骨架
|
|
"""
|
|
from threading import Lock
|
|
monkeypatch.setattr("app.monitor.monitor.MessageHelper", MagicMock(return_value=put_recorder))
|
|
monitor = object.__new__(Monitor)
|
|
# 自动重启后健康检查会发起补偿扫描，骨架需要一个分发器替身
|
|
monitor._dispatcher = MagicMock()
|
|
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
|
|
monitor._watchers = []
|
|
monitor._retired_watchers = []
|
|
monitor._watcher_lock = Lock()
|
|
monitor._pending_locals = []
|
|
monitor._alerted_paths = {}
|
|
monitor._restart_marks = {}
|
|
monitor._stable_cycles = {}
|
|
monitor._isolated = {}
|
|
monitor._pending_rebuild = {}
|
|
monitor._recovery = RecoveryExecutor()
|
|
return monitor
|
|
|
|
|
|
def _fake_watcher(mon_path, alive=True, stalled=False, restart_count=0):
|
|
"""
|
|
构造测试用监控线程替身。
|
|
:param mon_path: 监控目录
|
|
:param alive: 线程是否存活
|
|
:param stalled: 是否静默失效
|
|
:param restart_count: 自动重启次数
|
|
:return: 监控线程替身
|
|
"""
|
|
watcher = MagicMock()
|
|
watcher.watch_path = mon_path
|
|
watcher.is_alive.return_value = alive
|
|
watcher.is_stalled.return_value = stalled
|
|
watcher.restart_count = restart_count
|
|
return watcher
|
|
|
|
|
|
def test_watchdog_rebuilds_dead_watcher(tmp_path, monkeypatch):
|
|
"""
|
|
监控线程退出后健康检查应判定为待重建并告警。
|
|
|
|
重建本身会触碰挂载，已移出看门狗线程，因此检测环节只负责判定与告警，
|
|
真正的重建由 __drive_recovery 派发到一次性工作线程执行。
|
|
"""
|
|
put_recorder = MagicMock()
|
|
monitor = _build_monitor(monkeypatch, put_recorder)
|
|
watcher = _fake_watcher(tmp_path, alive=False)
|
|
monitor._watchers = [watcher]
|
|
|
|
assert monitor._Monitor__check_watchers() == [watcher]
|
|
|
|
put_recorder.put.assert_called_once()
|
|
|
|
|
|
def test_watchdog_rebuilds_stalled_watcher(tmp_path, monkeypatch):
|
|
"""
|
|
静默失效的监控线程也应被健康检查判定为待重建。
|
|
"""
|
|
put_recorder = MagicMock()
|
|
monitor = _build_monitor(monkeypatch, put_recorder)
|
|
watcher = _fake_watcher(tmp_path, alive=True, stalled=True)
|
|
monitor._watchers = [watcher]
|
|
|
|
assert monitor._Monitor__check_watchers() == [watcher]
|
|
|
|
|
|
def test_watchdog_alerts_on_restart_and_recovers_after_stable_window(tmp_path, monkeypatch):
|
|
"""
|
|
自动重启应触发一次告警，恢复消息需等满稳定窗口，避免来回刷屏。
|
|
"""
|
|
put_recorder = MagicMock()
|
|
monitor = _build_monitor(monkeypatch, put_recorder)
|
|
watcher = _fake_watcher(tmp_path, alive=True, stalled=False, restart_count=1)
|
|
monitor._watchers = [watcher]
|
|
|
|
monitor._Monitor__check_watchers()
|
|
assert str(tmp_path) in monitor._alerted_paths
|
|
assert put_recorder.put.call_count == 1
|
|
|
|
for _ in range(Monitor.RECOVERY_STABLE_CYCLES - 1):
|
|
monitor._Monitor__check_watchers()
|
|
assert str(tmp_path) in monitor._alerted_paths
|
|
|
|
monitor._Monitor__check_watchers()
|
|
assert str(tmp_path) not in monitor._alerted_paths
|
|
assert put_recorder.put.call_count == 2
|
|
|
|
|
|
def test_retry_pending_locals_backs_off(tmp_path, monkeypatch):
|
|
"""
|
|
启动失败的监控重试应按失败次数退避，避免持续故障时刷屏。
|
|
"""
|
|
put_recorder = MagicMock()
|
|
monitor = _build_monitor(monkeypatch, put_recorder)
|
|
monitor._pending_locals = [{"mon_path": tmp_path, "monitor_mode": "compatibility"}]
|
|
start = MagicMock(return_value=False)
|
|
setattr(monitor, "_Monitor__start_local_monitor", start)
|
|
|
|
for _ in range(6):
|
|
monitor._Monitor__retry_pending_locals()
|
|
|
|
assert start.call_count == 3
|
|
|
|
|
|
def test_dispatcher_retries_after_history_query_failure(monkeypatch):
|
|
"""
|
|
整理历史查询失败应登记待重试，重试成功后进入整理链并清除登记。
|
|
"""
|
|
from app.monitor.dispatcher import TransferDispatcher
|
|
dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={})
|
|
event_path = Path("/downloads/movie.mkv")
|
|
history = MagicMock(side_effect=[None, False])
|
|
monkeypatch.setattr(dispatcher, "_should_skip_by_history", history)
|
|
transfer_chain_instance = MagicMock()
|
|
monkeypatch.setattr("app.monitor.dispatcher.TransferChain",
|
|
MagicMock(return_value=transfer_chain_instance))
|
|
|
|
# 首次查询失败：不整理，登记待重试
|
|
assert dispatcher.handle_file(storage="local", event_path=event_path, file_size=1) is False
|
|
assert len(dispatcher._pending_retries) == 1
|
|
transfer_chain_instance.do_transfer.assert_not_called()
|
|
|
|
# 模拟 TTL 缓存过期后由健康检查驱动重试
|
|
dispatcher._cache.clear()
|
|
dispatcher.retry_pending()
|
|
|
|
transfer_chain_instance.do_transfer.assert_called_once()
|
|
assert dispatcher._pending_retries == {}
|
|
|
|
|
|
def test_dispatcher_clear_pending_drops_stale_entries(monkeypatch):
|
|
"""
|
|
停止/配置重载时应清空待重试队列，避免已移除的监控目录在数据库恢复后
|
|
仍被送入整理链。
|
|
"""
|
|
from app.monitor.dispatcher import TransferDispatcher
|
|
dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={})
|
|
monkeypatch.setattr(dispatcher, "_should_skip_by_history", MagicMock(return_value=None))
|
|
dispatcher.handle_file(storage="local", event_path=Path("/removed/movie.mkv"), file_size=1)
|
|
assert dispatcher._pending_retries
|
|
|
|
dispatcher.clear_pending()
|
|
|
|
assert dispatcher._pending_retries == {}
|
|
dispatcher.retry_pending()
|
|
assert dispatcher._pending_retries == {}
|
|
|
|
|
|
def test_monitor_stop_clears_dispatcher_pending(monkeypatch):
|
|
"""
|
|
Monitor.stop 应连带清理分发器的待重试队列。
|
|
"""
|
|
put_recorder = MagicMock()
|
|
monitor = _build_monitor(monkeypatch, put_recorder)
|
|
monitor._scheduler = None
|
|
monitor._dispatcher = MagicMock()
|
|
|
|
monitor.stop()
|
|
|
|
monitor._dispatcher.clear_pending.assert_called_once()
|
|
|
|
|
|
def test_dispatcher_drops_pending_after_max_attempts(monkeypatch):
|
|
"""
|
|
历史查询持续失败达到上限后应放弃重试，避免队列无限累积。
|
|
"""
|
|
from app.monitor.dispatcher import TransferDispatcher
|
|
abandoned = MagicMock()
|
|
dispatcher = TransferDispatcher(
|
|
all_exts=[".mkv"],
|
|
cache={},
|
|
retry_abandoned_callback=abandoned,
|
|
)
|
|
event_path = Path("/downloads/movie.mkv")
|
|
monkeypatch.setattr(dispatcher, "_should_skip_by_history", MagicMock(return_value=None))
|
|
|
|
dispatcher.handle_file(storage="local", event_path=event_path, file_size=1)
|
|
key = f"local:{event_path.as_posix()}"
|
|
assert key in dispatcher._pending_retries
|
|
dispatcher._pending_retries[key]["attempts"] = TransferDispatcher.MAX_RETRY_ATTEMPTS - 1
|
|
|
|
dispatcher._cache.clear()
|
|
dispatcher.retry_pending()
|
|
|
|
assert dispatcher._pending_retries == {}
|
|
abandoned.assert_called_once_with(
|
|
"local",
|
|
event_path,
|
|
f"整理历史查询失败连续失败 {TransferDispatcher.MAX_RETRY_ATTEMPTS} 次",
|
|
)
|
|
|
|
|
|
def test_dispatcher_reports_retry_queue_capacity_drop():
|
|
"""
|
|
待重试队列已满时必须通知用户被丢弃的具体文件和恢复原因。
|
|
"""
|
|
from app.monitor.dispatcher import TransferDispatcher
|
|
|
|
abandoned = MagicMock()
|
|
dispatcher = TransferDispatcher(
|
|
all_exts=[".mkv"],
|
|
cache={},
|
|
retry_abandoned_callback=abandoned,
|
|
)
|
|
dispatcher.MAX_PENDING_RETRIES = 1
|
|
dispatcher._register_pending("local", Path("/downloads/first.mkv"))
|
|
dispatcher._register_pending("local", Path("/downloads/dropped.mkv"))
|
|
|
|
abandoned.assert_called_once_with(
|
|
"local",
|
|
Path("/downloads/dropped.mkv"),
|
|
"整理重试队列已达到 1 条上限",
|
|
)
|
|
|
|
|
|
def test_monitor_watches_mode_env_keys_for_hot_reload():
|
|
"""
|
|
快速模式/轮询间隔只在监控线程创建时读取,必须监听对应 env 变更触发
|
|
init() 重建才能热生效;防止上游合并时丢失监听键。
|
|
"""
|
|
from app.schemas.types import SystemConfigKey
|
|
assert SystemConfigKey.Directories.value in Monitor.CONFIG_WATCH
|
|
assert "MONITOR_NETWORK_FAST_MODE" in Monitor.CONFIG_WATCH
|
|
assert "MONITOR_POLL_DELAY_NETWORK" in Monitor.CONFIG_WATCH
|