"""
验证异步 HTTP 请求在 HTTP/2 故障时的降级、熔断和恢复。
"""

import asyncio
import ssl
import time

import httpx2
import pytest

from app.adapters.network import http as http_module
from app.adapters.network.http import AsyncRequestUtils
from app.sdk.network import AsyncRequestUtils as SdkAsyncRequestUtils

PROXY = "http://proxy.example:7890"
URL = "https://raw.githubusercontent.com/demo/repo/main/package.json"


@pytest.fixture(autouse=True)
def _reset_h2_proxy_breaker():
    """
    每个用例前后清空熔断状态，避免跨用例污染进程级熔断字典。
    """
    http_module._h2_proxy_retry_at.clear()
    yield
    http_module._h2_proxy_retry_at.clear()


def _fake_dispatch(calls, fail_when):
    async def fake(_self, http2, _cookies_dict, _method, _url, raise_exception, **_kwargs):
        calls.append(http2)
        if fail_when(http2):
            if raise_exception:
                raise httpx2.RemoteProtocolError("tunnel closed")
            return None
        return "ok"

    return fake


def test_get_downgrades_to_h1_and_trips_breaker_on_h2_failure(monkeypatch):
    """
    走代理的幂等请求 h2 遇到连接层失败时，就地降级 h1 重试一次并触发该 (proxy, host) 的熔断。
    """
    calls = []
    monkeypatch.setattr(
        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: http2)
    )

    utils = AsyncRequestUtils(proxies={"https": PROXY})
    result = asyncio.run(utils.request("get", URL))

    assert result == "ok"
    assert calls == [True, False]
    assert http_module._h2_proxy_allowed(PROXY, URL) is False


def test_get_skips_h2_while_breaker_is_tripped(monkeypatch):
    """
    熔断冷却期内，走代理的请求直接使用 h1，不再尝试 h2。
    """
    http_module._trip_h2_proxy_breaker(PROXY, URL)
    calls = []
    monkeypatch.setattr(
        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: False)
    )

    utils = AsyncRequestUtils(proxies={"https": PROXY})
    result = asyncio.run(utils.request("get", URL))

    assert result == "ok"
    assert calls == [False]


def test_get_retries_h2_after_cooldown_expires(monkeypatch):
    """
    熔断冷却期结束后，代理请求恢复尝试 h2。
    """
    http_module._h2_proxy_retry_at[
        http_module._h2_proxy_breaker_key(PROXY, URL)
    ] = time.monotonic() - 1
    calls = []
    monkeypatch.setattr(
        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: False)
    )

    utils = AsyncRequestUtils(proxies={"https": PROXY})
    result = asyncio.run(utils.request("get", URL))

    assert result == "ok"
    assert calls == [True]


def test_timeout_does_not_trip_breaker_or_retry(monkeypatch):
    """
    超时等与 h2 隧道无关的错误不触发熔断、不做 h1 重试，且保持 raise_exception=False 返回 None 的语义。
    """
    calls = []

    async def fake(_self, http2, _cookies_dict, _method, _url, _raise_exception, **_kwargs):
        calls.append(http2)
        raise httpx2.ConnectTimeout("proxy slow")

    monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake)

    utils = AsyncRequestUtils(proxies={"https": PROXY})
    result = asyncio.run(utils.request("get", URL))

    assert result is None
    assert calls == [True]
    assert http_module._h2_proxy_allowed(PROXY, URL) is True


def test_timeout_still_raises_when_raise_exception_enabled(monkeypatch):
    """
    与 h2 隧道无关的错误在 raise_exception=True 时按原语义抛出。
    """
    calls = []

    async def fake(_self, http2, _cookies_dict, _method, _url, _raise_exception, **_kwargs):
        calls.append(http2)
        raise httpx2.ConnectTimeout("proxy slow")

    monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake)

    utils = AsyncRequestUtils(proxies={"https": PROXY})
    with pytest.raises(httpx2.ConnectTimeout):
        asyncio.run(utils.request("get", URL, raise_exception=True))

    assert calls == [True]
    assert http_module._h2_proxy_allowed(PROXY, URL) is True


