meida_backend/tests/unit/test_deepseek_provider.py

"""DeepSeek provider against a mocked transport. No network access."""

from __future__ import annotations

import json
from typing import Any

import httpx
import pytest

from packages.llm.errors import (
    ProviderAuthError,
    ProviderBadRequestError,
    ProviderRateLimitedError,
    ProviderServerError,
    ProviderTimeoutError,
)
from packages.llm.providers.deepseek import DeepSeekProvider
from packages.llm.types import Message, ProviderRequest

CHAT_RESPONSE: dict[str, Any] = {
    "id": "chatcmpl-1",
    "model": "deepseek-v4-flash",
    "choices": [
        {
            "index": 0,
            "message": {"role": "assistant", "content": "白衬衫配西裤"},
            "finish_reason": "stop",
        }
    ],
    "usage": {
        "prompt_tokens": 1200,
        "completion_tokens": 300,
        "prompt_cache_hit_tokens": 1000,
        "prompt_cache_miss_tokens": 200,
    },
}


def make_provider(handler: Any) -> DeepSeekProvider:
    return DeepSeekProvider(
        api_key="test-key",
        client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
    )


def request(json_mode: bool = False) -> ProviderRequest:
    return ProviderRequest(
        model="deepseek-v4-flash",
        messages=(Message(role="system", content="前缀"), Message(role="user", content="问题")),
        temperature=0.7,
        max_tokens=800,
        timeout_seconds=10.0,
        json_mode=json_mode,
    )


async def test_successful_completion_maps_usage_including_cache_hits() -> None:
    provider = make_provider(lambda _r: httpx.Response(200, json=CHAT_RESPONSE))
    result = await provider.complete(request())

    assert result.text == "白衬衫配西裤"
    assert result.model == "deepseek-v4-flash"
    assert result.usage.tokens_in == 1200
    assert result.usage.tokens_out == 300
    assert result.usage.tokens_cached == 1000
    assert result.usage.billable_input == 200
    await provider.aclose()


async def test_openai_style_cached_tokens_are_also_understood() -> None:
    payload = {
        **CHAT_RESPONSE,
        "usage": {
            "prompt_tokens": 500,
            "completion_tokens": 10,
            "prompt_tokens_details": {"cached_tokens": 400},
        },
    }
    provider = make_provider(lambda _r: httpx.Response(200, json=payload))
    result = await provider.complete(request())
    assert result.usage.tokens_cached == 400


async def test_request_shape_is_openai_compatible() -> None:
    captured: dict[str, Any] = {}

    def handler(req: httpx.Request) -> httpx.Response:
        captured["url"] = str(req.url)
        captured["auth"] = req.headers["authorization"]
        captured["body"] = json.loads(req.content)
        return httpx.Response(200, json=CHAT_RESPONSE)

    provider = make_provider(handler)
    await provider.complete(request(json_mode=True))

    assert captured["url"] == "https://api.deepseek.com/v1/chat/completions"
    assert captured["auth"] == "Bearer test-key"
    assert captured["body"]["model"] == "deepseek-v4-flash"
    assert captured["body"]["response_format"] == {"type": "json_object"}
    assert [m["role"] for m in captured["body"]["messages"]] == ["system", "user"]


@pytest.mark.parametrize(
    ("status", "expected"),
    [
        (429, ProviderRateLimitedError),
        (401, ProviderAuthError),
        (403, ProviderAuthError),
        (500, ProviderServerError),
        (503, ProviderServerError),
        (400, ProviderBadRequestError),
        (404, ProviderBadRequestError),
    ],
)
async def test_http_errors_map_to_the_taxonomy(status: int, expected: type[Exception]) -> None:
    provider = make_provider(lambda _r: httpx.Response(status, json={"error": "boom"}))
    with pytest.raises(expected):
        await provider.complete(request())


async def test_only_4xx_bad_requests_are_non_retryable() -> None:
    provider = make_provider(lambda _r: httpx.Response(400, json={"error": "bad"}))
    with pytest.raises(ProviderBadRequestError) as excinfo:
        await provider.complete(request())
    assert excinfo.value.retryable is False


async def test_timeouts_become_provider_timeout_errors() -> None:
    def handler(req: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("too slow", request=req)

    provider = make_provider(handler)
    with pytest.raises(ProviderTimeoutError):
        await provider.complete(request())


async def test_transport_errors_are_retryable_server_errors() -> None:
    def handler(req: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("dns failure", request=req)

    provider = make_provider(handler)
    with pytest.raises(ProviderServerError) as excinfo:
        await provider.complete(request())
    assert excinfo.value.retryable is True


async def test_malformed_response_body_is_a_server_error() -> None:
    provider = make_provider(lambda _r: httpx.Response(200, json={"unexpected": True}))
    with pytest.raises(ProviderServerError):
        await provider.complete(request())


async def test_zhipu_placeholder_explains_how_to_enable_it() -> None:
    from packages.llm.providers.zhipu import ZhipuGLMProvider

    provider = ZhipuGLMProvider()
    assert provider.name == "zhipu"
    assert provider.default_model == "glm-4-flash"
    with pytest.raises(NotImplementedError, match="备案"):
        await provider.complete(request())
    await provider.aclose()