AgentEvalTool/tests/integration/test_model_configs_api.py

"""Integration tests for the centralized model configuration API."""

from pathlib import Path

import pytest
from agenteval.config import get_settings
from agenteval.storage.db import ModelConfigDB
from agenteval.web.app import app
from agenteval.web.deps import get_db
from cryptography.fernet import Fernet
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine


@pytest.fixture()
def model_client(tmp_path: Path, monkeypatch):
    engine = create_engine(
        f"sqlite:///{tmp_path / 'models_api.db'}",
        connect_args={"check_same_thread": False},
    )
    SQLModel.metadata.create_all(engine)
    session = Session(engine)

    from agenteval.web import app as app_module

    monkeypatch.setattr(app_module, "init_db", lambda: None)
    monkeypatch.setattr(get_settings(), "secret_key", Fernet.generate_key().decode("ascii"))

    def override_get_db():
        yield session

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as client:
        yield client, session

    app.dependency_overrides.clear()
    session.close()
    engine.dispose()


def _payload(**overrides) -> dict:
    payload = {
        "name": "评估对话模型",
        "provider": "openai_compatible",
        "capability": "chat",
        "endpoint_url": "https://models.example.com/v1/chat/completions",
        "model_name": "judge-model",
        "api_key": "sk-private",
        "enabled": True,
        "is_default": True,
        "description": "integration test",
    }
    payload.update(overrides)
    return payload


def test_model_config_crud_never_returns_secret(model_client):
    client, session = model_client

    created_response = client.post("/api/model-configs", json=_payload())
    assert created_response.status_code == 200
    created = created_response.json()
    assert created["has_api_key"] is True
    assert "api_key" not in created
    assert "api_key_encrypted" not in created
    assert "sk-private" not in created_response.text

    stored = session.get(ModelConfigDB, created["id"])
    assert stored and stored.api_key_encrypted
    assert stored.api_key_encrypted != "sk-private"

    listed = client.get("/api/model-configs").json()
    assert len(listed) == 1
    assert "sk-private" not in str(listed)

    updated_response = client.put(
        f"/api/model-configs/{created['id']}",
        json=_payload(api_key=None, clear_api_key=False, model_name="judge-model-v2"),
    )
    assert updated_response.status_code == 200
    assert updated_response.json()["has_api_key"] is True
    assert updated_response.json()["model_name"] == "judge-model-v2"

    assert client.delete(f"/api/model-configs/{created['id']}").status_code == 200
    assert client.get(f"/api/model-configs/{created['id']}").status_code == 404


def test_referenced_config_cannot_be_deleted(model_client):
    client, _ = model_client
    config = client.post("/api/model-configs", json=_payload(api_key=None)).json()
    scenario = {
        "name": "引用模型的场景",
        "cases": [{"id": "case-1", "messages": ["hello"]}],
        "model_bindings": {"judge": config["id"]},
    }

    scenario_response = client.post("/api/scenarios", json=scenario)
    assert scenario_response.status_code == 200
    scenario_id = scenario_response.json()["id"]

    references = client.get(f"/api/model-configs/{config['id']}/references").json()
    assert references == [{"scenario_id": scenario_id, "scenario_name": "引用模型的场景", "purpose": "judge"}]

    delete_response = client.delete(f"/api/model-configs/{config['id']}")
    assert delete_response.status_code == 409


def test_scenario_rejects_capability_mismatch(model_client):
    client, _ = model_client
    embedding = client.post(
        "/api/model-configs",
        json=_payload(
            name="向量模型",
            capability="embedding",
            endpoint_url="https://models.example.com/v1/embeddings",
        ),
    ).json()

    response = client.post(
        "/api/scenarios",
        json={
            "name": "错误绑定",
            "cases": [{"id": "case-1", "messages": ["hello"]}],
            "model_bindings": {"judge": embedding["id"]},
        },
    )
    assert response.status_code == 400
    assert "不能用于 chat" in response.json()["detail"]


def test_model_config_api_supports_mainstream_protocols_and_validates_capability(model_client):
    client, _ = model_client

    for provider in ("anthropic", "google_gemini", "dashscope"):
        response = client.post(
            "/api/model-configs",
            json=_payload(
                name=f"{provider}-chat",
                provider=provider,
                endpoint_url=f"https://models.example.com/{provider}",
                api_key=None,
                is_default=False,
            ),
        )
        assert response.status_code == 200
        assert response.json()["provider"] == provider

    mismatch = client.post(
        "/api/model-configs",
        json=_payload(
            name="anthropic-embedding",
            provider="anthropic",
            capability="embedding",
            endpoint_url="https://models.example.com/anthropic/embeddings",
            api_key=None,
            is_default=False,
        ),
    )
    assert mismatch.status_code == 400
    assert "不支持 embedding 能力" in mismatch.json()["detail"]