def test_proxy_connect_error_retries_with_tls12(monkeypatch):
    """
    代理 TLS 握手连接错误时，幂等请求应只回退一次 TLS 1.2。
    """
    calls = []

    async def fake(
        _self,
        http2,
        _cookies_dict,
        _method,
        _url,
        _raise_exception,
        verify=None,
        **_kwargs,
    ):
        calls.append((http2, verify))
        if verify is None:
            raise httpx2.ConnectError("TLS handshake failed")
        return "ok"

    monkeypatch.setattr(AsyncRequestUtils, "_dispatch_request", fake)

    result = asyncio.run(
        AsyncRequestUtils(proxies={"https": PROXY}).request("get", URL)
    )

    assert result == "ok"
    assert calls[0] == (True, None)
    assert calls[1][0] is True
    assert isinstance(calls[1][1], ssl.SSLContext)
    assert calls[1][1].minimum_version is ssl.TLSVersion.TLSv1_2
    assert calls[1][1].maximum_version is ssl.TLSVersion.TLSv1_2


def test_stream_proxy_connect_error_retries_with_tls12(monkeypatch):
    """
    流式代理 HTTPS 请求也应在握手连接错误后回退 TLS 1.2。
    """
    seen_verifies = []

    def fake_transport(**kwargs):
        verify = kwargs["verify"]
        seen_verifies.append(verify)

        async def respond(request):
            if isinstance(verify, ssl.SSLContext):
                return httpx2.Response(200, text="ok", request=request)
            raise httpx2.ConnectError("TLS handshake failed")

        return httpx2.MockTransport(respond)

    monkeypatch.setattr(http_module, "_get_shared_async_transport", fake_transport)

    async def run_stream():
        async with AsyncRequestUtils(proxies={"https": PROXY}).get_stream(
            URL,
            raise_exception=True,
        ) as response:
            assert response is not None
            assert await response.aread() == b"ok"

    asyncio.run(run_stream())

    assert len(seen_verifies) == 2
    assert isinstance(seen_verifies[1], ssl.SSLContext)
    assert seen_verifies[1].maximum_version is ssl.TLSVersion.TLSv1_2


def test_post_does_not_downgrade_on_h2_failure(monkeypatch):
    """
    非幂等方法 h2 失败时不做 h1 降级重试，避免服务端可能已收到数据时重复产生副作用。
    """
    calls = []
    monkeypatch.setattr(
        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: True)
    )

    utils = AsyncRequestUtils(proxies={"https": PROXY})
    result = asyncio.run(utils.request("post", URL))

    assert result is None
    assert calls == [True]


def test_no_proxy_configured_skips_breaker_logic(monkeypatch):
    """
    未配置代理时不触发熔断判断，行为与原实现一致（仅走一次 h2）。
    """
    calls = []
    monkeypatch.setattr(
        AsyncRequestUtils, "_dispatch_request", _fake_dispatch(calls, fail_when=lambda http2: False)
    )

    utils = AsyncRequestUtils()
    result = asyncio.run(utils.request("get", URL))

    assert result == "ok"
    assert calls == [True]


def test_internal_client_uses_httpx2_transport(monkeypatch):
    """
    宿主自建客户端使用 HTTPX2，并返回对应响应对象。
    """

    async def respond(request: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(200, json={"client": "httpx2"}, request=request)

    monkeypatch.setattr(
        http_module,
        "_get_shared_async_transport",
        lambda **_kwargs: httpx2.MockTransport(respond),
    )

    response = asyncio.run(
        AsyncRequestUtils().get_res("https://example.com/data", raise_exception=True)
    )

    assert isinstance(response, httpx2.Response)
    assert response.json() == {"client": "httpx2"}


def test_plugin_sdk_uses_httpx2_by_default(monkeypatch):
    """
    插件 SDK 默认复用宿主 HTTPX2 客户端合同。
    """

    async def respond(request: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(200, request=request)

    monkeypatch.setattr(
        http_module,
        "_get_shared_async_transport",
        lambda **_kwargs: httpx2.MockTransport(respond),
    )

    response = asyncio.run(
        SdkAsyncRequestUtils().get_res(
            "https://example.com/sdk", raise_exception=True
        )
    )

    assert SdkAsyncRequestUtils is AsyncRequestUtils
    assert isinstance(response, httpx2.Response)


def test_legacy_http_module_uses_httpx2_by_default(monkeypatch):
    """
    旧插件 HTTP 导入入口随宿主默认客户端迁移到 HTTPX2。
    """
    import importlib

    legacy_http = importlib.import_module("app.utils.http")

    async def respond(request: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(200, request=request)

    monkeypatch.setattr(
        http_module,
        "_get_shared_async_transport",
        lambda **_kwargs: httpx2.MockTransport(respond),
    )

    response = asyncio.run(
        legacy_http.AsyncRequestUtils().get_res(
            "https://example.com/legacy", raise_exception=True
        )
    )

    assert legacy_http.AsyncRequestUtils is AsyncRequestUtils
    assert isinstance(response, httpx2.Response)