File size: 4,449 Bytes
31c8075 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | import os
import pytest
from arena.agent import _load_dotenv, _normalize_openrouter_model, create_session
from arena.model_catalog import ModelProviderConfig
from arena.model_provider import create_chat_model
def test_create_session_returns_local_provider_with_env() -> None:
os.environ["DEEPAGENT_SANDBOX_PROVIDER"] = "local"
try:
session = create_session("test-session")
finally:
os.environ.pop("DEEPAGENT_SANDBOX_PROVIDER", None)
assert session.provider == "local"
assert session.thread_id == "test-session"
assert session.workspace_dir == "/workspace"
session.close()
def test_session_creates_composite_backend() -> None:
os.environ["DEEPAGENT_SANDBOX_PROVIDER"] = "local"
try:
session = create_session("test-composite")
finally:
os.environ.pop("DEEPAGENT_SANDBOX_PROVIDER", None)
from deepagents.backends import CompositeBackend
assert isinstance(session.backend, CompositeBackend)
assert session.workspace_dir == "/workspace"
assert session.agent is not None
session.close()
def test_disallowed_provider_raises() -> None:
os.environ["DEEPAGENT_SANDBOX_PROVIDER"] = "nonexistent"
try:
with pytest.raises(ValueError, match="Unsupported sandbox provider"):
create_session("test-bad")
finally:
os.environ.pop("DEEPAGENT_SANDBOX_PROVIDER", None)
def test_default_model_is_used_when_not_set() -> None:
os.environ["DEEPAGENT_SANDBOX_PROVIDER"] = "local"
os.environ.pop("DEEPAGENT_MODEL", None)
try:
session = create_session("test-default-model")
finally:
os.environ.pop("DEEPAGENT_SANDBOX_PROVIDER", None)
assert session.agent is not None
session.close()
@pytest.mark.parametrize(
("configured", "expected"),
[
("openrouter/qwen/qwen-2.5-7b-instruct", "qwen/qwen-2.5-7b-instruct"),
("openrouter:qwen/qwen-2.5-7b-instruct", "qwen/qwen-2.5-7b-instruct"),
("qwen/qwen-2.5-7b-instruct", "qwen/qwen-2.5-7b-instruct"),
],
)
def test_openrouter_model_normalizes_legacy_prefixes(configured: str, expected: str) -> None:
assert _normalize_openrouter_model(configured) == expected
def test_create_chat_model_uses_custom_openai_compatible_endpoint() -> None:
model = create_chat_model(
ModelProviderConfig(
provider="custom",
model="custom-model",
api_key="custom-key",
base_url="https://example.test/v1",
)
)
assert getattr(model, "model_name", getattr(model, "model", "")) == "custom-model"
assert str(model.model_dump()["openai_api_base"]).rstrip("/") == "https://example.test/v1"
def test_create_chat_model_uses_openrouter_deepagents_supported_provider() -> None:
model = create_chat_model(
ModelProviderConfig(
provider="openrouter",
model="qwen/qwen3.5-9b",
api_key="user-key",
)
)
assert type(model).__name__ == "ChatOpenRouter"
assert getattr(model, "model_name", getattr(model, "model", "")) == "qwen/qwen3.5-9b"
def test_create_chat_model_uses_cached_local_model_without_endpoint() -> None:
model = create_chat_model(
ModelProviderConfig(
provider="local",
model="unsloth/gemma-4-12B-it-qat-GGUF",
)
)
assert type(model).__name__ == "CachedLocalChatModel"
assert model.model_id == "unsloth/gemma-4-12B-it-qat-GGUF"
def test_create_chat_model_uses_openai_compatible_local_endpoint() -> None:
model = create_chat_model(
ModelProviderConfig(
provider="local",
model="local-model",
base_url="http://127.0.0.1:8000/v1",
)
)
assert type(model).__name__ == "ChatOpenAI"
assert str(model.model_dump()["openai_api_base"]).rstrip("/") == "http://127.0.0.1:8000/v1"
def test_per_run_openrouter_config_does_not_fall_back_to_server_key(monkeypatch) -> None:
monkeypatch.setenv("OPENROUTER_API_KEY", "server-secret")
with pytest.raises(Exception, match="OPENROUTER_API_KEY must be set"):
create_chat_model(
ModelProviderConfig(
provider="openrouter",
model="qwen/qwen3.5-9b",
)
)
def test_load_dotenv_does_not_override_existing_env() -> None:
os.environ["TEST_KEY"] = "from-env"
try:
_load_dotenv()
finally:
os.environ.pop("TEST_KEY", None)
|