HeXi/tests/test_video_sender_link.py

"""媒体发送成功后「补传 S3 取下载链接」的单元测试

背景：2026-09-22 起 `hexi/core/outbound_media` 的全局钩子把本地路径改写成
`/media/<token>`，本地直发那一级才真正可用。此前它因协议端不在本机、`file://`
必然 ENOENT 而**恒失败**，所以每次都落到第 2 级 S3 —— `public_url` 因此总有值。
本地直发通了以后，如果什么都不做，`public_url` 会恒为 None，`entry.py` 的
`if pub and policy.sends_link` 就会让「发送下载链接」**静默失效**。

`_local_sent` / `_public_link_for` 就是为此把「取链接」与「发送」解耦：
只有策略真要链接（`sends_link`）时才补一次 S3，其余情况一概不碰 S3。

sender.py 有插件内相对导入（`..config` / `..policy` / `..services...`），
无法裸加载，故在 sys.modules 里预置这些依赖的替身再按路径加载目标模块。
"""

import importlib.util
import sys
import types
from pathlib import Path

import pytest

_REPO = Path(__file__).resolve().parents[1]
_PKG = "hexi.plugins.nonebot_plugin_video_analysis"
_SENDER_PATH = (
    _REPO
    / "hexi"
    / "plugins"
    / "nonebot_plugin_video_analysis"
    / "handlers"
    / "sender.py"
)

#: 预置的替身模块名 → 该模块要暴露的属性占位（真实函数在测试里替换）
_STUBS = {
    _PKG: {},
    f"{_PKG}.handlers": {},
    f"{_PKG}.config": {"group_file_settings": lambda: ([], None)},
    f"{_PKG}.policy": {"Policy": type("Policy", (), {})},
    f"{_PKG}.services": {},
    f"{_PKG}.services.storage": {},
    f"{_PKG}.services.storage.group_file": {"upload_group_files": None},
    f"{_PKG}.services.storage.s3": {"upload_with_plan": None},
    f"{_PKG}.utils": {"media_rel_dir_of": lambda _p: ""},
}


@pytest.fixture(scope="module")
def sender():
    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

    target = f"{_PKG}.handlers.sender"
    saved[target] = sys.modules.get(target)
    spec = importlib.util.spec_from_file_location(target, _SENDER_PATH)
    module = importlib.util.module_from_spec(spec)
    sys.modules[target] = module
    spec.loader.exec_module(module)

    yield module

    for name, previous in saved.items():
        if previous is None:
            sys.modules.pop(name, None)
        else:
            sys.modules[name] = previous


class _Policy:
    """只带 sends_link 的策略替身（真实 Policy 需要一堆字段/归一化）。"""

    def __init__(self, sends_link: bool):
        self.sends_link = sends_link


@pytest.fixture
def uploads(monkeypatch, sender):
    """记录 upload_with_plan 的调用；返回 (调用列表, 设置返回值的钩子)。"""
    calls: list[Path] = []
    result: dict = {"pub": "https://pub.example/x.mp4", "raise": None}

    def fake_upload(file_path, *, policy=None):
        calls.append(Path(file_path))
        if result["raise"]:
            raise result["raise"]
        return "local-url", result["pub"]

    monkeypatch.setattr(sender, "upload_with_plan", fake_upload)
    return calls, result


@pytest.fixture
def media(sender):
    def build(sends_link: bool, n: int = 1, policy=None):
        files = [Path(f"D:/temp/f{i}.mp4") for i in range(n)]
        return sender.PendingMedia(
            files=files,
            policy=_Policy(sends_link) if policy is None else policy,
        )

    return build


async def test_no_s3_when_link_disabled(sender, media, uploads):
    """默认路径：本地发成功就够了，绝不碰 S3（媒体不走 S3 归档是有意的）。"""
    calls, _ = uploads
    ok, pub = await sender._local_sent(media(sends_link=False))
    assert (ok, pub) == (True, None)
    assert calls == []


async def test_s3_fetched_when_link_enabled(sender, media, uploads):
    """策略要下载链接时：补传 S3 并把 public_url 交出去（本次修的回归）。"""
    calls, _ = uploads
    ok, pub = await sender._local_sent(media(sends_link=True))
    assert (ok, pub) == (True, "https://pub.example/x.mp4")
    assert len(calls) == 1


async def test_multi_file_uploads_each(sender, media, uploads):
    calls, _ = uploads
    ok, pub = await sender._local_sent(media(sends_link=True, n=3))
    assert ok is True
    assert pub == "https://pub.example/x.mp4"
    assert len(calls) == 3


async def test_no_public_url_keeps_none(sender, media, uploads):
    """S3 有本地链接但没公网（upload_public 关）→ 不带链接，但也不算失败。"""
    calls, result = uploads
    result["pub"] = None
    ok, pub = await sender._local_sent(media(sends_link=True))
    assert (ok, pub) == (True, None)
    assert len(calls) == 1


async def test_upload_failure_never_breaks_send(sender, media, uploads):
    """补传失败只记日志：发送本身已经成功了，不能反过来判失败。"""
    calls, result = uploads
    result["raise"] = RuntimeError("S3 挂了")
    ok, pub = await sender._local_sent(media(sends_link=True))
    assert (ok, pub) == (True, None)
    assert len(calls) == 1


async def test_policy_none_is_safe(sender, media, uploads):
    """无策略（如 Web 面板路径）时不补传也不炸。"""
    calls, _ = uploads
    ok, pub = await sender._local_sent(media(sends_link=False, policy=None))
    assert (ok, pub) == (True, None)
    assert calls == []