mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-11-02 13:45:42 +08:00
667f769a2d
- apprise：urls 含令牌，改为 secret_field；secret_field 支持 default_factory，可声明列表 / 字典秘密字段
- run-command：stderr 改为自建管道边读边丢，只保留最后 2000 字节；完成仍只看 proc.wait()，结束后给读取 0.5 秒再关闭管道
- run-command：{files} 按原始路径列表展开，含换行的路径不再拆成两个参数
- 通知：插件渠道 extra 里的已声明字段与内置渠道一样展开 $VAR
- slack：只在 HTTP 400 invalid_blocks 时退回纯文字重试，其它失败不重发
- onebot：先发文字，再单独发海报；结果只看文字，去掉重试
- strm-writer：media_root 为 "/" 时正确映射完整路径
83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
import asyncio
|
|
import logging
|
|
|
|
import pytest
|
|
from strm_writer import StrmWriter
|
|
|
|
from ab_sdk.events import OrganizedFile, TorrentOrganized
|
|
from ab_sdk.testing import create_plugin
|
|
|
|
MEDIA = "/downloads/Bangumi/A/Season 1/A S01E01.mkv"
|
|
|
|
|
|
def organized(*files: OrganizedFile) -> TorrentOrganized:
|
|
return TorrentOrganized(torrent_hash="h", bangumi_id=1, files=files)
|
|
|
|
|
|
def run(
|
|
tmp_path,
|
|
url_prefix: str,
|
|
*files: OrganizedFile,
|
|
media_root: str = "/downloads/Bangumi/",
|
|
) -> None:
|
|
options = {
|
|
"strm_root": str(tmp_path / "STRM"),
|
|
"url_prefix": url_prefix,
|
|
"media_root": media_root,
|
|
}
|
|
plugin, _ = create_plugin(StrmWriter, options)
|
|
asyncio.run(plugin.on_organized(organized(*files)))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("url_prefix", "content"),
|
|
[
|
|
(
|
|
"http://nas:8096/media/",
|
|
"http://nas:8096/media/A/Season%201/A%20S01E01.mkv",
|
|
),
|
|
("/mnt/cloud/", "/mnt/cloud/A/Season 1/A S01E01.mkv"),
|
|
],
|
|
)
|
|
def test_on_organized_mirrors_media_path_into_strm(tmp_path, url_prefix, content):
|
|
run(
|
|
tmp_path,
|
|
url_prefix,
|
|
OrganizedFile(MEDIA, "media"),
|
|
OrganizedFile(MEDIA.replace(".mkv", ".ass"), "subtitle"),
|
|
)
|
|
season = tmp_path / "STRM" / "A" / "Season 1"
|
|
assert [p.name for p in season.iterdir()] == ["A S01E01.strm"]
|
|
assert (season / "A S01E01.strm").read_text(encoding="utf-8") == content
|
|
|
|
|
|
def test_on_organized_media_root_slash_maps_whole_path(tmp_path):
|
|
# 根目录 "/" 去掉尾部斜杠后是空串，曾把所有绝对路径都判为不在根目录下
|
|
run(tmp_path, "/mnt/", OrganizedFile("/dl/A/B.mkv", "media"), media_root="/")
|
|
strm = tmp_path / "STRM" / "dl" / "A" / "B.strm"
|
|
assert strm.read_text(encoding="utf-8") == "/mnt/dl/A/B.mkv"
|
|
|
|
|
|
def test_on_organized_prefix_change_rewrites_strm(tmp_path):
|
|
run(tmp_path, "http://old/", OrganizedFile(MEDIA, "media"))
|
|
run(tmp_path, "http://new/", OrganizedFile(MEDIA, "media"))
|
|
strm = tmp_path / "STRM" / "A" / "Season 1" / "A S01E01.strm"
|
|
assert strm.read_text(encoding="utf-8").startswith("http://new/")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/other/A S01E01.mkv",
|
|
"/downloads/Bangumi/../../etc/x.mkv",
|
|
# 规范化后是媒体根目录本身，曾写出 STRM 目录之外的 STRM.strm
|
|
"/downloads/Bangumi/x/..",
|
|
"/downloads/Bangumi/",
|
|
],
|
|
)
|
|
def test_on_organized_path_outside_media_root_is_skipped(tmp_path, caplog, path):
|
|
with caplog.at_level(logging.WARNING):
|
|
run(tmp_path, "http://nas/", OrganizedFile(path, "media"))
|
|
assert list(tmp_path.iterdir()) == []
|
|
assert "不在媒体根目录" in caplog.text
|