e1e067bac4
Introduce ModelPurpose.ANALYSIS and a globally-unique is_analysis_default marker on chat model configs so campaign analysis can resolve its model. Service rejects disabled or non-chat configs; repo clears the previous holder on set. Documented the analysis role in CONTEXT.md.
284 lines
9.6 KiB
Python
284 lines
9.6 KiB
Python
"""Model configuration persistence, encryption, and binding rules."""
|
|
|
|
import pytest
|
|
from agenteval.models import Case, ModelPurpose, Scenario
|
|
from agenteval.services.model_configs import (
|
|
ModelConfigError,
|
|
ModelConfigInUseError,
|
|
ModelConfigService,
|
|
SecretCipher,
|
|
SecretKeyError,
|
|
)
|
|
from agenteval.storage.repository import ScenarioRepository
|
|
from cryptography.fernet import Fernet
|
|
|
|
|
|
def _service(session):
|
|
return ModelConfigService(session, SecretCipher(Fernet.generate_key().decode("ascii")))
|
|
|
|
|
|
def _create_config(
|
|
service,
|
|
*,
|
|
name="评估模型",
|
|
capability="chat",
|
|
is_default=False,
|
|
is_analysis_default=False,
|
|
api_key="secret",
|
|
):
|
|
return service.create(
|
|
name=name,
|
|
provider="openai_compatible",
|
|
capability=capability,
|
|
endpoint_url="https://models.example.com/v1/chat/completions",
|
|
model_name="test-model" if capability != "moderation" else None,
|
|
api_key=api_key,
|
|
enabled=True,
|
|
is_default=is_default,
|
|
is_analysis_default=is_analysis_default,
|
|
description="test",
|
|
)
|
|
|
|
|
|
def _scenario(config_id: str) -> Scenario:
|
|
return Scenario(
|
|
id="scenario-model-binding",
|
|
name="模型绑定测试",
|
|
cases=[Case(id="case-1", messages=["hello"])],
|
|
model_bindings={ModelPurpose.JUDGE: config_id},
|
|
)
|
|
|
|
|
|
def test_secret_cipher_round_trip_and_invalid_key():
|
|
cipher = SecretCipher(Fernet.generate_key().decode("ascii"))
|
|
encrypted = cipher.encrypt("sk-private")
|
|
assert encrypted and encrypted != "sk-private"
|
|
assert cipher.decrypt(encrypted) == "sk-private"
|
|
|
|
with pytest.raises(SecretKeyError):
|
|
SecretCipher("invalid").encrypt("value")
|
|
with pytest.raises(SecretKeyError):
|
|
SecretCipher("").encrypt("value")
|
|
|
|
|
|
def test_create_update_and_single_default(db_session):
|
|
service = _service(db_session)
|
|
first = _create_config(service, name="first", is_default=True)
|
|
second = _create_config(service, name="second", is_default=True)
|
|
|
|
assert service.require(first.id).is_default is False
|
|
assert service.require(second.id).is_default is True
|
|
encrypted = second.api_key_encrypted
|
|
|
|
updated = service.update(
|
|
second.id,
|
|
name="second",
|
|
provider="openai_compatible",
|
|
capability="chat",
|
|
endpoint_url="https://models.example.com/v1/chat/completions",
|
|
model_name="new-model",
|
|
api_key=None,
|
|
clear_api_key=False,
|
|
enabled=True,
|
|
is_default=True,
|
|
description="updated",
|
|
)
|
|
assert updated.api_key_encrypted == encrypted
|
|
assert service.resolve(second.id).api_key == "secret"
|
|
|
|
|
|
def test_binding_validation_and_reference_protection(db_session):
|
|
service = _service(db_session)
|
|
chat = _create_config(service)
|
|
embedding = _create_config(service, name="向量模型", capability="embedding")
|
|
|
|
with pytest.raises(ModelConfigError, match="不能用于 chat"):
|
|
service.validate_bindings({"judge": embedding.id})
|
|
|
|
created = ScenarioRepository(db_session).create(_scenario(chat.id))
|
|
assert created.model_bindings == {ModelPurpose.JUDGE: chat.id}
|
|
loaded = ScenarioRepository(db_session).get(created.id)
|
|
assert loaded and loaded.model_bindings[ModelPurpose.JUDGE] == chat.id
|
|
|
|
with pytest.raises(ModelConfigInUseError) as exc_info:
|
|
service.delete(chat.id)
|
|
assert exc_info.value.references[0]["scenario_id"] == created.id
|
|
|
|
assert ScenarioRepository(db_session).delete(created.id)
|
|
service.delete(chat.id)
|
|
assert service.repo.get(chat.id) is None
|
|
|
|
|
|
def test_failed_binding_does_not_create_scenario(db_session):
|
|
service = _service(db_session)
|
|
moderation = _create_config(service, name="审核模型", capability="moderation")
|
|
scenario = _scenario(moderation.id)
|
|
|
|
with pytest.raises(ModelConfigError):
|
|
ScenarioRepository(db_session).create(scenario)
|
|
assert ScenarioRepository(db_session).get(scenario.id) is None
|
|
|
|
|
|
@pytest.mark.parametrize("provider", ["anthropic", "google_gemini", "dashscope"])
|
|
def test_chat_only_protocols_accept_chat_and_reject_other_capabilities(db_session, provider):
|
|
service = _service(db_session)
|
|
created = service.create(
|
|
name=f"{provider}-chat",
|
|
provider=provider,
|
|
capability="chat",
|
|
endpoint_url=f"https://models.example.com/{provider}",
|
|
model_name="chat-model",
|
|
api_key=None,
|
|
enabled=True,
|
|
is_default=False,
|
|
description="",
|
|
)
|
|
assert created.provider == provider
|
|
|
|
with pytest.raises(ModelConfigError, match="不支持 embedding 能力"):
|
|
service.create(
|
|
name=f"{provider}-embedding",
|
|
provider=provider,
|
|
capability="embedding",
|
|
endpoint_url=f"https://models.example.com/{provider}/embeddings",
|
|
model_name="embedding-model",
|
|
api_key=None,
|
|
enabled=True,
|
|
is_default=False,
|
|
description="",
|
|
)
|
|
|
|
|
|
def test_model_metadata_validation_and_runtime_snapshot(db_session):
|
|
service = _service(db_session)
|
|
config = service.create(
|
|
name="多模态模型",
|
|
provider="openai_compatible",
|
|
capability="chat",
|
|
endpoint_url="https://models.example.com/v1/chat/completions",
|
|
model_name="vision-model",
|
|
api_key=None,
|
|
enabled=True,
|
|
is_default=False,
|
|
description="",
|
|
vendor_name="Example AI",
|
|
input_modalities=["text", "image", "image"],
|
|
output_modalities=["text"],
|
|
context_window=128000,
|
|
max_output_tokens=8192,
|
|
supports_streaming=True,
|
|
supports_tool_calling=True,
|
|
region="cn-test-1",
|
|
documentation_url="https://models.example.com/docs",
|
|
)
|
|
|
|
runtime = service.resolve(config.id)
|
|
snapshot = runtime.snapshot()
|
|
assert snapshot["vendor_name"] == "Example AI"
|
|
assert snapshot["input_modalities"] == ["text", "image"]
|
|
assert snapshot["context_window"] == 128000
|
|
assert snapshot["supports_tool_calling"] is True
|
|
assert snapshot["region"] == "cn-test-1"
|
|
assert "api_key" not in snapshot
|
|
|
|
with pytest.raises(ModelConfigError, match="不支持的模型模态"):
|
|
service.create(
|
|
name="非法模态",
|
|
provider="openai_compatible",
|
|
capability="chat",
|
|
endpoint_url="https://models.example.com/v1/chat/completions",
|
|
model_name="invalid-model",
|
|
api_key=None,
|
|
enabled=True,
|
|
is_default=False,
|
|
description="",
|
|
input_modalities=["document"],
|
|
)
|
|
|
|
with pytest.raises(ModelConfigError, match="官方文档地址"):
|
|
service.create(
|
|
name="非法文档",
|
|
provider="openai_compatible",
|
|
capability="chat",
|
|
endpoint_url="https://models.example.com/v1/chat/completions",
|
|
model_name="invalid-docs",
|
|
api_key=None,
|
|
enabled=True,
|
|
is_default=False,
|
|
description="",
|
|
documentation_url="not-a-url",
|
|
)
|
|
|
|
|
|
def _update(service, config_id, **overrides):
|
|
params = {
|
|
"name": "updated",
|
|
"provider": "openai_compatible",
|
|
"capability": "chat",
|
|
"endpoint_url": "https://models.example.com/v1/chat/completions",
|
|
"model_name": "test-model",
|
|
"api_key": None,
|
|
"clear_api_key": False,
|
|
"enabled": True,
|
|
"is_default": False,
|
|
"is_analysis_default": False,
|
|
"description": "",
|
|
}
|
|
params.update(overrides)
|
|
return service.update(config_id, **params)
|
|
|
|
|
|
def test_analysis_default_is_unique_across_configs(db_session):
|
|
service = _service(db_session)
|
|
first = _create_config(service, name="analysis-1", is_analysis_default=True)
|
|
second = _create_config(service, name="analysis-2", is_analysis_default=True)
|
|
|
|
assert service.require(first.id).is_analysis_default is False
|
|
assert service.require(second.id).is_analysis_default is True
|
|
|
|
|
|
def test_analysis_default_update_clears_previous(db_session):
|
|
service = _service(db_session)
|
|
first = _create_config(service, name="analysis-1", is_analysis_default=True)
|
|
second = _create_config(service, name="analysis-2")
|
|
|
|
_update(service, second.id, is_analysis_default=True)
|
|
|
|
assert service.require(first.id).is_analysis_default is False
|
|
assert service.require(second.id).is_analysis_default is True
|
|
|
|
|
|
def test_analysis_default_requires_chat_capability(db_session):
|
|
service = _service(db_session)
|
|
with pytest.raises(ModelConfigError, match="分析默认"):
|
|
_create_config(service, name="emb", capability="embedding", is_analysis_default=True)
|
|
|
|
|
|
def test_analysis_default_requires_enabled(db_session):
|
|
service = _service(db_session)
|
|
with pytest.raises(ModelConfigError, match="分析默认"):
|
|
service.create(
|
|
name="disabled-analysis",
|
|
provider="openai_compatible",
|
|
capability="chat",
|
|
endpoint_url="https://models.example.com/v1/chat/completions",
|
|
model_name="test-model",
|
|
api_key="secret",
|
|
enabled=False,
|
|
is_default=False,
|
|
is_analysis_default=True,
|
|
description="",
|
|
)
|
|
|
|
config = _create_config(service, name="to-disable", is_analysis_default=True)
|
|
with pytest.raises(ModelConfigError, match="分析默认"):
|
|
_update(service, config.id, enabled=False, is_analysis_default=True)
|
|
|
|
|
|
def test_scenario_binding_rejects_analysis_purpose(db_session):
|
|
service = _service(db_session)
|
|
config = _create_config(service)
|
|
with pytest.raises(ModelConfigError, match="不支持的模型用途"):
|
|
service.validate_bindings({ModelPurpose.ANALYSIS.value: config.id})
|
|
|