group_fqcd_jr/tests/unit/service/test_model_embedding.py

"""模型向量化（embedding）能力的契约测试。

与文本生成同族：OpenAI-compatible `/embeddings`、密钥只从 `secret_ref` 解析、
端点按声明顺序尝试、降级时显式标记，任何结构违约都失败关闭。
"""

import json
from typing import Any

import httpx
import pytest

from app.core.errors import RecoverableAgentError
from app.service.model_gateway import (
    ModelDispatchService,
    ModelEmbeddingService,
    OpenAICompatibleGateway,
)


class Endpoint:
    def __init__(self, code: str, *, timeout_ms: int = 1000) -> None:
        self.endpoint_code = code
        self.base_url = "https://model.example/v1"
        self.model_name = "embed-small"
        self.secret_ref = "env:MODEL_TEST_KEY"
        self.timeout_ms = timeout_ms


class SeedResolver:
    """替身密钥解析器：断言只接受 env: 引用，返回值固定。"""

    def resolve(self, secret_ref: str) -> str:
        assert secret_ref.startswith("env:")
        return "test-key"


class RecordingGateway:
    """替身网关：按端点记录调用并返回预设结果或抛错。"""

    def __init__(self, behaviour: dict[str, Any]) -> None:
        self.behaviour = behaviour
        self.calls: list[str] = []

    async def generate(self, *, endpoint_code: str, prompt: str, timeout_ms: int) -> str:
        raise AssertionError("embedding 测试不应调用文本生成")

    async def embed(self, *, endpoint_code: str, text: str, timeout_ms: int) -> list[float]:
        self.calls.append(endpoint_code)
        result = self.behaviour[endpoint_code]
        if isinstance(result, Exception):
            raise result
        return list(result)


async def test_embed_parses_openai_compatible_response() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path.endswith("/embeddings")
        assert request.headers["authorization"] == "Bearer test-key"
        body = json.loads(request.content)
        assert body["model"] == "embed-small"
        assert body["input"] == "稳健型"
        return httpx.Response(200, json={"data": [{"embedding": [0.1, 0.2, 0.3]}]})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        gateway = OpenAICompatibleGateway(
            {"ep-1": Endpoint("ep-1")}, secret_resolver=SeedResolver(), client=client
        )
        vector = await gateway.embed(endpoint_code="ep-1", text="稳健型", timeout_ms=1000)

    assert vector == [0.1, 0.2, 0.3]


@pytest.mark.parametrize(
    "payload",
    [
        {"data": []},
        {"data": [{}]},
        {"data": [{"embedding": []}]},
        {"data": [{"embedding": ["not-a-number"]}]},
        {"unexpected": True},
    ],
)
async def test_malformed_embedding_response_fails_closed(payload: dict[str, Any]) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=payload)

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        gateway = OpenAICompatibleGateway(
            {"ep-1": Endpoint("ep-1")}, secret_resolver=SeedResolver(), client=client
        )
        with pytest.raises(RecoverableAgentError):
            await gateway.embed(endpoint_code="ep-1", text="稳健型", timeout_ms=1000)


async def test_http_failure_is_recoverable() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(503, json={"error": "unavailable"})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        gateway = OpenAICompatibleGateway(
            {"ep-1": Endpoint("ep-1")}, secret_resolver=SeedResolver(), client=client
        )
        with pytest.raises(RecoverableAgentError):
            await gateway.embed(endpoint_code="ep-1", text="稳健型", timeout_ms=1000)


async def test_dispatch_falls_back_in_declared_order() -> None:
    gateway = RecordingGateway({
        "ep-1": RecoverableAgentError("端点不可用"),
        "ep-2": [0.5, 0.6],
    })
    dispatch = ModelDispatchService(gateway)  # type: ignore[arg-type]

    execution = await dispatch.embed([Endpoint("ep-1"), Endpoint("ep-2")], "稳健型")

    assert execution.endpoint_code == "ep-2"
    assert execution.vector == [0.5, 0.6]
    assert execution.degraded is True
    assert execution.attempts == 2
    assert gateway.calls == ["ep-1", "ep-2"]


async def test_dispatch_raises_when_all_endpoints_fail() -> None:
    gateway = RecordingGateway({
        "ep-1": RecoverableAgentError("端点不可用"),
        "ep-2": RecoverableAgentError("端点不可用"),
    })
    dispatch = ModelDispatchService(gateway)  # type: ignore[arg-type]

    with pytest.raises(RecoverableAgentError):
        await dispatch.embed([Endpoint("ep-1"), Endpoint("ep-2")], "稳健型")


async def test_embedding_service_rejects_empty_endpoint_list() -> None:
    dispatch = ModelDispatchService(RecordingGateway({}))  # type: ignore[arg-type]
    service = ModelEmbeddingService(dispatch)

    with pytest.raises(RecoverableAgentError):
        await service.embed([], "稳健型")