"""temp 目录清理逻辑（video_analysis/cleanup.py）单元测试 这组函数现在同时被「/清理temp 命令」「每日定时」「Web 缓存面板的手动清理」 三条路复用，是**真的会删文件**的代码，所以单测覆盖"删什么/不删什么"。 cleanup.py 有插件内相对导入（`.utils`）和 import 期副作用（`get_plugin_config` 建配置对象、`_register_auto_job()` 注册定时任务），无法裸加载 —— 按 test_video_sender_link.py 的办法：sys.modules 预置替身 + monkeypatch nonebot 的 几个工厂函数，再按路径加载目标模块。 文件一律用 tmp_path 造真实文件/目录 + `os.utime` 改 mtime，**不 mock 文件系统**： 这些判断（mtime 过期、目录为空、60s 目录下限）全在真实 stat 语义上。 """ import importlib.util import os import sys import time import types from pathlib import Path import pytest _REPO = Path(__file__).resolve().parents[1] _PLUGIN = _REPO / "hexi" / "plugins" / "nonebot_plugin_video_analysis" _PKG = "hexi.plugins.nonebot_plugin_video_analysis" _STUBS = { "hexi": {}, "hexi.core": {}, "hexi.core.custom_utils": {"check_manage": None}, "hexi.core.scheduler": {"add_job": lambda *a, **k: None}, # cleanup.py 现在把清理提交到任务槽；这里只测 clean_temp_files / temp_stats， # 槽本身由 tests/test_task_slots.py 覆盖，所以给个够 import 的替身即可。 "hexi.core.task_slots": {"TASKS": None, "Task": type("Task", (), {})}, "nonebot_plugin_apscheduler": {"scheduler": types.SimpleNamespace(remove_job=None)}, "nonebot_plugin_alconna": {"UniMessage": type("UniMessage", (), {})}, } class _FakeMatcher: """`on_command(...)` 的替身：`.handle()` 当装饰器用。""" def handle(self, *args, **kwargs): def deco(func): return func return deco class _TempRoot: """`utils.get_temp_root` 的替身：指向测试用的临时根。""" def __init__(self) -> None: self.root: Path | None = None def __call__(self, sub: str = "") -> Path: assert self.root is not None, "测试里要先 sched.get_temp_root.root = ..." return self.root / sub if sub else self.root @pytest.fixture def cleanup(monkeypatch, tmp_path): saved = {name: sys.modules.get(name) for name in _STUBS} for name, attrs in _STUBS.items(): module = types.ModuleType(name) for key, value in attrs.items(): setattr(module, key, value) sys.modules[name] = module # 插件包本体也用替身，避免触发 __init__ 的 NoneBot 初始化 saved[_PKG] = sys.modules.get(_PKG) sys.modules[_PKG] = types.ModuleType(_PKG) temp_root = _TempRoot() utils_mod = types.ModuleType(f"{_PKG}.utils") utils_mod.get_temp_root = temp_root saved[f"{_PKG}.utils"] = sys.modules.get(f"{_PKG}.utils") sys.modules[f"{_PKG}.utils"] = utils_mod import nonebot monkeypatch.setattr(nonebot, "require", lambda name: None, raising=False) monkeypatch.setattr( nonebot, "get_plugin_config", lambda cls: cls(), raising=False ) # 默认配置：auto / 7 天 / 4 点 monkeypatch.setattr(nonebot, "on_command", lambda *a, **k: _FakeMatcher(), raising=False) target = f"{_PKG}.cleanup" saved[target] = sys.modules.get(target) spec = importlib.util.spec_from_file_location(target, _PLUGIN / "cleanup.py") module = importlib.util.module_from_spec(spec) sys.modules[target] = module spec.loader.exec_module(module) temp_root.root = tmp_path / "temp" temp_root.root.mkdir() module._tmp = temp_root.root # 方便测试直接拿到根 yield module for name, previous in saved.items(): if previous is None: sys.modules.pop(name, None) else: sys.modules[name] = previous def _make(root: Path, rel: str, age_days: float = 0.0, data: bytes = b"x") -> Path: """造一个文件，并把 mtime 往回拨 age_days 天。""" path = root / rel path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) if age_days: t = time.time() - age_days * 86400 os.utime(path, (t, t)) return path # ───────────────────────── 清理 ───────────────────────── def test_removes_stale_keeps_fresh(cleanup): root = cleanup._tmp stale = _make(root, "douyin/作者_1/旧作品.mp4", age_days=2) fresh = _make(root, "douyin/作者_1/新作品.mp4", age_days=0) removed, total = cleanup.clean_temp_files(days=1) assert (removed, total) == (1, 2) assert not stale.exists() assert fresh.exists() def test_days_zero_clears_files_but_keeps_fresh_dirs(cleanup): """`/清理temp 0`（= Web 的「全部清空」）：文件全删，但刚建的目录留着。 目录另有 60s 下限（见 `_prune_empty_dirs`）—— 下载落盘是先 mkdir 再写文件， 把目录一起删掉会让正在进行的下载失败。 """ root = cleanup._tmp a = _make(root, "kuaishou/作者_2/作品.mp4", age_days=0) b = _make(root, "xiaohongshu/作者_3/笔记/001.jpg", age_days=0) removed, total = cleanup.clean_temp_files(days=0) assert (removed, total) == (2, 2) assert not a.exists() and not b.exists() assert (root / "kuaishou/作者_2").is_dir() # 新鲜目录保留 assert (root / "xiaohongshu/作者_3/笔记").is_dir() def test_prunes_stale_empty_dirs_bottom_up(cleanup): """过期空目录会被收掉，但是**逐轮**回收而不是同一轮全清。 删文件（或删子目录）会把所在目录的 mtime 刷成"现在"，当轮它就不再"过期"， 要等下一轮 —— 这正是 `_prune_empty_dirs` 注释里写的行为（也被这条测试钉住）。 """ root = cleanup._tmp _make(root, "bilibili/作者_4/作品/001.jpg", age_days=3) nested = root / "bilibili/作者_4/作品" author = root / "bilibili/作者_4" removed, _ = cleanup.clean_temp_files(days=1) assert removed == 1 assert nested.is_dir() and author.is_dir() # 文件刚删，目录 mtime 是新的 old = time.time() - 3 * 86400 # 模拟又过了几天 os.utime(nested, (old, old)) os.utime(author, (old, old)) cleanup.clean_temp_files(days=1) assert not nested.exists() # 深的先走 os.utime(author, (old, old)) # 上一轮删子目录，又把父目录的 mtime 刷新了 cleanup.clean_temp_files(days=1) assert not author.exists() def test_keeps_stale_but_nonempty_dirs(cleanup): """过期但里面还有新鲜文件的目录不能删。""" root = cleanup._tmp _make(root, "douyin/作者_5/新片.mp4", age_days=0) old_dir = root / "douyin/作者_5" os.utime(old_dir, (time.time() - 3 * 86400,) * 2) cleanup.clean_temp_files(days=1) assert old_dir.is_dir() def test_missing_root_is_noop(cleanup): """根目录不存在（还没下载过任何东西）→ (0, 0)，不抛。""" cleanup._tmp.rmdir() assert cleanup.clean_temp_files(days=1) == (0, 0) # ───────────────────────── 统计 ───────────────────────── def test_stats_counts_files_bytes_dirs(cleanup): root = cleanup._tmp _make(root, "douyin/作者_6/a.mp4", age_days=2, data=b"12345") _make(root, "douyin/作者_6/b.mp4", age_days=0, data=b"123") _make(root, "douyin/作者_7/c.jpg", age_days=5, data=b"1") st = cleanup.temp_stats() assert st["files"] == 3 assert st["bytes"] == 9 assert st["dirs"] == 3 # douyin / 作者_6 / 作者_7（子目录各算一层） def test_stats_missing_root(cleanup): cleanup._tmp.rmdir() assert cleanup.temp_stats() == {"files": 0, "bytes": 0, "dirs": 0}