re-xxm/tests/test_httpx_transport.py

# -*- 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()