Antoine Van Elstraete
92833060e2
Synthèse optionnelle via SDK openai (client injectable, prompt système FR, max 800 car., timeout 30 s, temp 0.3). Mode dégradé strict : generate() ne lève jamais, retourne None si clé absente/timeout/erreur. Provider litellm optionnel (extra ai-litellm) réutilisant le prompt OpenAI. Factory get_synthesis_provider() selon AISettings. 23 tests sans réseau, couverture synthesis/ 93%. Co-authored-by: opencode/coder <coder@agents.invalid> Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
524 lines
18 KiB
Python
524 lines
18 KiB
Python
"""Tests unitaires pour le module de synthèse IA (M9).
|
|
|
|
Ce module teste les fournisseurs de synthèse IA (OpenAI, LiteLLM) et la
|
|
factory de sélection, en vérifiant :
|
|
- La construction du prompt à partir des données d'entrée.
|
|
- Le comportement dégradé (retour ``None``) en cas d'erreur.
|
|
- L'absence de fuite de secrets dans les logs.
|
|
- La troncature et le nettoyage des réponses.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, time
|
|
from typing import TYPE_CHECKING, Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from pydantic import SecretStr
|
|
|
|
from pronote_sync.config.settings import AISettings
|
|
from pronote_sync.models.agenda import (
|
|
Lesson,
|
|
LessonStatus,
|
|
SchoolEvent,
|
|
SchoolEventKind,
|
|
TheoreticalLesson,
|
|
)
|
|
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.openai import OpenAISynthesisProvider
|
|
from pronote_sync.synthesis.provider import SynthesisProvider
|
|
|
|
if TYPE_CHECKING:
|
|
from pytest_mock import MockerFixture
|
|
|
|
|
|
# --- Fixtures ---
|
|
|
|
|
|
@pytest.fixture
|
|
def target_date() -> date:
|
|
"""Date cible pour les tests."""
|
|
return date(2025, 9, 15)
|
|
|
|
|
|
@pytest.fixture
|
|
def empty_input(target_date: date) -> SynthesisInput:
|
|
"""Entrée de synthèse vide (sans agenda_diff, messages ou événements)."""
|
|
return SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
|
|
|
|
@pytest.fixture
|
|
def lesson() -> Lesson:
|
|
"""Cours pour les tests."""
|
|
return Lesson(
|
|
id="lesson-1",
|
|
start=datetime(2025, 9, 15, 8, 0),
|
|
end=datetime(2025, 9, 15, 9, 0),
|
|
subject="Mathématiques",
|
|
teachers=("M. Dupont",),
|
|
rooms=("Salle 101",),
|
|
group=None,
|
|
status=LessonStatus.NORMAL,
|
|
content=None,
|
|
homework_blocks=(),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def theoretical_lesson() -> TheoreticalLesson:
|
|
"""Cours théorique pour les tests."""
|
|
return TheoreticalLesson(
|
|
id="theoretical-1",
|
|
day_of_week=0,
|
|
start_time=time(8, 0),
|
|
end_time=time(9, 0),
|
|
subject="Mathématiques",
|
|
teachers=("M. Dupont",),
|
|
rooms=("Salle 101",),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def agenda_diff_added(lesson: Lesson, target_date: date) -> AgendaDiff:
|
|
"""AgendaDiff avec un cours ajouté."""
|
|
return AgendaDiff(
|
|
target_date=target_date,
|
|
changes=(
|
|
AgendaChange(type=AgendaChangeType.ADDED, lesson=lesson, theoretical_lesson=None),
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def agenda_diff_removed(theoretical_lesson: TheoreticalLesson, target_date: date) -> AgendaDiff:
|
|
"""AgendaDiff avec un cours supprimé."""
|
|
return AgendaDiff(
|
|
target_date=target_date,
|
|
changes=(
|
|
AgendaChange(
|
|
type=AgendaChangeType.REMOVED,
|
|
lesson=None,
|
|
theoretical_lesson=theoretical_lesson,
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def agenda_diff_modified(
|
|
lesson: Lesson, theoretical_lesson: TheoreticalLesson, target_date: date
|
|
) -> AgendaDiff:
|
|
"""AgendaDiff avec un cours modifié."""
|
|
return AgendaDiff(
|
|
target_date=target_date,
|
|
changes=(
|
|
AgendaChange(
|
|
type=AgendaChangeType.MODIFIED,
|
|
lesson=lesson,
|
|
theoretical_lesson=theoretical_lesson,
|
|
details="Changement de salle",
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def unread_message() -> Message:
|
|
"""Message non lu pour les tests."""
|
|
return Message(
|
|
id="msg-1",
|
|
type=MessageType.INFORMATION,
|
|
title="Réunion",
|
|
content="Réunion à 14h",
|
|
author="M. Martin",
|
|
date=datetime(2025, 9, 14, 10, 0),
|
|
read=False,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def read_message() -> Message:
|
|
"""Message lu pour les tests."""
|
|
return Message(
|
|
id="msg-2",
|
|
type=MessageType.INFORMATION,
|
|
title="Ancien message",
|
|
content="Contenu ancien",
|
|
author="M. Martin",
|
|
date=datetime(2025, 9, 10, 10, 0),
|
|
read=True,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def school_event() -> SchoolEvent:
|
|
"""Événement scolaire pour les tests."""
|
|
return SchoolEvent(
|
|
kind=SchoolEventKind.HOLIDAY,
|
|
label="Vacances de Noël",
|
|
from_date=date(2025, 12, 20),
|
|
to_date=date(2026, 1, 5),
|
|
)
|
|
|
|
|
|
# --- OpenAISynthesisProvider._build_prompt tests ---
|
|
|
|
|
|
def test_build_prompt_empty_input(empty_input: SynthesisInput) -> None:
|
|
"""Vérifie que _build_prompt retourne le message par défaut pour une entrée vide."""
|
|
result = OpenAISynthesisProvider._build_prompt(empty_input)
|
|
assert result == "Aucune information importante à signaler."
|
|
|
|
|
|
def test_build_prompt_with_added_lesson(lesson: Lesson, target_date: date) -> None:
|
|
"""Vérifie que _build_prompt inclut les cours ajoutés."""
|
|
input_data = SynthesisInput(
|
|
target_date=target_date,
|
|
agenda_diff=AgendaDiff(
|
|
target_date=target_date,
|
|
changes=(
|
|
AgendaChange(type=AgendaChangeType.ADDED, lesson=lesson, theoretical_lesson=None),
|
|
),
|
|
),
|
|
)
|
|
result = OpenAISynthesisProvider._build_prompt(input_data)
|
|
assert "Cours ajouté : Mathématiques" in result
|
|
assert f"Date cible : {target_date.strftime('%d/%m/%Y')}" in result
|
|
|
|
|
|
def test_build_prompt_with_removed_lesson(
|
|
theoretical_lesson: TheoreticalLesson, target_date: date
|
|
) -> None:
|
|
"""Vérifie que _build_prompt inclut les cours supprimés."""
|
|
input_data = SynthesisInput(
|
|
target_date=target_date,
|
|
agenda_diff=AgendaDiff(
|
|
target_date=target_date,
|
|
changes=(
|
|
AgendaChange(
|
|
type=AgendaChangeType.REMOVED,
|
|
lesson=None,
|
|
theoretical_lesson=theoretical_lesson,
|
|
),
|
|
),
|
|
),
|
|
)
|
|
result = OpenAISynthesisProvider._build_prompt(input_data)
|
|
assert "Cours supprimé : Mathématiques" in result
|
|
|
|
|
|
def test_build_prompt_with_modified_lesson(
|
|
lesson: Lesson, theoretical_lesson: TheoreticalLesson, target_date: date
|
|
) -> None:
|
|
"""Vérifie que _build_prompt inclut les cours modifiés avec détails."""
|
|
input_data = SynthesisInput(
|
|
target_date=target_date,
|
|
agenda_diff=AgendaDiff(
|
|
target_date=target_date,
|
|
changes=(
|
|
AgendaChange(
|
|
type=AgendaChangeType.MODIFIED,
|
|
lesson=lesson,
|
|
theoretical_lesson=theoretical_lesson,
|
|
details="Changement de salle",
|
|
),
|
|
),
|
|
),
|
|
)
|
|
result = OpenAISynthesisProvider._build_prompt(input_data)
|
|
assert "Cours modifié : Mathématiques (Changement de salle)" in result
|
|
|
|
|
|
def test_build_prompt_with_unread_messages(
|
|
unread_message: Message, read_message: Message, target_date: date
|
|
) -> None:
|
|
"""Vérifie que _build_prompt inclut uniquement les messages non lus."""
|
|
input_data = SynthesisInput(
|
|
target_date=target_date,
|
|
agenda_diff=None,
|
|
messages=[unread_message, read_message],
|
|
)
|
|
result = OpenAISynthesisProvider._build_prompt(input_data)
|
|
assert f"Message de {unread_message.author}: {unread_message.title}" in result
|
|
assert f"Message de {read_message.author}: {read_message.title}" not in result
|
|
|
|
|
|
def test_build_prompt_with_school_events(school_event: SchoolEvent, target_date: date) -> None:
|
|
"""Vérifie que _build_prompt formate correctement les événements scolaires."""
|
|
input_data = SynthesisInput(
|
|
target_date=target_date,
|
|
agenda_diff=None,
|
|
school_events=[school_event],
|
|
)
|
|
result = OpenAISynthesisProvider._build_prompt(input_data)
|
|
assert f"{school_event.label} du {school_event.from_date.strftime('%d/%m')}" in result
|
|
|
|
|
|
# --- OpenAISynthesisProvider.generate tests ---
|
|
|
|
|
|
def test_generate_success(mocker: MockerFixture, target_date: date) -> None:
|
|
"""Vérifie que generate retourne SynthesisResult en cas de succès."""
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "Synthèse OK."
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
provider = OpenAISynthesisProvider(api_key="test-key", client=mock_client)
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is not None
|
|
assert result.text == "Synthèse OK."
|
|
|
|
|
|
def test_generate_returns_none_on_empty_response(mocker: MockerFixture, target_date: date) -> None:
|
|
"""Vérifie que generate retourne None si la réponse est vide."""
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = None
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
provider = OpenAISynthesisProvider(api_key="test-key", client=mock_client)
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is None
|
|
|
|
|
|
def test_generate_returns_none_on_empty_string_response(
|
|
mocker: MockerFixture, target_date: date
|
|
) -> None:
|
|
"""Vérifie que generate retourne None si la réponse est une chaîne vide."""
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = ""
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
provider = OpenAISynthesisProvider(api_key="test-key", client=mock_client)
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is None
|
|
|
|
|
|
def test_generate_truncates_to_max_length(mocker: MockerFixture, target_date: date) -> None:
|
|
"""Vérifie que generate tronque la réponse à MAX_LENGTH."""
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
long_content = "A" * 1000
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = long_content
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
provider = OpenAISynthesisProvider(api_key="test-key", client=mock_client)
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is not None
|
|
assert result.text is not None
|
|
assert result.text == "A" * 800
|
|
assert len(result.text) == OpenAISynthesisProvider.MAX_LENGTH
|
|
|
|
|
|
def test_generate_strips_whitespace(mocker: MockerFixture, target_date: date) -> None:
|
|
"""Vérifie que generate supprime les espaces en début et fin."""
|
|
mock_client = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "\n Synthèse \n"
|
|
mock_client.chat.completions.create.return_value = mock_response
|
|
|
|
provider = OpenAISynthesisProvider(api_key="test-key", client=mock_client)
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is not None
|
|
assert result.text == "Synthèse"
|
|
|
|
|
|
def test_generate_returns_none_on_exception(
|
|
mocker: MockerFixture, target_date: date, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
"""Vérifie que generate retourne None en cas d'exception et journalise l'erreur."""
|
|
mock_client = MagicMock()
|
|
mock_client.chat.completions.create.side_effect = Exception("timeout")
|
|
|
|
provider = OpenAISynthesisProvider(api_key="test-key", client=mock_client)
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is None
|
|
assert "Échec de la génération de la synthèse IA" in caplog.text
|
|
|
|
|
|
def test_generate_does_not_leak_api_key(
|
|
mocker: MockerFixture, target_date: date, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
"""Vérifie que generate ne fuite pas l'api_key dans les logs."""
|
|
sentinel = "sk-secret-12345"
|
|
mock_client = MagicMock()
|
|
mock_client.chat.completions.create.side_effect = Exception(f"key={sentinel}")
|
|
|
|
provider = OpenAISynthesisProvider(api_key=sentinel, client=mock_client)
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is None
|
|
assert sentinel not in caplog.text
|
|
assert "REDACTED" in caplog.text
|
|
|
|
|
|
# --- LiteLLMSynthesisProvider.generate tests ---
|
|
|
|
|
|
def test_litellm_generate_success(mocker: MockerFixture, target_date: date) -> None:
|
|
"""Vérifie que LiteLLMSynthesisProvider.generate retourne SynthesisResult en cas de succès."""
|
|
from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
|
|
|
|
mock_completion = mocker.patch("litellm.completion")
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "Synthèse litellm."
|
|
mock_completion.return_value = mock_response
|
|
|
|
provider = LiteLLMSynthesisProvider(api_key="test-key", model="gpt-4o-mini")
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is not None
|
|
assert result.text == "Synthèse litellm."
|
|
|
|
|
|
def test_litellm_generate_passes_api_key_and_timeout(
|
|
mocker: MockerFixture, target_date: date
|
|
) -> None:
|
|
"""Vérifie que LiteLLMSynthesisProvider.generate passe api_key et timeout."""
|
|
from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
|
|
|
|
mock_completion = mocker.patch("litellm.completion")
|
|
mock_response = MagicMock()
|
|
mock_response.choices = [MagicMock()]
|
|
mock_response.choices[0].message.content = "Synthèse litellm."
|
|
mock_completion.return_value = mock_response
|
|
|
|
provider = LiteLLMSynthesisProvider(
|
|
api_key="test-key", # pragma: allowlist secret
|
|
base_url="https://api.example.com",
|
|
model="gpt-4o-mini",
|
|
)
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
provider.generate(input_data)
|
|
|
|
mock_completion.assert_called_once()
|
|
call_kwargs: dict[str, Any] = mock_completion.call_args[1]
|
|
assert call_kwargs["api_key"] == "test-key" # pragma: allowlist secret
|
|
assert call_kwargs["base_url"] == "https://api.example.com"
|
|
assert call_kwargs["timeout"] == LiteLLMSynthesisProvider.TIMEOUT
|
|
|
|
|
|
def test_litellm_generate_returns_none_on_exception(
|
|
mocker: MockerFixture, target_date: date
|
|
) -> None:
|
|
"""Vérifie que LiteLLMSynthesisProvider.generate retourne None en cas d'exception."""
|
|
from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
|
|
|
|
mock_completion = mocker.patch("litellm.completion")
|
|
mock_completion.side_effect = Exception("error")
|
|
|
|
provider = LiteLLMSynthesisProvider(api_key="test-key", model="gpt-4o-mini")
|
|
input_data = SynthesisInput(target_date=target_date, agenda_diff=None)
|
|
result = provider.generate(input_data)
|
|
|
|
assert result is None
|
|
|
|
|
|
# --- get_synthesis_provider factory tests ---
|
|
|
|
|
|
def test_factory_returns_none_if_disabled() -> None:
|
|
"""Vérifie que la factory retourne None si la synthèse IA est désactivée."""
|
|
settings = AISettings(enabled=False, api_key=SecretStr("test-key"))
|
|
result = get_synthesis_provider(settings)
|
|
assert result is None
|
|
|
|
|
|
def test_factory_returns_none_if_no_api_key() -> None:
|
|
"""Vérifie que la factory retourne None si aucune clé API n'est configurée."""
|
|
settings = AISettings(enabled=True, api_key=None)
|
|
result = get_synthesis_provider(settings)
|
|
assert result is None
|
|
|
|
|
|
def test_factory_returns_openai_provider_by_default() -> None:
|
|
"""Vérifie que la factory retourne OpenAISynthesisProvider par défaut."""
|
|
settings = AISettings(
|
|
enabled=True,
|
|
api_key=SecretStr("test-key"),
|
|
provider="openai",
|
|
)
|
|
result = get_synthesis_provider(settings)
|
|
assert isinstance(result, OpenAISynthesisProvider)
|
|
|
|
|
|
def test_factory_returns_litellm_provider_when_requested() -> None:
|
|
"""Vérifie que la factory retourne LiteLLMSynthesisProvider si demandé."""
|
|
from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
|
|
|
|
settings = AISettings(
|
|
enabled=True,
|
|
api_key=SecretStr("test-key"),
|
|
provider="litellm",
|
|
)
|
|
result = get_synthesis_provider(settings)
|
|
assert isinstance(result, LiteLLMSynthesisProvider)
|
|
|
|
|
|
def test_factory_returns_none_with_warning_if_litellm_not_available(
|
|
mocker: MockerFixture, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
"""Vérifie que la factory retourne None avec un avertissement si litellm n'est pas disponible."""
|
|
# Forcer une ImportError lors de l'import
|
|
import builtins
|
|
|
|
original_import = builtins.__import__
|
|
|
|
def mock_import(name: str, *args: Any, **kwargs: Any) -> Any:
|
|
if name == "pronote_sync.synthesis.litellm":
|
|
raise ImportError("No module named 'litellm'")
|
|
return original_import(name, *args, **kwargs)
|
|
|
|
mocker.patch.object(builtins, "__import__", mock_import)
|
|
settings = AISettings(
|
|
enabled=True,
|
|
api_key=SecretStr("test-key"),
|
|
provider="litellm",
|
|
)
|
|
result = get_synthesis_provider(settings)
|
|
assert result is None
|
|
assert "Extra 'ai-litellm' requis pour le provider litellm" in caplog.text
|
|
|
|
|
|
# --- Provider protocol compliance ---
|
|
|
|
|
|
def test_openai_provider_is_synthesis_provider() -> None:
|
|
"""Vérifie que OpenAISynthesisProvider implémente SynthesisProvider."""
|
|
provider = OpenAISynthesisProvider(api_key="test-key")
|
|
assert isinstance(provider, SynthesisProvider)
|
|
|
|
|
|
def test_litellm_provider_is_synthesis_provider() -> None:
|
|
"""Vérifie que LiteLLMSynthesisProvider implémente SynthesisProvider."""
|
|
from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
|
|
|
|
provider = LiteLLMSynthesisProvider(api_key="test-key")
|
|
assert isinstance(provider, SynthesisProvider)
|