mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-10-31 12:45:41 +08:00
81bbf9c71d
- run-command：等待进程退出而不是等 stderr 关闭（后台子进程不再导致误判超时），
stderr 写临时文件取末尾；只替换整项为 {字段} 的参数，{files} 展开为每个文件
一个参数，夹带的占位符原样保留；不把宿主自己的 AB_* 环境变量传给命令；README
说明脚本从 AB_* 环境变量读取事件数据
- strm-writer：路径规范化后为媒体根目录本身（. / ..）时跳过，不再写出 STRM 目录之外的文件
- slack-notifier / onebot-notifier：带海报发送失败时去掉图片只发文字重试一次
- wpush-notifier / onebot-notifier：非 2xx 时报告 HTTP 状态，响应不是 JSON 时不再报解析异常
- 插件通知渠道的 extra 包含已填写的已声明字段（group、icon 等），不只是 model_extra
- 新增通知渠道时切换类型会清掉上一类型的字段，避免 Bark 带上残留的 Webhook URL
79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import httpx2
|
|
import pytest
|
|
from wpush_notifier import API_URL, WPushNotifier
|
|
|
|
from ab_sdk.notify import NotificationMessage, NotifierSettings
|
|
from ab_sdk.testing import NotifierContract, create_plugin
|
|
|
|
|
|
def channel(status: int, body: dict | str, options: dict | None = None, seen=None):
|
|
def handler(request: httpx2.Request) -> httpx2.Response:
|
|
if seen is not None:
|
|
seen.append(request)
|
|
if isinstance(body, str):
|
|
return httpx2.Response(status, text=body)
|
|
return httpx2.Response(status, json=body)
|
|
|
|
plugin, _ = create_plugin(
|
|
WPushNotifier, {"apikey": "k"} if options is None else options
|
|
)
|
|
plugin.client_factory = lambda: httpx2.AsyncClient( # type: ignore[method-assign]
|
|
transport=httpx2.MockTransport(handler)
|
|
)
|
|
return plugin.channel()(NotifierSettings())
|
|
|
|
|
|
class TestChannel(NotifierContract):
|
|
def create(self):
|
|
return channel(200, {"code": 0, "message": "ok"})
|
|
|
|
def create_failing(self):
|
|
return channel(200, {"code": 1, "message": "apikey 无效"})
|
|
|
|
|
|
def test_send_posts_payload_with_topic_code():
|
|
seen: list[httpx2.Request] = []
|
|
ch = channel(
|
|
200,
|
|
{"code": 0},
|
|
{"apikey": "k", "channel": "sms", "topic_code": "g1"},
|
|
seen,
|
|
)
|
|
msg = NotificationMessage(kind="event", title="标题", body="正文")
|
|
assert asyncio.run(ch.send(msg)) is True
|
|
[request] = seen
|
|
assert str(request.url) == API_URL
|
|
assert json.loads(request.content) == {
|
|
"apikey": "k",
|
|
"title": "标题",
|
|
"content": "正文",
|
|
"channel": "sms",
|
|
"topic_code": "g1",
|
|
}
|
|
|
|
|
|
def test_test_reports_api_message_on_http_error():
|
|
ok, detail = asyncio.run(channel(500, {"message": "内部错误"}).test())
|
|
assert ok is False
|
|
assert "内部错误" in detail
|
|
|
|
|
|
def test_send_without_apikey_returns_false_without_request():
|
|
seen: list[httpx2.Request] = []
|
|
ch = channel(200, {"code": 0}, {}, seen)
|
|
assert asyncio.run(ch.test()) == (False, "未填写 apikey")
|
|
assert seen == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "body", "expected"),
|
|
[(502, "<html>Bad Gateway</html>", "HTTP 502"), (200, "<html>", "不是预期的 JSON")],
|
|
)
|
|
def test_test_non_json_response_reports_status(status, body, expected):
|
|
ok, detail = asyncio.run(channel(status, body).test())
|
|
assert ok is False
|
|
assert expected in detail
|