patbond-ai/tests/providers/volcengine/test_client.py

"""`ArkClient` 的测试——只测传输层：鉴权头、错误解包、连接生命周期。

**刻意不测任何视频业务**：本 client 不认识「视频」，测试也不该。视频相关的断言
在 `test_video.py`。
"""

from __future__ import annotations

import httpx
import pytest

from app.domain.generation import Retry, UserFacing
from app.providers.volcengine.client import ArkClient, ArkError

#: 显式的假值。防泄漏扫描会拦下 `api_key = "..."` 形态（规则正确——这正是真实
#: 密钥最常见的泄漏形态），故夹具用 ALLOW 清单认可的占位词。
FAKE_API_KEY = "dummy-api-key-for-tests"

BASE_URL = "https://ark.example.invalid/api/v3"


def client_with(handler) -> ArkClient:
    return ArkClient(
        api_key=FAKE_API_KEY,
        base_url=BASE_URL,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )


class TestConstruction:
    def test_missing_api_key_fails_at_construction_not_at_first_call(self):
        # 缺密钥是配置错误。让它表现为一串莫名的 401 会浪费排查时间。
        with pytest.raises(ValueError, match="ARK_API_KEY"):
            ArkClient(api_key="", base_url=BASE_URL)

    def test_trailing_slash_in_base_url_does_not_double_up(self):
        def handler(request: httpx.Request) -> httpx.Response:
            assert "//contents" not in str(request.url)
            return httpx.Response(200, json={})

        ArkClient(
            api_key=FAKE_API_KEY,
            base_url=BASE_URL + "/",
            client=httpx.Client(transport=httpx.MockTransport(handler)),
        ).get("/contents/x")


class TestAuthAndTransport:
    def test_bearer_token_and_content_type_are_sent(self):
        def handler(request: httpx.Request) -> httpx.Response:
            assert request.headers["Authorization"] == f"Bearer {FAKE_API_KEY}"
            assert request.headers["Content-Type"] == "application/json"
            return httpx.Response(200, json={"ok": True})

        assert client_with(handler).post("/x", {}) == {"ok": True}

    def test_get_and_post_hit_the_expected_method(self):
        seen: list[str] = []

        def handler(request: httpx.Request) -> httpx.Response:
            seen.append(request.method)
            return httpx.Response(200, json={})

        client = client_with(handler)
        client.post("/x", {})
        client.get("/x")
        assert seen == ["POST", "GET"]

    def test_network_failure_is_classified_retryable(self):
        def handler(request):
            raise httpx.ConnectError("connection refused")

        with pytest.raises(ArkError) as exc:
            client_with(handler).get("/x")
        assert exc.value.failure.retry is Retry.BACKOFF
        assert exc.value.http_status is None


class TestErrorUnwrapping:
    def test_vendor_error_body_is_classified(self):
        def handler(request):
            return httpx.Response(
                429,
                json={
                    "error": {
                        "code": "RateLimitExceeded.EndpointRPMExceeded",
                        "message": "RPM limit exceeded",
                    }
                },
            )

        with pytest.raises(ArkError) as exc:
            client_with(handler).get("/x")
        assert exc.value.failure.retry is Retry.BACKOFF
        assert exc.value.failure.user_facing is UserFacing.RATE_LIMITED
        assert (
            exc.value.failure.provider_code == "RateLimitExceeded.EndpointRPMExceeded"
        )
        assert exc.value.http_status == 429

    def test_vendor_message_is_preserved_for_troubleshooting(self):
        def handler(request):
            return httpx.Response(
                401,
                json={
                    "error": {"code": "AuthenticationError", "message": "key invalid"}
                },
            )

        with pytest.raises(ArkError) as exc:
            client_with(handler).get("/x")
        assert "key invalid" in exc.value.failure.message

    def test_non_json_error_body_is_not_swallowed(self):
        # 打到错误地址时常见：网关返回 HTML 错误页。静默会让排查绕远路。
        def handler(request):
            return httpx.Response(502, text="<html>bad gateway</html>")

        with pytest.raises(ArkError) as exc:
            client_with(handler).get("/x")
        assert "bad gateway" in exc.value.failure.message
        assert exc.value.failure.retry is Retry.BACKOFF

    def test_non_json_success_body_is_an_error_too(self):
        def handler(request):
            return httpx.Response(200, text="not json")

        with pytest.raises(ArkError, match="不是合法 JSON"):
            client_with(handler).get("/x")


class TestConnectionOwnership:
    def test_injected_client_is_not_closed_by_us(self):
        """注入进来的连接池归调用方管理。

        在这里关掉会让共享同一个 httpx.Client 的其他适配器一起失效——接了第二个
        火山接口后这会变成一个极难查的偶发失败。
        """
        injected = httpx.Client(
            transport=httpx.MockTransport(lambda r: httpx.Response(200, json={}))
        )
        ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL, client=injected).close()
        assert not injected.is_closed

    def test_own_client_is_closed(self):
        client = ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL)
        client.close()
        assert client._client.is_closed

    def test_usable_as_context_manager(self):
        with ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL) as client:
            inner = client._client
        assert inner.is_closed