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

"""OpenAI HTTP 传输层的测试：鉴权、两种载荷形式、错误解包。

全部离线（`httpx.MockTransport`），不触网。
"""

from __future__ import annotations

import httpx
import pytest

from app.domain.generation import Retry, UserFacing
from app.providers.openai.client import (
    DEFAULT_BASE_URL,
    OpenAiClient,
    OpenAiError,
)

FAKE_API_KEY = "dummy-api-key-for-tests"


def client_with(handler, **kwargs) -> OpenAiClient:
    return OpenAiClient(
        api_key=FAKE_API_KEY,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
        **kwargs,
    )


class TestCredentialDiscipline:
    def test_an_empty_key_fails_loudly_at_construction(self):
        """缺密钥是配置问题，应当在装配时就说清楚，而不是等第一次调用。"""
        with pytest.raises(ValueError, match="OPENAI_API_KEY 未配置"):
            OpenAiClient(api_key="")

    def test_the_key_is_sent_as_a_bearer_token(self):
        seen: dict = {}

        def handler(request: httpx.Request) -> httpx.Response:
            seen["auth"] = request.headers.get("Authorization")
            return httpx.Response(200, json={"ok": True})

        client_with(handler).post_json("/images/generations", {})
        assert seen["auth"] == f"Bearer {FAKE_API_KEY}"

    def test_organization_header_is_sent_only_when_configured(self):
        seen: list = []

        def handler(request: httpx.Request) -> httpx.Response:
            seen.append(request.headers.get("OpenAI-Organization"))
            return httpx.Response(200, json={})

        client_with(handler).post_json("/x", {})
        client_with(handler, organization="org-abc").post_json("/x", {})
        assert seen == [None, "org-abc"]


class TestUrlComposition:
    def test_default_base_url_is_the_official_one(self):
        assert DEFAULT_BASE_URL == "https://api.openai.com/v1"

    def test_a_trailing_slash_on_the_base_url_does_not_double_up(self):
        seen: dict = {}

        def handler(request: httpx.Request) -> httpx.Response:
            seen["url"] = str(request.url)
            return httpx.Response(200, json={})

        client_with(handler, base_url="https://proxy.invalid/v1/").post_json("/img", {})
        assert seen["url"] == "https://proxy.invalid/v1/img"


class TestPayloadForms:
    def test_post_json_sends_json(self):
        seen: dict = {}

        def handler(request: httpx.Request) -> httpx.Response:
            seen["content_type"] = request.headers.get("Content-Type")
            seen["body"] = request.content
            return httpx.Response(200, json={"echo": 1})

        result = client_with(handler).post_json("/x", {"prompt": "hi"})
        assert seen["content_type"] == "application/json"
        assert b'"prompt"' in seen["body"]
        assert result == {"echo": 1}

    def test_post_multipart_repeats_a_field_name(self):
        """用 list 而非 dict 传 files 的全部理由：同名字段可以重复。

        编辑端点支持多图，它们共用字段名 `image[]`——dict 表达不了。
        """
        seen: dict = {}

        def handler(request: httpx.Request) -> httpx.Response:
            seen["content_type"] = request.headers.get("Content-Type", "")
            seen["raw"] = request.content
            return httpx.Response(200, json={})

        client_with(handler).post_multipart(
            "/images/edits",
            data={"prompt": "merge"},
            files=[
                ("image[]", ("a.png", b"first", "image/png")),
                ("image[]", ("b.png", b"second", "image/png")),
            ],
        )

        assert "multipart/form-data" in seen["content_type"]
        assert seen["raw"].count(b'name="image[]"') == 2
        assert b"first" in seen["raw"]
        assert b"second" in seen["raw"]

    def test_content_type_is_not_hand_written(self):
        """multipart 还需要 boundary，手写必错——交给 httpx 按载荷推断。"""
        seen: dict = {}

        def handler(request: httpx.Request) -> httpx.Response:
            seen["ct"] = request.headers.get("Content-Type", "")
            return httpx.Response(200, json={})

        client_with(handler).post_multipart(
            "/x", data={}, files=[("image[]", ("a", b"x", "image/png"))]
        )
        assert "boundary=" in seen["ct"]


class TestErrorUnwrapping:
    def test_the_vendor_error_envelope_is_parsed(self):
        def handler(request):
            return httpx.Response(
                429,
                json={
                    "error": {
                        "message": "Your request rate increased too quickly",
                        "type": "rate_limit_error",
                        "code": "slow_down",
                        "param": None,
                    }
                },
            )

        with pytest.raises(OpenAiError) as caught:
            client_with(handler).post_json("/x", {})

        error = caught.value
        assert error.http_status == 429
        assert error.failure.retry is Retry.BACKOFF
        assert error.failure.provider_code == "slow_down"
        assert "increased too quickly" in error.failure.message

    def test_the_param_field_is_extracted_because_classification_needs_it(self):
        """`param` 是区分「参数错」与「内容被拒」的唯一结构化线索。

        漏了它，所有 400 都会被当成内容被拒——包括真正的参数错。
        """

        def handler(request):
            return httpx.Response(
                400,
                json={
                    "error": {
                        "message": "Invalid service_tier argument",
                        "type": "invalid_request_error",
                        "param": "service_tier",
                    }
                },
            )

        with pytest.raises(OpenAiError) as caught:
            client_with(handler).post_json("/x", {})

        # 带 param ⇒ 参数错 ⇒ 服务端错误，而不是让用户去改提示词。
        assert caught.value.failure.user_facing is UserFacing.SERVER_ERROR

    def test_a_non_json_error_body_still_classifies_by_status(self):
        """错误体不一定是文档描述的那个形状（网关、代理会插自己的页面）。

        解析不出时不能静默丢掉线索。
        """

        def handler(request):
            return httpx.Response(502, text="<html>bad gateway</html>")

        with pytest.raises(OpenAiError) as caught:
            client_with(handler).post_json("/x", {})

        assert caught.value.failure.retry is Retry.BACKOFF
        assert caught.value.failure.provider_code == "HTTP_502"
        # 原始文本被带上，否则排查时只知道「502」。
        assert "bad gateway" in caught.value.failure.message

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

        with pytest.raises(OpenAiError, match="不是合法 JSON"):
            client_with(handler).post_json("/x", {})

    def test_a_network_failure_is_classified_as_retryable(self):
        """⚠️ 同步接口的超时最危险：那次调用可能已经完成并计费。

        图片接口没有任务 ID 可供事后查询，故这个窗口只能由 max_attempts 兜住。
        """

        def handler(request):
            raise httpx.ConnectTimeout("timed out")

        with pytest.raises(OpenAiError) as caught:
            client_with(handler).post_json("/x", {})

        assert caught.value.http_status is None
        assert caught.value.failure.retry is Retry.BACKOFF
        assert caught.value.failure.provider_code == "NETWORK_ERROR"


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

        关掉它会让共享同一个 client 的其他适配器一起失效。
        """
        injected = httpx.Client(
            transport=httpx.MockTransport(lambda r: httpx.Response(200, json={}))
        )
        OpenAiClient(api_key=FAKE_API_KEY, client=injected).close()
        assert not injected.is_closed

    def test_our_own_client_is_closed_on_exit(self):
        with OpenAiClient(api_key=FAKE_API_KEY) as client:
            inner = client._client
        assert inner.is_closed