refactor(ia): unifier le contrat AI_BASE_URL entre les providers
Applique une validation structurelle partagée (https, absence de credentials et de paramètres sensibles, aucun ajout /v1) aux providers openai, litellm et openai-compatible ; homogénéise les warnings expurgés et durcit le parsing pour que la factory ne lève jamais. Refs #17
This commit is contained in:
@@ -11,7 +11,7 @@ factory de sélection, en vérifiant :
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime, time
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import TYPE_CHECKING, Any, Literal
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
@@ -29,6 +29,7 @@ from pronote_sync.models.diff import AgendaChange, AgendaChangeType, AgendaDiff
|
||||
from pronote_sync.models.message import Message, MessageType
|
||||
from pronote_sync.models.synthesis import SynthesisInput
|
||||
from pronote_sync.synthesis import get_synthesis_provider
|
||||
from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
|
||||
from pronote_sync.synthesis.openai import OpenAISynthesisProvider
|
||||
from pronote_sync.synthesis.provider import SynthesisProvider
|
||||
|
||||
@@ -872,6 +873,70 @@ def test_openai_compatible_valid_https_url_accepted() -> None:
|
||||
assert isinstance(result, OpenAISynthesisProvider)
|
||||
|
||||
|
||||
def test_openai_compatible_url_returned_unchanged() -> None:
|
||||
"""Vérifie que l'URL est retournée strictement inchangée, sans manipulation de /v1."""
|
||||
custom_url = "https://api.example.com/custom/path?query=value"
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai-compatible",
|
||||
base_url=custom_url,
|
||||
model="test-model",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert isinstance(result, OpenAISynthesisProvider)
|
||||
# OpenAI SDK appends a trailing slash to base_url, so we check the string representation
|
||||
assert str(result._client.base_url).rstrip("/") == custom_url
|
||||
|
||||
|
||||
def test_openai_url_returned_unchanged() -> None:
|
||||
"""Vérifie que l'URL est retournée strictement inchangée pour openai."""
|
||||
custom_url = "https://api.example.com/custom/path?query=value"
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url=custom_url,
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert isinstance(result, OpenAISynthesisProvider)
|
||||
# OpenAI SDK appends a trailing slash to base_url, so we check the string representation
|
||||
assert str(result._client.base_url).rstrip("/") == custom_url
|
||||
|
||||
|
||||
def test_litellm_url_returned_unchanged() -> None:
|
||||
"""Vérifie que l'URL est retournée strictement inchangée pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
custom_url = "https://api.example.com/custom/path?query=value"
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url=custom_url,
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is not None
|
||||
assert isinstance(result, LiteLLMSynthesisProvider)
|
||||
assert result._base_url == custom_url
|
||||
|
||||
|
||||
def test_openai_compatible_malformed_port_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'un port malformé est refusé pour openai-compatible."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai-compatible",
|
||||
base_url="https://host:bad/v1",
|
||||
model="test-model",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL invalide" in caplog.text
|
||||
assert "openai-compatible" in caplog.text
|
||||
|
||||
|
||||
def test_openai_compatible_http_refused_by_default(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
@@ -887,6 +952,7 @@ def test_openai_compatible_http_refused_by_default(
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
|
||||
assert "openai-compatible" in caplog.text
|
||||
|
||||
|
||||
def test_openai_compatible_http_accepted_with_allow_insecure_http() -> None:
|
||||
@@ -917,6 +983,7 @@ def test_openai_compatible_credentials_in_url_refused(
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Credentials dans l'URL refusés" in caplog.text
|
||||
assert "openai-compatible" in caplog.text
|
||||
|
||||
|
||||
def test_openai_compatible_sensitive_query_params_refused(
|
||||
@@ -933,6 +1000,24 @@ def test_openai_compatible_sensitive_query_params_refused(
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
|
||||
assert "openai-compatible" in caplog.text
|
||||
|
||||
|
||||
def test_openai_compatible_sensitive_query_params_valueless_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que les query params sensibles sans valeur sont refusés pour openai-compatible."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai-compatible",
|
||||
base_url="https://host/v1?token",
|
||||
model="test-model",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
|
||||
assert "openai-compatible" in caplog.text
|
||||
|
||||
|
||||
def test_openai_compatible_connection_error_returns_none(
|
||||
@@ -991,3 +1076,383 @@ def test_openai_compatible_factory_no_network_calls(
|
||||
assert isinstance(result, OpenAISynthesisProvider)
|
||||
mock_get.assert_not_called()
|
||||
mock_post.assert_not_called()
|
||||
|
||||
|
||||
# --- Tests de validation AI_BASE_URL pour openai et litellm (Issue #17) ---
|
||||
|
||||
|
||||
def test_openai_valid_https_base_url_accepted() -> None:
|
||||
"""Vérifie qu'une URL HTTPS valide est acceptée pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="https://api.openai.com/v1",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert isinstance(result, OpenAISynthesisProvider)
|
||||
|
||||
|
||||
def test_openai_http_refused_by_default(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que HTTP est refusé par défaut pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="http://127.0.0.1:11434/v1",
|
||||
allow_insecure_http=False,
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
|
||||
assert "openai" in caplog.text
|
||||
|
||||
|
||||
def test_openai_http_accepted_with_allow_insecure_http() -> None:
|
||||
"""Vérifie que HTTP est accepté avec allow_insecure_http=True pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="http://127.0.0.1:11434/v1",
|
||||
allow_insecure_http=True,
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert isinstance(result, OpenAISynthesisProvider)
|
||||
|
||||
|
||||
def test_openai_credentials_in_url_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que les credentials dans l'URL sont refusés pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="https://user:pass@host/v1", # pragma: allowlist secret
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Credentials dans l'URL refusés" in caplog.text
|
||||
assert "openai" in caplog.text
|
||||
|
||||
|
||||
def test_openai_sensitive_query_params_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que les query params sensibles sont refusés pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="https://host/v1?token=secret",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
|
||||
assert "openai" in caplog.text
|
||||
|
||||
|
||||
def test_openai_sensitive_query_params_valueless_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que les query params sensibles sans valeur sont refusés pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="https://host/v1?token",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
|
||||
assert "openai" in caplog.text
|
||||
|
||||
|
||||
def test_openai_malformed_url_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'une URL malformée est refusée pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="not-a-valid-url",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL invalide" in caplog.text
|
||||
assert "openai" in caplog.text
|
||||
|
||||
|
||||
def test_openai_no_hostname_url_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'une URL sans hostname est refusée pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="https:///v1",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL sans hostname" in caplog.text
|
||||
assert "openai" in caplog.text
|
||||
|
||||
|
||||
def test_openai_malformed_port_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'un port malformé est refusé pour openai."""
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="https://host:bad/v1",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL invalide" in caplog.text
|
||||
assert "openai" in caplog.text
|
||||
|
||||
|
||||
def test_openai_no_network_calls_during_validation(
|
||||
mocker: MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'aucun appel réseau n'est effectué pendant la validation pour openai."""
|
||||
mock_get = mocker.patch("requests.get")
|
||||
mock_post = mocker.patch("requests.post")
|
||||
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="openai",
|
||||
base_url="https://api.example.com/v1",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert isinstance(result, OpenAISynthesisProvider)
|
||||
mock_get.assert_not_called()
|
||||
mock_post.assert_not_called()
|
||||
|
||||
|
||||
def test_openai_sentinel_key_not_in_logs(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'une clé sentinelle est absente des logs pour openai."""
|
||||
sentinel = "sk-SENTINEL-OPENAI-BASE-URL-12345"
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr(sentinel),
|
||||
provider="openai",
|
||||
base_url="https://user:pass@host/v1", # pragma: allowlist secret
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert sentinel not in caplog.text
|
||||
|
||||
|
||||
def test_litellm_valid_https_base_url_accepted() -> None:
|
||||
"""Vérifie qu'une URL HTTPS valide est acceptée pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="https://api.litellm.ai/v1",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is not None
|
||||
|
||||
|
||||
def test_litellm_http_refused_by_default(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que HTTP est refusé par défaut pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="http://127.0.0.1:11434/v1",
|
||||
allow_insecure_http=False,
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
|
||||
assert "litellm" in caplog.text
|
||||
|
||||
|
||||
def test_litellm_http_accepted_with_allow_insecure_http() -> None:
|
||||
"""Vérifie que HTTP est accepté avec allow_insecure_http=True pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="http://127.0.0.1:11434/v1",
|
||||
allow_insecure_http=True,
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is not None
|
||||
|
||||
|
||||
def test_litellm_credentials_in_url_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que les credentials dans l'URL sont refusés pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="https://user:pass@host/v1", # pragma: allowlist secret
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Credentials dans l'URL refusés" in caplog.text
|
||||
assert "litellm" in caplog.text
|
||||
|
||||
|
||||
def test_litellm_sensitive_query_params_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que les query params sensibles sont refusés pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="https://host/v1?token=secret",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
|
||||
assert "litellm" in caplog.text
|
||||
|
||||
|
||||
def test_litellm_sensitive_query_params_valueless_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que les query params sensibles sans valeur sont refusés pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="https://host/v1?token",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
|
||||
assert "litellm" in caplog.text
|
||||
|
||||
|
||||
def test_litellm_malformed_url_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'une URL malformée est refusée pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="not-a-valid-url",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL invalide" in caplog.text
|
||||
assert "litellm" in caplog.text
|
||||
|
||||
|
||||
def test_litellm_no_hostname_url_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'une URL sans hostname est refusée pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="https:///v1",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL sans hostname" in caplog.text
|
||||
assert "litellm" in caplog.text
|
||||
|
||||
|
||||
def test_litellm_malformed_port_refused(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'un port malformé est refusé pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="https://host:bad/v1",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL invalide" in caplog.text
|
||||
assert "litellm" in caplog.text
|
||||
|
||||
|
||||
def test_litellm_no_network_calls_during_validation(
|
||||
mocker: MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'aucun appel réseau n'est effectué pendant la validation pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
mock_get = mocker.patch("requests.get")
|
||||
mock_post = mocker.patch("requests.post")
|
||||
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider="litellm",
|
||||
base_url="https://api.example.com/v1",
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is not None
|
||||
mock_get.assert_not_called()
|
||||
mock_post.assert_not_called()
|
||||
|
||||
|
||||
def test_litellm_sentinel_key_not_in_logs(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'une clé sentinelle est absente des logs pour litellm."""
|
||||
pytest.importorskip("litellm")
|
||||
sentinel = "sk-SENTINEL-LITELLM-BASE-URL-67890"
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr(sentinel),
|
||||
provider="litellm",
|
||||
base_url="https://user:pass@host/v1", # pragma: allowlist secret
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert sentinel not in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider", ["openai", "litellm", "openai-compatible"])
|
||||
def test_all_providers_http_refused_same_warning(
|
||||
provider: Literal["openai", "litellm", "openai-compatible"], caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Vérifie que tous les providers émettent le même message d'avertissement pour HTTP refusé."""
|
||||
if provider == "litellm":
|
||||
pytest.importorskip("litellm")
|
||||
settings = AISettings(
|
||||
enabled=True,
|
||||
api_key=SecretStr("test"),
|
||||
provider=provider,
|
||||
base_url="http://127.0.0.1:11434/v1",
|
||||
allow_insecure_http=False,
|
||||
model="test-model" if provider == "openai-compatible" else None,
|
||||
)
|
||||
result = get_synthesis_provider(settings)
|
||||
assert result is None
|
||||
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
|
||||
|
||||
Reference in New Issue
Block a user