mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-10-31 12:45:41 +08:00
124b156b7a
- ab-plugin new 与示例插件的 requires-python 改为 >=3.14，与 SDK 0.6.0 一致 - 示例插件 ntfy-notifier / webhook-on-event 改用 httpx2 - openai / anthropic / google-genai 下限改为实际需要的大版本 - 插件 MCP 工具的 input_schema 无效时计入熔断 - 复制 RSS 链接在 HTTP 部署下回退到 execCommand（useClipboard legacy） - CI 步骤名、PLAYWRIGHT_VERSION、README 徽章与 e2e 指南改为 Python 3.14 / Playwright 1.64.1
69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
import asyncio
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
|
|
import httpx2
|
|
import pytest
|
|
from webhook_on_event import WebhookOnEvent
|
|
|
|
from ab_sdk.events import OrganizedFile, TorrentOrganized
|
|
from ab_sdk.testing import create_plugin
|
|
|
|
ORGANIZED = TorrentOrganized(
|
|
torrent_hash="abc",
|
|
bangumi_id=1,
|
|
files=(OrganizedFile(path="/a/b.mkv", kind="media"),),
|
|
)
|
|
|
|
|
|
def deliver(options: dict, status: int = 200):
|
|
"""返回 (收到的请求, 发送函数)。"""
|
|
plugin, _ = create_plugin(WebhookOnEvent, options)
|
|
seen: list[httpx2.Request] = []
|
|
|
|
def handler(request: httpx2.Request) -> httpx2.Response:
|
|
seen.append(request)
|
|
return httpx2.Response(status)
|
|
|
|
plugin.client_factory = lambda: httpx2.AsyncClient( # type: ignore[method-assign]
|
|
transport=httpx2.MockTransport(handler)
|
|
)
|
|
return seen, lambda: asyncio.run(plugin.on_event(ORGANIZED))
|
|
|
|
|
|
def test_on_event_selected_kind_posts_signed_json():
|
|
seen, send = deliver({"url": "https://hook.test/x", "secret": "s3cret"})
|
|
send()
|
|
[request] = seen
|
|
body = json.loads(request.content)
|
|
assert body["kind"] == "torrent.organized"
|
|
assert body["data"]["files"][0]["path"] == "/a/b.mkv"
|
|
expected = hmac.new(b"s3cret", request.content, hashlib.sha256).hexdigest()
|
|
assert request.headers["X-AB-Signature"] == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"options",
|
|
[
|
|
{"url": ""},
|
|
{"url": "https://hook.test/x", "events": ["rss_failure"]},
|
|
],
|
|
)
|
|
def test_on_event_unwanted_event_sends_nothing(options):
|
|
seen, send = deliver(options)
|
|
send()
|
|
assert seen == []
|
|
|
|
|
|
def test_on_event_without_secret_omits_signature():
|
|
seen, send = deliver({"url": "https://hook.test/x", "events": ["*"]})
|
|
send()
|
|
assert "X-AB-Signature" not in seen[0].headers
|
|
|
|
|
|
def test_on_event_server_error_is_logged_not_raised():
|
|
seen, send = deliver({"url": "https://hook.test/x"}, status=500)
|
|
send()
|
|
assert len(seen) == 1
|