HeXi/tests/test_video_cleanup.py

"""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}