fix(auth): corriger les retours de revue du PIN Pronote
This commit is contained in:
+120
-3
@@ -10,10 +10,10 @@ from __future__ import annotations
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
from pydantic import SecretStr, ValidationError
|
||||
|
||||
from pronote_sync.config.env import load_settings
|
||||
from pronote_sync.config.settings import PronoteSettings, Settings
|
||||
from pronote_sync.config.settings import AppSettings, PronoteSettings, Settings
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
@@ -224,4 +224,121 @@ def test_account_pin_loaded_as_secretstr_and_redacted(monkeypatch: MonkeyPatch)
|
||||
assert "**********" in settings.model_dump_json()
|
||||
|
||||
|
||||
# Ensure trailing newline
|
||||
def test_sync_past_days_negative_direct_instantiation() -> None:
|
||||
"""Vérifie que ``sync_past_days`` négatif lève ``ValidationError`` à l'instanciation.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
with pytest.raises(ValidationError):
|
||||
AppSettings(sync_past_days=-1)
|
||||
|
||||
|
||||
def test_sync_future_days_negative_direct_instantiation() -> None:
|
||||
"""Vérifie que ``sync_future_days`` négatif lève ``ValidationError`` à l'instanciation.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
with pytest.raises(ValidationError):
|
||||
AppSettings(sync_future_days=-1)
|
||||
|
||||
|
||||
def test_sync_past_days_zero_accepted() -> None:
|
||||
"""Vérifie que ``sync_past_days=0`` est accepté.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = AppSettings(sync_past_days=0)
|
||||
assert settings.sync_past_days == 0
|
||||
|
||||
|
||||
def test_sync_future_days_zero_accepted() -> None:
|
||||
"""Vérifie que ``sync_future_days=0`` est accepté.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = AppSettings(sync_future_days=0)
|
||||
assert settings.sync_future_days == 0
|
||||
|
||||
|
||||
def test_sync_past_days_positive_accepted() -> None:
|
||||
"""Vérifie que ``sync_past_days`` positif est accepté.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = AppSettings(sync_past_days=7)
|
||||
assert settings.sync_past_days == 7
|
||||
|
||||
|
||||
def test_sync_future_days_positive_accepted() -> None:
|
||||
"""Vérifie que ``sync_future_days`` positif est accepté.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = AppSettings(sync_future_days=30)
|
||||
assert settings.sync_future_days == 30
|
||||
|
||||
|
||||
def test_sync_past_days_negative_env_loading(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``SYNC_PAST_DAYS=-1`` lève ``ValidationError`` via chargement env.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("SYNC_PAST_DAYS", "-1")
|
||||
with pytest.raises(ValidationError):
|
||||
load_settings()
|
||||
|
||||
|
||||
def test_sync_future_days_negative_env_loading(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``SYNC_FUTURE_DAYS=-1`` lève ``ValidationError`` via chargement env.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("SYNC_FUTURE_DAYS", "-1")
|
||||
with pytest.raises(ValidationError):
|
||||
load_settings()
|
||||
|
||||
|
||||
def test_sync_past_days_zero_env_loading(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``SYNC_PAST_DAYS=0`` est accepté via chargement env.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("SYNC_PAST_DAYS", "0")
|
||||
settings = load_settings()
|
||||
assert settings.app.sync_past_days == 0
|
||||
|
||||
|
||||
def test_sync_future_days_zero_env_loading(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``SYNC_FUTURE_DAYS=0`` est accepté via chargement env.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("SYNC_FUTURE_DAYS", "0")
|
||||
settings = load_settings()
|
||||
assert settings.app.sync_future_days == 0
|
||||
|
||||
|
||||
def test_sync_past_days_positive_env_loading(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``SYNC_PAST_DAYS`` positif est accepté via chargement env.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("SYNC_PAST_DAYS", "7")
|
||||
settings = load_settings()
|
||||
assert settings.app.sync_past_days == 7
|
||||
|
||||
|
||||
def test_sync_future_days_positive_env_loading(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``SYNC_FUTURE_DAYS`` positif est accepté via chargement env.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("SYNC_FUTURE_DAYS", "30")
|
||||
settings = load_settings()
|
||||
assert settings.app.sync_future_days == 30
|
||||
|
||||
@@ -9,6 +9,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import traceback
|
||||
from collections.abc import Generator
|
||||
from contextlib import contextmanager
|
||||
from datetime import date, datetime
|
||||
@@ -621,12 +622,17 @@ def test_get_messages_degraded_on_error(
|
||||
|
||||
|
||||
def test_get_informations_degraded_on_error(
|
||||
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que get_informations retourne une liste vide en cas d'erreur réseau.
|
||||
|
||||
Assert que le chemin d'erreur retourne toujours [] avec un log ERROR.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param pronote_settings: Paramètres Pronote valides.
|
||||
:param caplog: Fixture pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
mock_client = mocker.MagicMock()
|
||||
@@ -634,22 +640,265 @@ def test_get_informations_degraded_on_error(
|
||||
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||
|
||||
client = PronoteClient(pronote_settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
with caplog.at_level(logging.ERROR, logger="pronote_sync.sources.pronote.client"):
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
# Assert ERROR log is present
|
||||
error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
|
||||
assert len(error_records) >= 1
|
||||
assert any(
|
||||
"Échec de la récupération des informations Pronote" in r.message for r in error_records
|
||||
)
|
||||
|
||||
|
||||
# --- QR code / token authentication tests ---
|
||||
|
||||
|
||||
# --- get_informations qr_token mode guard tests ---
|
||||
|
||||
|
||||
def test_get_informations_skips_in_qr_token_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie que get_informations retourne [] immédiatement en mode qr_token.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
client = PronoteClient(settings)
|
||||
assert client._client is None
|
||||
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
assert client._client is None
|
||||
|
||||
|
||||
def test_get_informations_no_connect_in_qr_token_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie que _connect n'est jamais appelé en mode qr_token pour get_informations.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
connect_spy = mocker.spy(PronoteClient, "_connect")
|
||||
|
||||
client = PronoteClient(settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
connect_spy.assert_not_called()
|
||||
|
||||
|
||||
def test_get_informations_no_information_and_surveys_in_qr_token_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie que information_and_surveys n'est jamais appelé en mode qr_token.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_client.information_and_surveys = mocker.MagicMock()
|
||||
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||
|
||||
client = PronoteClient(settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
mock_client.information_and_surveys.assert_not_called()
|
||||
|
||||
|
||||
def test_get_informations_qr_token_no_side_effects(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Prouve le contrat complet sans effet de bordure du chemin de saut qr_token.
|
||||
|
||||
Avec PronoteAuthState et _qr_token_operation_lock et _persist_credentials
|
||||
mockés, assert que sur le saut : le gestionnaire de contexte de verrou n'est
|
||||
PAS entré, auth-state load() n'est PAS appelé, _persist_credentials() n'est
|
||||
PAS appelé, et l'export des credentials n'est PAS invoqué. Assert aussi que
|
||||
self._client est inchangé.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param caplog: Fixture pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load = mocker.MagicMock()
|
||||
auth_state.lock = mocker.MagicMock()
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
sentinel = MagicMock()
|
||||
client._client = sentinel
|
||||
persist_spy = mocker.spy(client, "_persist_credentials")
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="pronote_sync.sources.pronote.client"):
|
||||
messages = client.get_informations()
|
||||
|
||||
# Assert no side effects
|
||||
assert messages == []
|
||||
assert client._client is sentinel
|
||||
assert isinstance(sentinel, MagicMock)
|
||||
sentinel.export_credentials.assert_not_called()
|
||||
persist_spy.assert_not_called()
|
||||
auth_state.load.assert_not_called()
|
||||
auth_state.lock.assert_not_called()
|
||||
|
||||
# Assert exactly one INFO log record with the exact message
|
||||
info_records = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||
assert len(info_records) == 1
|
||||
assert (
|
||||
info_records[0].message
|
||||
== "Récupération des informations Pronote ignorée : endpoint PageActualites "
|
||||
"indisponible en mode d'authentification qr_token."
|
||||
)
|
||||
# Assert no secret sentinel appears in any log record
|
||||
for record in caplog.records:
|
||||
assert "testpass" not in record.message
|
||||
assert "testuser" not in record.message
|
||||
assert "pronote.example.com" not in record.message
|
||||
|
||||
|
||||
def test_get_informations_logs_info_in_qr_token_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que get_informations log un message INFO exact en mode qr_token.
|
||||
|
||||
Assert exactement un enregistrement logging.INFO avec le message exact
|
||||
(inspection de caplog.records, pas seulement caplog.text), et qu'aucune
|
||||
sentinelle de secret n'apparaît.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param caplog: Fixture pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="pronote_sync.sources.pronote.client"):
|
||||
client = PronoteClient(settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
# Assert exactly one INFO record with the exact message
|
||||
info_records = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||
assert len(info_records) == 1
|
||||
assert (
|
||||
info_records[0].message
|
||||
== "Récupération des informations Pronote ignorée : endpoint PageActualites "
|
||||
"indisponible en mode d'authentification qr_token."
|
||||
)
|
||||
# Assert no secret sentinel appears in any record
|
||||
for record in caplog.records:
|
||||
assert "testpass" not in record.message
|
||||
assert "testuser" not in record.message
|
||||
assert "pronote.example.com" not in record.message
|
||||
|
||||
|
||||
def test_get_informations_unchanged_in_password_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
) -> None:
|
||||
"""Vérifie que get_informations en mode password reste inchangé (régression).
|
||||
|
||||
Assert que _connect() A ÉTÉ appelé et information_and_surveys() A ÉTÉ appelé
|
||||
(en cas de succès), en conservant les assertions de mappage existantes.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param pronote_settings: Paramètres Pronote valides en mode password.
|
||||
:return: None
|
||||
"""
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_info = mocker.MagicMock()
|
||||
mock_info.id = "info-456"
|
||||
mock_info.title = "Important Info"
|
||||
mock_info.content.return_value = "Important content"
|
||||
mock_info.author = "Admin"
|
||||
mock_info.creation_date = datetime(2024, 9, 2, 14, 30, 0)
|
||||
mock_info.read = False
|
||||
mock_info.survey = True
|
||||
mock_client.information_and_surveys.return_value = [mock_info]
|
||||
|
||||
# Patch _connect to return mock_client and track calls
|
||||
connect_patch = mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||
|
||||
client = PronoteClient(pronote_settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
# Assert _connect() WAS called and information_and_surveys() WAS called
|
||||
connect_patch.assert_called_once()
|
||||
mock_client.information_and_surveys.assert_called_once()
|
||||
|
||||
# Retain existing mapping assertions
|
||||
assert isinstance(messages, list)
|
||||
assert len(messages) == 1
|
||||
message = messages[0]
|
||||
assert isinstance(message, Message)
|
||||
assert message.id == "info-456"
|
||||
assert message.type == MessageType.SURVEY
|
||||
assert message.title == "Important Info"
|
||||
assert message.content == "Important content"
|
||||
assert message.author == "Admin"
|
||||
assert message.date == datetime(2024, 9, 2, 14, 30, 0)
|
||||
assert message.read is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("account_pin", [None, "account-pin-42"])
|
||||
def test_connect_password_mode_unchanged(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
account_pin: str | None,
|
||||
) -> None:
|
||||
"""Vérifie que le mode password conserve le comportement historique.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param pronote_settings: Paramètres Pronote valides en mode password.
|
||||
:param account_pin: PIN de compte éventuellement configuré.
|
||||
:return: None
|
||||
"""
|
||||
from unittest.mock import Mock
|
||||
@@ -659,7 +908,10 @@ def test_connect_password_mode_unchanged(
|
||||
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
|
||||
mocker.patch("pronotepy.Client")
|
||||
|
||||
client = PronoteClient(pronote_settings, auth_state=None)
|
||||
settings = pronote_settings.model_copy(
|
||||
update={"account_pin": SecretStr(account_pin) if account_pin is not None else None}
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=None)
|
||||
connected = client._connect()
|
||||
|
||||
assert connected is mock_client
|
||||
@@ -698,10 +950,26 @@ def test_connect_qr_token_with_persisted_creds(
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = creds
|
||||
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_client: MagicMock = mocker.MagicMock()
|
||||
mock_client.logged_in = True
|
||||
mock_client.export_credentials.return_value = rotated_creds
|
||||
mocker.patch("pronotepy.ParentClient.token_login", return_value=mock_client)
|
||||
|
||||
def token_login_fake(
|
||||
pronote_url: str,
|
||||
username: str,
|
||||
password: str,
|
||||
uuid: str,
|
||||
account_pin: str | None = None,
|
||||
client_identifier: str | None = None,
|
||||
device_name: str | None = None,
|
||||
) -> MagicMock:
|
||||
"""Retourne le faux client avec la signature pronotepy 2.15.7."""
|
||||
del pronote_url, username, password, uuid, account_pin, client_identifier, device_name
|
||||
return mock_client
|
||||
|
||||
token_login = mocker.patch(
|
||||
"pronotepy.ParentClient.token_login", autospec=True, side_effect=token_login_fake
|
||||
)
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
@@ -716,9 +984,7 @@ def test_connect_qr_token_with_persisted_creds(
|
||||
connected = client._connect()
|
||||
|
||||
assert connected is mock_client
|
||||
pronotepy.ParentClient.token_login.assert_called_once_with( # type: ignore[attr-defined]
|
||||
**creds, account_pin=account_pin
|
||||
)
|
||||
token_login.assert_called_once_with(**creds, account_pin=account_pin)
|
||||
auth_state.save.assert_called_once_with(rotated_creds)
|
||||
|
||||
|
||||
@@ -755,10 +1021,26 @@ def test_connect_qr_token_no_creds_with_qr_code(
|
||||
"password": "new-token", # pragma: allowlist secret
|
||||
"uuid": "new-uuid",
|
||||
}
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_client: MagicMock = mocker.MagicMock()
|
||||
mock_client.logged_in = True
|
||||
mock_client.export_credentials.return_value = creds
|
||||
mocker.patch("pronotepy.ParentClient.qrcode_login", return_value=mock_client)
|
||||
|
||||
def qrcode_login_fake(
|
||||
qr_code: dict[str, str],
|
||||
pin: str,
|
||||
uuid: str,
|
||||
account_pin: str | None = None,
|
||||
client_identifier: str | None = None,
|
||||
device_name: str | None = None,
|
||||
skip_2fa: bool = False,
|
||||
) -> MagicMock:
|
||||
"""Retourne le faux client avec la signature pronotepy 2.15.7."""
|
||||
del qr_code, pin, uuid, account_pin, client_identifier, device_name, skip_2fa
|
||||
return mock_client
|
||||
|
||||
qrcode_login = mocker.patch(
|
||||
"pronotepy.ParentClient.qrcode_login", autospec=True, side_effect=qrcode_login_fake
|
||||
)
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
@@ -775,9 +1057,8 @@ def test_connect_qr_token_no_creds_with_qr_code(
|
||||
connected = client._connect()
|
||||
|
||||
assert connected is mock_client
|
||||
qrcode_login = pronotepy.ParentClient.qrcode_login
|
||||
qrcode_login.assert_called_once() # type: ignore[attr-defined]
|
||||
kwargs = qrcode_login.call_args.kwargs # type: ignore[attr-defined]
|
||||
qrcode_login.assert_called_once()
|
||||
kwargs = qrcode_login.call_args.kwargs
|
||||
assert kwargs["pin"] == "123456"
|
||||
assert kwargs["account_pin"] == account_pin
|
||||
assert kwargs["qr_code"] == {
|
||||
@@ -1156,6 +1437,12 @@ def test_no_secrets_in_rotation_error_messages(
|
||||
assert sentinel_pin not in caplog.text
|
||||
assert sentinel_token not in caplog.text
|
||||
assert "sentinel-url" not in caplog.text
|
||||
assert exc_info.value.__cause__ is None
|
||||
assert exc_info.value.__context__ is None
|
||||
formatted_traceback = "".join(traceback.format_exception(exc_info.value))
|
||||
assert sentinel_pin not in formatted_traceback
|
||||
assert sentinel_token not in formatted_traceback
|
||||
assert "sentinel-url" not in formatted_traceback
|
||||
|
||||
|
||||
def test_no_raw_secrets_in_logs(
|
||||
@@ -1234,6 +1521,13 @@ def test_no_raw_secrets_in_logs(
|
||||
assert sentinel_pin not in caplog.text
|
||||
assert sentinel_jeton not in caplog.text
|
||||
assert sentinel_account_pin not in caplog.text
|
||||
assert exc_info.value.__cause__ is None
|
||||
assert exc_info.value.__context__ is None
|
||||
formatted_traceback = "".join(traceback.format_exception(exc_info.value))
|
||||
assert sentinel_token not in formatted_traceback
|
||||
assert sentinel_pin not in formatted_traceback
|
||||
assert sentinel_jeton not in formatted_traceback
|
||||
assert sentinel_account_pin not in formatted_traceback
|
||||
|
||||
|
||||
# --- Persistence of credentials after data operations ---
|
||||
|
||||
@@ -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