RAG/backend/tests/contract/test_bailian_chat.py

from __future__ import annotations

import json

import httpx
import pytest

from app.adapters.bailian.chat import BailianChatAdapter
from app.ports.model_providers import ChatMessage, ModelProviderError

BASE_URL = "https://workspace.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
API_KEY = "sk-test-chat-secret"


@pytest.mark.asyncio
async def test_chat_completion_disables_thinking_and_search() -> None:
    messages = [ChatMessage(role="user", content="Give a grounded answer")]

    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/compatible-mode/v1/chat/completions"
        assert request.headers["Authorization"] == f"Bearer {API_KEY}"
        payload = json.loads(request.content)
        assert payload["model"] == "deepseek-v4-flash"
        assert payload["messages"] == [{"role": "user", "content": "Give a grounded answer"}]
        assert payload["enable_thinking"] is False
        assert payload["enable_search"] is False
        assert payload["stream"] is False
        return httpx.Response(
            200,
            json={
                "id": "req_chat_1",
                "model": "deepseek-v4-flash",
                "choices": [
                    {
                        "message": {"role": "assistant", "content": "grounded"},
                        "finish_reason": "stop",
                    }
                ],
                "usage": {
                    "prompt_tokens": 5,
                    "completion_tokens": 1,
                    "total_tokens": 6,
                },
            },
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        adapter = BailianChatAdapter(
            api_key=API_KEY,
            base_url=BASE_URL,
            http_client=client,
        )
        result = await adapter.complete(messages, max_tokens=32)

    assert result.content == "grounded"
    assert result.finish_reason == "stop"
    assert result.usage.total_tokens == 6


@pytest.mark.asyncio
async def test_chat_stream_disables_thinking_and_search_and_parses_sse() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content)
        assert payload["enable_thinking"] is False
        assert payload["enable_search"] is False
        assert payload["stream"] is True
        assert payload["stream_options"] == {"include_usage": True}
        chunks = (
            'data: {"id":"req_stream","model":"deepseek-v4-flash",'
            '"choices":[{"delta":{"content":"矿"},"finish_reason":null}]}\n\n'
            'data: {"id":"req_stream","model":"deepseek-v4-flash",'
            '"choices":[{"delta":{"content":"体"},"finish_reason":"stop"}]}\n\n'
            'data: {"id":"req_stream","model":"deepseek-v4-flash",'
            '"choices":[],"usage":{"prompt_tokens":4,"completion_tokens":2,'
            '"total_tokens":6}}\n\n'
            "data: [DONE]\n\n"
        )
        return httpx.Response(
            200,
            headers={"content-type": "text/event-stream"},
            content=chunks.encode(),
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        adapter = BailianChatAdapter(
            api_key=API_KEY,
            base_url=BASE_URL,
            http_client=client,
        )
        events = [
            event
            async for event in adapter.stream(
                [ChatMessage(role="user", content="question")],
                max_tokens=32,
            )
        ]

    assert "".join(event.delta for event in events) == "矿体"
    assert events[1].finish_reason == "stop"
    assert events[2].usage.total_tokens == 6


@pytest.mark.asyncio
async def test_chat_error_does_not_expose_key_or_message_body() -> None:
    sensitive_text = "CONFIDENTIAL_PROMPT_BODY"

    def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(
            401,
            json={
                "error": {
                    "code": API_KEY,
                    "message": f"{sensitive_text}: {API_KEY}",
                }
            },
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        adapter = BailianChatAdapter(
            api_key=API_KEY,
            base_url=BASE_URL,
            http_client=client,
        )
        with pytest.raises(ModelProviderError) as exc_info:
            await adapter.complete(
                [ChatMessage(role="user", content=sensitive_text)],
                max_tokens=8,
            )

    rendered = f"{exc_info.value!s} {exc_info.value!r}"
    assert API_KEY not in rendered
    assert sensitive_text not in rendered


@pytest.mark.asyncio
async def test_chat_stream_honors_retry_after_before_any_token() -> None:
    calls = 0

    def handler(_: httpx.Request) -> httpx.Response:
        nonlocal calls
        calls += 1
        if calls == 1:
            return httpx.Response(
                429,
                headers={"Retry-After": "0"},
                json={"error": {"code": "rate_limited"}},
            )
        return httpx.Response(
            200,
            headers={"content-type": "text/event-stream"},
            content=(
                'data: {"model":"deepseek-v4-flash",'
                '"choices":[{"delta":{"content":"正常"},"finish_reason":"stop"}]}\n\n'
                "data: [DONE]\n\n"
            ).encode(),
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        adapter = BailianChatAdapter(
            api_key=API_KEY,
            base_url=BASE_URL,
            http_client=client,
            max_retries=1,
            retry_base_seconds=0,
        )
        events = [
            event
            async for event in adapter.stream(
                [ChatMessage(role="user", content="question")],
                max_tokens=8,
            )
        ]

    assert "".join(event.delta for event in events) == "正常"
    assert calls == 2