147 lines
5.2 KiB
Python
147 lines
5.2 KiB
Python
"""Request and response contract tests for supported model protocols."""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
|
|
import httpx
|
|
import pytest
|
|
from agenteval.model_gateway import ModelGateway, ModelGatewayError
|
|
from agenteval.models import ModelCapability
|
|
from agenteval.services.model_configs import ModelRuntimeConfig
|
|
|
|
|
|
def _config(provider: str, capability: ModelCapability = ModelCapability.CHAT) -> ModelRuntimeConfig:
|
|
return ModelRuntimeConfig(
|
|
id=f"{provider}-config",
|
|
name=provider,
|
|
provider=provider,
|
|
capability=capability,
|
|
endpoint_url=f"https://models.example.com/{provider}",
|
|
model_name="test-model",
|
|
api_key="test-key",
|
|
updated_at=datetime(2026, 7, 17),
|
|
)
|
|
|
|
|
|
def _gateway(handler) -> ModelGateway:
|
|
return ModelGateway(transport=httpx.MockTransport(handler))
|
|
|
|
|
|
async def test_openai_compatible_chat_embedding_and_moderation_contracts():
|
|
requests: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(request)
|
|
payload = json.loads(request.content)
|
|
if "messages" in payload:
|
|
return httpx.Response(200, json={"choices": [{"message": {"content": "chat reply"}}]})
|
|
if payload.get("input") == ["first", "second"]:
|
|
return httpx.Response(
|
|
200,
|
|
json={"data": [{"index": 1, "embedding": [2.0]}, {"index": 0, "embedding": [1.0]}]},
|
|
)
|
|
return httpx.Response(200, json={"results": [{"flagged": False}]})
|
|
|
|
gateway = _gateway(handler)
|
|
config = _config("openai_compatible")
|
|
|
|
assert await gateway.chat(config, [{"role": "user", "content": "hello"}], temperature=0.3) == "chat reply"
|
|
assert await gateway.embed(config, ["first", "second"]) == [[1.0], [2.0]]
|
|
assert await gateway.moderate(config, "safe text") == {"flagged": False}
|
|
|
|
assert all(request.headers["authorization"] == "Bearer test-key" for request in requests)
|
|
assert json.loads(requests[0].content) == {
|
|
"model": "test-model",
|
|
"messages": [{"role": "user", "content": "hello"}],
|
|
"temperature": 0.3,
|
|
}
|
|
|
|
|
|
async def test_anthropic_messages_contract():
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.headers["x-api-key"] == "test-key"
|
|
assert request.headers["anthropic-version"] == "2023-06-01"
|
|
assert "authorization" not in request.headers
|
|
assert json.loads(request.content) == {
|
|
"model": "test-model",
|
|
"max_tokens": 1024,
|
|
"messages": [
|
|
{"role": "user", "content": "hello"},
|
|
{"role": "assistant", "content": "hi"},
|
|
],
|
|
"temperature": 0.1,
|
|
"system": "be concise",
|
|
}
|
|
return httpx.Response(
|
|
200,
|
|
json={"content": [{"type": "text", "text": "first"}, {"type": "text", "text": "second"}]},
|
|
)
|
|
|
|
result = await _gateway(handler).chat(
|
|
_config("anthropic"),
|
|
[
|
|
{"role": "system", "content": "be concise"},
|
|
{"role": "user", "content": "hello"},
|
|
{"role": "assistant", "content": "hi"},
|
|
],
|
|
temperature=0.1,
|
|
)
|
|
|
|
assert result == "first\nsecond"
|
|
|
|
|
|
async def test_google_gemini_generate_content_contract():
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.headers["x-goog-api-key"] == "test-key"
|
|
assert json.loads(request.content) == {
|
|
"contents": [
|
|
{"role": "user", "parts": [{"text": "hello"}]},
|
|
{"role": "model", "parts": [{"text": "hi"}]},
|
|
],
|
|
"generationConfig": {"temperature": 0.2},
|
|
"systemInstruction": {"parts": [{"text": "be concise"}]},
|
|
}
|
|
return httpx.Response(
|
|
200,
|
|
json={"candidates": [{"content": {"parts": [{"text": "Gemini reply"}]}}]},
|
|
)
|
|
|
|
result = await _gateway(handler).chat(
|
|
_config("google_gemini"),
|
|
[
|
|
{"role": "system", "content": "be concise"},
|
|
{"role": "user", "content": "hello"},
|
|
{"role": "assistant", "content": "hi"},
|
|
],
|
|
)
|
|
|
|
assert result == "Gemini reply"
|
|
|
|
|
|
async def test_dashscope_native_contract():
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.headers["authorization"] == "Bearer test-key"
|
|
assert json.loads(request.content) == {
|
|
"model": "test-model",
|
|
"input": {"messages": [{"role": "user", "content": "hello"}]},
|
|
"parameters": {"result_format": "message", "temperature": 0.4},
|
|
}
|
|
return httpx.Response(
|
|
200,
|
|
json={"output": {"choices": [{"message": {"content": "DashScope reply"}}]}},
|
|
)
|
|
|
|
result = await _gateway(handler).chat(
|
|
_config("dashscope"),
|
|
[{"role": "user", "content": "hello"}],
|
|
temperature=0.4,
|
|
)
|
|
|
|
assert result == "DashScope reply"
|
|
|
|
|
|
@pytest.mark.parametrize("provider", ["anthropic", "google_gemini", "dashscope"])
|
|
async def test_chat_only_protocols_reject_embedding(provider: str):
|
|
with pytest.raises(ModelGatewayError, match="不支持 embedding 能力"):
|
|
await _gateway(lambda _: httpx.Response(500)).embed(_config(provider), "hello")
|