kairee
5b22ed1c5d
Add httpx and route Bilibili and Telegram requests through pooled clients with keepalive and explicit timeouts. Align Bilibili headers with data/raw-request.txt: Chrome 155 UA, sec-ch-ua, space.bilibili.com origin, folder-specific referer, and priority header. Close replaced clients when credentials change. Add transport contract tests and live read-only validation for nav, folder sync, WBI search, and Telegram getUpdates.
146 lines
4.1 KiB
Python
146 lines
4.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""httpx 传输层契约测试（不联网）: uv run python tests/test_httpx_transport.py"""
|
|
|
|
import os
|
|
import sys
|
|
|
|
import httpx
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
from reaction_board.bilibili.client import ( # noqa: E402
|
|
HttpxTransport,
|
|
TransportError,
|
|
)
|
|
from reaction_board.notify.telegram import ( # noqa: E402
|
|
HttpxTelegramTransport,
|
|
TelegramError,
|
|
)
|
|
|
|
|
|
def test_bilibili_httpx_transport():
|
|
captured = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured["method"] = request.method
|
|
captured["url"] = str(request.url)
|
|
captured["ua"] = request.headers["user-agent"]
|
|
return httpx.Response(
|
|
200,
|
|
json={"code": 0, "message": "OK", "data": []},
|
|
headers={"content-type": "application/json"},
|
|
)
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(handler))
|
|
transport = HttpxTransport(client=client)
|
|
try:
|
|
response = transport.request(
|
|
"GET",
|
|
"https://api.bilibili.com/x/test",
|
|
{"user-agent": "UA-TEST"},
|
|
data=None,
|
|
timeout=30,
|
|
)
|
|
finally:
|
|
transport.close()
|
|
assert response.status == 200
|
|
assert response.json()["code"] == 0
|
|
assert captured == {
|
|
"method": "GET",
|
|
"url": "https://api.bilibili.com/x/test",
|
|
"ua": "UA-TEST",
|
|
}
|
|
|
|
|
|
def test_bilibili_httpx_transport_maps_errors():
|
|
def handler(request):
|
|
raise httpx.ConnectError("boom")
|
|
|
|
transport = HttpxTransport(
|
|
client=httpx.Client(transport=httpx.MockTransport(handler))
|
|
)
|
|
try:
|
|
transport.request("GET", "https://api.bilibili.com/x/test", {})
|
|
except TransportError as exc:
|
|
assert "boom" in str(exc)
|
|
else:
|
|
raise AssertionError("httpx 连接错误应映射为 TransportError")
|
|
finally:
|
|
transport.close()
|
|
|
|
|
|
def test_telegram_httpx_transport():
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json={"ok": True, "result": [{"update_id": 1}]})
|
|
|
|
transport = HttpxTelegramTransport(
|
|
client=httpx.Client(transport=httpx.MockTransport(handler))
|
|
)
|
|
try:
|
|
payload = transport.post(
|
|
"https://api.telegram.org/botx/getUpdates", {"timeout": 0}, 30
|
|
)
|
|
finally:
|
|
transport.close()
|
|
assert payload["ok"] is True
|
|
assert payload["result"][0]["update_id"] == 1
|
|
|
|
|
|
def test_telegram_httpx_transport_error_shapes():
|
|
def bad_json(request):
|
|
return httpx.Response(
|
|
400, json={"ok": False, "description": "query is too old"}
|
|
)
|
|
|
|
transport = HttpxTelegramTransport(
|
|
client=httpx.Client(transport=httpx.MockTransport(bad_json))
|
|
)
|
|
try:
|
|
payload = transport.post("https://api.telegram.org/botx/x", {}, 30)
|
|
finally:
|
|
transport.close()
|
|
assert payload["description"] == "query is too old"
|
|
|
|
def html_error(request):
|
|
return httpx.Response(500, text="bad gateway")
|
|
|
|
transport = HttpxTelegramTransport(
|
|
client=httpx.Client(transport=httpx.MockTransport(html_error))
|
|
)
|
|
try:
|
|
transport.post("https://api.telegram.org/botx/x", {}, 30)
|
|
except TelegramError as exc:
|
|
assert "500" in str(exc)
|
|
else:
|
|
raise AssertionError("非 JSON 错误应抛出 TelegramError")
|
|
finally:
|
|
transport.close()
|
|
|
|
def network_error(request):
|
|
raise httpx.ConnectError("offline")
|
|
|
|
transport = HttpxTelegramTransport(
|
|
client=httpx.Client(transport=httpx.MockTransport(network_error))
|
|
)
|
|
try:
|
|
transport.post("https://api.telegram.org/botx/x", {}, 30)
|
|
except TelegramError as exc:
|
|
assert "offline" in str(exc)
|
|
else:
|
|
raise AssertionError("网络错误应抛出 TelegramError")
|
|
finally:
|
|
transport.close()
|
|
|
|
|
|
def main():
|
|
test_bilibili_httpx_transport()
|
|
test_bilibili_httpx_transport_maps_errors()
|
|
test_telegram_httpx_transport()
|
|
test_telegram_httpx_transport_error_shapes()
|
|
print("OK: httpx B 站与 Telegram 传输层测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|