50 lines
2.1 KiB
Python
50 lines
2.1 KiB
Python
"""Model HTTP clients consistently opt out of ambient proxies by default."""
|
|
|
|
import ssl
|
|
|
|
import pytest
|
|
|
|
from artemis.config.settings import Settings, settings
|
|
from artemis.llm.router import ModelEndpoint, ModelFactory
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("use_proxy", [False, True])
|
|
async def test_sync_and_async_clients_follow_env_only_proxy_policy(monkeypatch, use_proxy):
|
|
monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:9")
|
|
monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.1:9")
|
|
monkeypatch.setenv("OPENAI_PROXY", "http://127.0.0.1:9")
|
|
monkeypatch.delenv("ARTEMIS_FAKE_LLM", raising=False)
|
|
monkeypatch.setattr(settings, "OPENAI_USE_SYSTEM_PROXY", use_proxy)
|
|
model = ModelFactory.create_model(
|
|
ModelEndpoint(api_key="test-only", api_base="https://example.invalid/v1", transport="http")
|
|
)
|
|
try:
|
|
for client in (model.http_client, model.http_async_client):
|
|
assert client.trust_env is use_proxy
|
|
assert bool([mount for mount in client._mounts.values() if mount]) is use_proxy
|
|
assert client._transport._pool._ssl_context.verify_mode == ssl.CERT_REQUIRED
|
|
assert model.openai_proxy == ""
|
|
finally:
|
|
model.http_client.close()
|
|
await model.http_async_client.aclose()
|
|
|
|
|
|
def test_proxy_setting_is_env_only_and_part_of_client_cache_identity(monkeypatch):
|
|
from apps.admin_console.services.model_settings_service import (
|
|
ModelSettingsUpdate,
|
|
get_model_settings,
|
|
)
|
|
|
|
monkeypatch.delenv("OPENAI_USE_SYSTEM_PROXY", raising=False)
|
|
assert Settings(_env_file=None).OPENAI_USE_SYSTEM_PROXY is False
|
|
monkeypatch.setenv("OPENAI_USE_SYSTEM_PROXY", "true")
|
|
assert Settings(_env_file=None).OPENAI_USE_SYSTEM_PROXY is True
|
|
endpoint = ModelEndpoint(api_key="test-only")
|
|
monkeypatch.setattr(settings, "OPENAI_USE_SYSTEM_PROXY", False)
|
|
direct = endpoint.cache_key()
|
|
monkeypatch.setattr(settings, "OPENAI_USE_SYSTEM_PROXY", True)
|
|
assert endpoint.cache_key() != direct
|
|
assert not any("proxy" in key.lower() for key in get_model_settings())
|
|
assert not any("proxy" in key.lower() for key in ModelSettingsUpdate.model_fields)
|