feat: implement XmppChannel and SyncXmppChannel (M10-U4+U5)

XmppChannel sends direct messages via slixmpp ClientXMPP with:
- _format_message: 5 emoji sections (synthèse, agenda, devoirs, messages, infos)
  with sanitize_plaintext on all content (SEC-XMPP-06)
- send_async: public async method with connect, STARTTLS verification,
  send_message, disconnect lifecycle (D1, SEC-XMPP-04)
- send: sync wrapper via asyncio.run() for direct callers

SyncXmppChannel adapts async XmppChannel for synchronous pipeline use (D4):
- asyncio.run() when no event loop running (nominal pipeline)
- daemon thread with timeout when event loop already running
- Returns False on any error, never raises (non-blocking)

Security:
- auto_reconnect=False, failed_auth → disconnect + PipelineWarning (SEC-XMPP-04)
- __cause__ and __context__ cleared on all PipelineWarning raises (SEC-XMPP-05)
- redact_secrets with extra_secrets=[jid, password, to] on all logs (SEC-XMPP-02)
- STARTTLS features check post-connection, disconnect on failure (D1)
- dry-run mode logs redacted message without connecting

28 unit tests (20 channel + 8 adapter) covering success, errors, security.

Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
Co-authored-by: opencode/coder <coder@agents.invalid>
This commit is contained in:
2026-09-07 23:28:18 +02:00
parent dcf7f69c5a
commit 1962e13eba
5 changed files with 1286 additions and 45 deletions

View File

@@ -0,0 +1,630 @@
"""Tests unitaires pour le canal XMPP (XmppChannel).
Ce module teste l'implémentation de :class:`pronote_sync.channels.xmpp.XmppChannel`
selon les spécifications du projet (GUIDE_DEV_PYTHON.md §10, décisions D1-D3,
audit de sécurité SEC-XMPP-02/04/05/06).
Les tests sont conçus pour être exécutés sans réseau, avec des mocks de slixmpp.
"""
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from datetime import date, datetime
from unittest.mock import patch
import pytest
from pydantic import SecretStr
from pronote_sync.channels.xmpp import XmppChannel, XmppMessage
from pronote_sync.config.settings import XmppSettings
from pronote_sync.errors import PipelineWarning
from pronote_sync.models.agenda import Lesson
from pronote_sync.models.blog import BlogArticle, ExternalInfo
from pronote_sync.models.diff import AgendaChange, AgendaChangeType
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
# Sentinelles pour tests de non-fuite de secrets
BOT_SENTINEL_JID = "BOT_SENTINEL_JID@example.com"
PASS_SENTINEL_123 = "PASS_SENTINEL_123"
RECIPIENT_SENTINEL = "RECIPIENT_SENTINEL@example.com"
@pytest.fixture
def xmpp_settings() -> XmppSettings:
"""Fixture fournissant des paramètres XMPP valides pour les tests.
:return: Instance de XmppSettings avec des valeurs par défaut valides.
:rtype: XmppSettings
"""
return XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"), # pragma: allowlist secret
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
use_tls=True,
timeout=30,
)
@pytest.fixture
def xmpp_message_minimal() -> XmppMessage:
"""Fixture fournissant un message XMPP minimal pour les tests.
:return: Instance de XmppMessage avec seulement la date cible.
:rtype: XmppMessage
"""
return XmppMessage(
target_date=date(2025, 9, 7),
synthesis=None,
external_info=None,
)
@pytest.fixture
def xmpp_message_full() -> XmppMessage:
"""Fixture fournissant un message XMPP complet pour les tests.
:return: Instance de XmppMessage avec tous les champs remplis.
:rtype: XmppMessage
"""
homework = Homework(
id="hw1",
subject="Mathématiques",
teachers=("M. Dupont",),
assigned_on=date(2025, 9, 1),
due_on=date(2025, 9, 15),
text="Faire l'exercice 5 page 42",
html="<p>Faire l'exercice 5 page 42</p>",
)
lesson = Lesson(
id="lesson1",
subject="Physique",
start=datetime.fromisoformat("2025-09-07T08:00:00"),
end=datetime.fromisoformat("2025-09-07T09:00:00"),
rooms=("B201",),
teachers=("M. Martin",),
group=None,
content=None,
)
change = AgendaChange(
type=AgendaChangeType.ADDED,
lesson=lesson,
theoretical_lesson=None,
details="Cours déplacé",
)
message = Message(
id="msg1",
type=MessageType.INFORMATION,
title="Réunion parents-professeurs",
content="Une réunion est organisée le 15/09 à 18h.",
author="CPE",
date=datetime.fromisoformat("2025-09-01T10:00:00"),
read=False,
)
article = BlogArticle(
id="art1",
title="Sortie scolaire",
url="https://blog.example.com/sortie",
published_at=datetime.fromisoformat("2025-09-01T09:00:00"),
updated_at=None,
category="Actualités",
author="Collège",
content_html="<p>Sortie prévue le 20/09.</p>",
content_text="Sortie prévue le 20/09.",
)
external = ExternalInfo(
blog_articles=(article,),
pronote_messages=(message,),
other_info=("Info supplémentaire",),
)
return XmppMessage(
target_date=date(2025, 9, 7),
synthesis="Voici la synthèse des activités du jour.",
homeworks=(homework,),
changes=(change,),
messages=(),
external_info=external,
)
class MockClientXMPP:
"""Mock de slixmpp.ClientXMPP pour tests unitaires.
Ce mock simule le comportement minimal de ClientXMPP nécessaire pour
tester XmppChannel sans dépendre du réseau ou d'un serveur XMPP réel.
"""
def __init__(
self,
jid: str,
password: str,
*args: object,
**kwargs: object,
) -> None:
"""Initialise le mock ClientXMPP.
:param jid: JID du compte bot.
:param password: Mot de passe du compte bot.
:param args: Arguments positionnels supplémentaires (ignoré).
:param kwargs: Arguments nommés supplémentaires (ignoré).
"""
self.jid = jid
self.password = password
self.connected = False
self.disconnected = False
self.features: set[str] = set()
self.handlers: dict[str, list[Callable[..., Awaitable[None]]]] = {
"session_start": [],
"failed_auth": [],
"disconnected": [],
}
async def connect(
self,
*args: object,
**kwargs: object,
) -> None:
"""Simule la connexion au serveur XMPP.
Déclenche les handlers appropriés selon le scénario de test.
:param args: Arguments supplémentaires (ignoré).
:param kwargs: Arguments nommés supplémentaires (ignoré).
"""
self.connected = True
# Simuler STARTTLS si demandé
if kwargs.get("use_tls", False):
self.features.add("starttls")
# Déclencher session_start par défaut
for handler in self.handlers["session_start"]:
await handler()
async def disconnect(self) -> None:
"""Simule la déconnexion du serveur XMPP.
:raises: AssertionError si déjà déconnecté.
"""
if self.disconnected:
msg = "Client already disconnected"
raise AssertionError(msg)
self.disconnected = True
def send_message(self, *args: object, **kwargs: object) -> None:
"""Simule l'envoi d'un message.
:param args: Arguments supplémentaires (ignoré).
:param kwargs: Arguments nommés supplémentaires (ignoré).
"""
# Pas d'implémentation réelle nécessaire pour les tests
def add_event_handler(self, event: str, handler: Callable[..., Awaitable[None]]) -> None:
"""Enregistre un gestionnaire d'événement.
:param event: Nom de l'événement (ex: 'session_start').
:param handler: Fonction gestionnaire.
:raises: AssertionError si l'événement n'est pas supporté.
"""
if event not in self.handlers:
msg = f"Unsupported event: {event}"
raise AssertionError(msg)
self.handlers[event].append(handler)
@pytest.mark.usefixtures("xmpp_settings")
class TestXmppChannelFormatMessage:
"""Tests unitaires pour la méthode _format_message de XmppChannel.
Ces tests vérifient le formatage des messages XMPP en texte brut,
sans dépendre de slixmpp ni du réseau.
"""
def test_format_message_with_synthesis(self, xmpp_message_full: XmppMessage) -> None:
"""Test que la synthèse est bien formatée avec l'emoji dédié.
:param xmpp_message_full: Message XMPP complet.
"""
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(xmpp_message_full)
assert "📌 Synthèse" in formatted
assert "Voici la synthèse des activités du jour." in formatted
def test_format_message_without_synthesis(self, xmpp_message_minimal: XmppMessage) -> None:
"""Test que l'absence de synthèse affiche le texte par défaut.
:param xmpp_message_minimal: Message XMPP minimal.
"""
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(xmpp_message_minimal)
assert "📌 Synthèse" in formatted
assert "Aucune synthèse disponible." in formatted
def test_format_message_with_homeworks(self, xmpp_message_full: XmppMessage) -> None:
"""Test que les devoirs sont bien formatés avec l'emoji dédié.
:param xmpp_message_full: Message XMPP complet.
"""
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(xmpp_message_full)
assert "📚 Devoirs" in formatted
assert "Mathématiques" in formatted
assert "Faire l'exercice 5 page 42" in formatted
def test_format_message_without_homeworks(self, xmpp_message_minimal: XmppMessage) -> None:
"""Test que l'absence de devoirs affiche le texte par défaut.
:param xmpp_message_minimal: Message XMPP minimal.
"""
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(xmpp_message_minimal)
assert "📚 Devoirs" in formatted
assert "Aucun devoir." in formatted
def test_format_message_with_changes(self, xmpp_message_full: XmppMessage) -> None:
"""Test que les changements d'agenda sont bien formatés.
:param xmpp_message_full: Message XMPP complet.
"""
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(xmpp_message_full)
assert "📅 Changements d'agenda" in formatted
assert "Cours déplacé" in formatted
def test_format_message_with_messages(self, xmpp_message_full: XmppMessage) -> None:
"""Test que les messages sont bien formatés.
:param xmpp_message_full: Message XMPP complet.
"""
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(xmpp_message_full)
assert "💬 Messages" in formatted
def test_format_message_with_external_info(self, xmpp_message_full: XmppMessage) -> None:
"""Test que les informations externes sont bien formatées.
:param xmpp_message_full: Message XMPP complet.
"""
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(xmpp_message_full)
assert "📢 Informations diverses" in formatted
assert "Sortie prévue le 20/09." in formatted
def test_format_message_empty(self) -> None:
"""Test que tous les sections vides affichent le texte par défaut."""
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(msg)
assert "📌 Synthèse\nAucune synthèse disponible." in formatted
assert "📚 Devoirs\nAucun devoir." in formatted
assert "📅 Changements d'agenda\nAucun changement." in formatted
assert "💬 Messages\nAucun message." in formatted
assert "📢 Informations diverses\nAucune information." in formatted
def test_format_message_html_sanitized(self) -> None:
"""Test que le HTML est supprimé du contenu des devoirs et messages."""
homework = Homework(
id="hw_html",
subject="SVT",
teachers=("M. Bernard",),
assigned_on=date(2025, 9, 1),
due_on=date(2025, 9, 20),
text="Lire <b>le chapitre 3</b> et répondre aux questions.",
html="<p>Lire <b>le chapitre 3</b> et répondre aux questions.</p>",
)
msg = XmppMessage(
target_date=date(2025, 9, 7),
synthesis=None,
homeworks=(homework,),
external_info=None,
)
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(msg)
# Le HTML doit être supprimé
assert "<b>" not in formatted
assert "le chapitre 3" in formatted
def test_format_message_control_chars_stripped(self) -> None:
"""Test que les caractères de contrôle sont supprimés du contenu."""
homework = Homework(
id="hw_ctrl",
subject="Histoire",
teachers=("Mme Dubois",),
assigned_on=date(2025, 9, 1),
due_on=date(2025, 9, 25),
text="Fiche\x00n°4\x01à\x07rendre\x1f",
html="",
)
msg = XmppMessage(
target_date=date(2025, 9, 7),
synthesis=None,
homeworks=(homework,),
external_info=None,
)
channel = XmppChannel(XmppSettings(), dry_run=True)
formatted = channel._format_message(msg)
# Les caractères de contrôle doivent être supprimés
assert "\x00" not in formatted
assert "\x01" not in formatted
assert "\x07" not in formatted
assert "\x1f" not in formatted
# Le texte doit rester lisible
assert "Fiche" in formatted
assert "n°4" in formatted
assert "à" in formatted
assert "rendre" in formatted
@pytest.mark.usefixtures("xmpp_settings")
class TestXmppChannelSend:
"""Tests unitaires pour la méthode send de XmppChannel.
Ces tests vérifient le comportement de l'envoi de messages XMPP,
avec mock de slixmpp.ClientXMPP.
"""
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_send_dry_run_returns_true(self, xmpp_settings: XmppSettings) -> None:
"""Test que dry_run=True retourne True sans appeler slixmpp.
:param xmpp_settings: Paramètres XMPP valides.
"""
channel = XmppChannel(xmpp_settings, dry_run=True)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = channel.send(msg)
assert result is True
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_send_success_returns_true(self, xmpp_settings: XmppSettings) -> None:
"""Test que send retourne True en cas de succès de connexion.
:param xmpp_settings: Paramètres XMPP valides.
"""
channel = XmppChannel(xmpp_settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = channel.send(msg)
assert result is True
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_send_failed_auth_raises_pipeline_warning(self, xmpp_settings: XmppSettings) -> None:
"""Vérifie qu'un échec d'authentification lève un PipelineWarning.
:param xmpp_settings: Paramètres XMPP valides.
"""
# Créer un mock qui déclenche failed_auth
class FailedAuthClient(MockClientXMPP):
async def connect(self, *args: object, **kwargs: object) -> None:
self.connected = True
# Déclencher failed_auth au lieu de session_start
for handler in self.handlers["failed_auth"]:
await handler()
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=FailedAuthClient):
channel = XmppChannel(xmpp_settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
with pytest.raises(PipelineWarning):
channel.send(msg)
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_send_connection_timeout_returns_false(self, xmpp_settings: XmppSettings) -> None:
"""Test que send retourne False en cas de timeout de connexion.
:param xmpp_settings: Paramètres XMPP valides.
"""
# Créer un mock qui bloque indéfiniment : méthode non-asynchrone pour
# éviter de créer une coroutine jamais attendue (RuntimeWarning),
# ``asyncio.wait_for`` étant patché pour lever immédiatement.
class TimeoutClient(MockClientXMPP):
def connect(self, *args: object, **kwargs: object) -> None: # type: ignore[override]
# Ne jamais terminer la connexion -> timeout
return None
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=TimeoutClient):
channel = XmppChannel(xmpp_settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
# Utiliser un timeout court pour le test
with patch("asyncio.wait_for", side_effect=asyncio.TimeoutError):
result = channel.send(msg)
assert result is False
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_send_starttls_not_available(self, xmpp_settings: XmppSettings) -> None:
"""Test que send retourne False si STARTTLS non disponible avec use_tls=True.
:param xmpp_settings: Paramètres XMPP valides.
"""
class NoTLSClient(MockClientXMPP):
async def connect(self, *args: object, **kwargs: object) -> None:
self.connected = True
# Ne pas ajouter 'starttls' aux features
for handler in self.handlers["session_start"]:
await handler()
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=NoTLSClient):
# Force use_tls=True
settings = xmpp_settings.model_copy(update={"use_tls": True})
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = channel.send(msg)
assert result is False
@pytest.mark.usefixtures("xmpp_settings")
class TestXmppChannelSecurity:
"""Tests de sécurité pour XmppChannel (non-fuite de secrets).
Ces tests vérifient que les secrets (JID, mot de passe, destinataire)
ne sont jamais exposés dans les logs, messages d'erreur ou causes d'exceptions.
"""
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_no_jid_in_logs_on_error(self, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie que le JID n'apparaît pas dans les logs en cas d'erreur.
:param caplog: Fixture pytest pour capturer les logs.
"""
settings = XmppSettings(
enabled=True,
jid=BOT_SENTINEL_JID,
password=SecretStr("ignored"), # pragma: allowlist secret
host="xmpp.example.com",
port=5222,
to="ignored",
resource="pronote-sync",
use_tls=True,
timeout=30,
)
class ErrorClient(MockClientXMPP):
async def connect(self, *args: object, **kwargs: object) -> None:
raise RuntimeError("Connexion impossible")
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ErrorClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
try:
channel.send(msg)
except Exception:
pass
# Vérifier que le JID sentinelle n'apparaît pas dans les logs
logs = caplog.text
assert BOT_SENTINEL_JID not in logs
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_no_password_in_logs_on_error(self, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie que le mot de passe n'apparaît pas dans les logs en cas d'erreur.
:param caplog: Fixture pytest pour capturer les logs.
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr(PASS_SENTINEL_123),
host="xmpp.example.com",
port=5222,
to="ignored",
resource="pronote-sync",
use_tls=True,
timeout=30,
)
class ErrorClient(MockClientXMPP):
async def connect(self, *args: object, **kwargs: object) -> None:
raise RuntimeError("Authentification échouée")
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ErrorClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
try:
channel.send(msg)
except Exception:
pass
# Vérifier que le mot de passe sentinelle n'apparaît pas dans les logs
logs = caplog.text
assert PASS_SENTINEL_123 not in logs
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_no_recipient_in_logs_on_error(self, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie que le destinataire n'apparaît pas dans les logs en cas d'erreur.
:param caplog: Fixture pytest pour capturer les logs.
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("ignored"),
host="xmpp.example.com",
port=5222,
to=RECIPIENT_SENTINEL,
resource="pronote-sync",
use_tls=True,
timeout=30,
)
class ErrorClient(MockClientXMPP):
async def connect(self, *args: object, **kwargs: object) -> None:
raise RuntimeError("Envoi impossible")
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ErrorClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
try:
channel.send(msg)
except Exception:
pass
# Vérifier que le destinataire sentinelle n'apparaît pas dans les logs
logs = caplog.text
assert RECIPIENT_SENTINEL not in logs
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_no_secret_in_exception_cause(self) -> None:
"""Vérifie que les secrets ne sont pas présents dans __cause__ d'une PipelineWarning."""
settings = XmppSettings(
enabled=True,
jid=BOT_SENTINEL_JID,
password=SecretStr(PASS_SENTINEL_123),
host="xmpp.example.com",
port=5222,
to=RECIPIENT_SENTINEL,
resource="pronote-sync",
use_tls=True,
timeout=30,
)
class AuthErrorClient(MockClientXMPP):
async def connect(self, *args: object, **kwargs: object) -> None:
self.connected = True
for handler in self.handlers["failed_auth"]:
await handler()
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=AuthErrorClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
with pytest.raises(PipelineWarning) as exc_info:
channel.send(msg)
# Vérifier que l'exception levée est bien une PipelineWarning
assert isinstance(exc_info.value, PipelineWarning)
# Vérifier que les secrets ne sont pas dans __cause__
cause_str = str(exc_info.value.__cause__) if exc_info.value.__cause__ else ""
assert BOT_SENTINEL_JID not in cause_str
assert PASS_SENTINEL_123 not in cause_str
assert RECIPIENT_SENTINEL not in cause_str
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=MockClientXMPP)
def test_dry_run_no_secret_in_log(self, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie que dry_run n'expose pas de secrets dans les logs.
:param caplog: Fixture pytest pour capturer les logs.
"""
settings = XmppSettings(
enabled=True,
jid=BOT_SENTINEL_JID,
password=SecretStr(PASS_SENTINEL_123),
host="xmpp.example.com",
port=5222,
to=RECIPIENT_SENTINEL,
resource="pronote-sync",
use_tls=True,
timeout=30,
)
channel = XmppChannel(settings, dry_run=True)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
channel.send(msg)
# Vérifier que les sentinelles n'apparaissent pas dans les logs
logs = caplog.text
assert BOT_SENTINEL_JID not in logs
assert PASS_SENTINEL_123 not in logs
assert RECIPIENT_SENTINEL not in logs

View File

@@ -0,0 +1,246 @@
"""Tests unitaires pour l'adaptateur SyncXmppChannel.
Ce module teste l'implémentation de :class:`pronote_sync.channels.xmpp.SyncXmppChannel`
qui est un adaptateur wrapant XmppChannel pour fournir une interface synchrone.
Les tests sont conçus pour être exécutés sans réseau, avec des mocks de XmppChannel
ou de slixmpp, et vérifient le comportement de l'envoi synchrone selon la décision D4.
Conformément à D4, SyncXmppChannel.send() utilise asyncio.run() directement sans
créer de nouvelle event loop inutilement. Le comportement est :
- Pas de boucle en cours → asyncio.run(channel.send_async(message))
- Boucle en cours détectée → thread daemon avec asyncio.run() + join(timeout)
- Retourne True en cas de succès, False en cas d'erreur (attrape toute exception)
- Aucun secret dans les logs.
"""
from __future__ import annotations
from datetime import date
from unittest.mock import AsyncMock, patch
import pytest
from pronote_sync.channels.protocol import Channel
from pronote_sync.channels.xmpp import SyncXmppChannel, XmppMessage
from pronote_sync.config.settings import XmppSettings
# Sentinelles pour tests de non-fuite de secrets
BOT_SENTINEL_JID = "BOT_SENTINEL_JID@example.com"
PASS_SENTINEL_123 = "PASS_SENTINEL_123"
RECIPIENT_SENTINEL = "RECIPIENT_SENTINEL@example.com"
@pytest.fixture
def xmpp_settings() -> XmppSettings:
"""Fixture fournissant des paramètres XMPP valides pour les tests.
:return: Instance de XmppSettings avec des valeurs par défaut valides.
:rtype: XmppSettings
"""
from pydantic import SecretStr
return XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
to="parent@example.com",
use_tls=True,
)
@pytest.fixture
def xmpp_message() -> XmppMessage:
"""Fixture fournissant un message XMPP minimal pour les tests.
:return: Instance de XmppMessage avec seulement la date cible.
:rtype: XmppMessage
"""
return XmppMessage(
target_date=date(2025, 9, 7),
synthesis=None,
homeworks=(),
changes=(),
messages=(),
external_info=None,
)
class TestSyncXmppChannelSend:
"""Tests unitaires pour la méthode send de SyncXmppChannel.
Ces tests vérifient le comportement de l'envoi synchrone de messages XMPP
selon la décision D4 : utilisation directe de asyncio.run() et retour de
booléen (True/False) sans lever d'exception.
"""
def test_sync_adapter_send_returns_true_on_success(
self, xmpp_settings: XmppSettings, xmpp_message: XmppMessage
) -> None:
"""Test que send retourne True en cas de succès.
:param xmpp_settings: Paramètres XMPP valides.
:param xmpp_message: Message XMPP minimal.
"""
# Mock XmppChannel.send_async pour simuler un succès
with patch("pronote_sync.channels.xmpp.XmppChannel") as MockChannel:
mock_instance = MockChannel.return_value
mock_instance.send_async = AsyncMock(return_value=True)
mock_instance.settings = xmpp_settings
adapter = SyncXmppChannel(xmpp_settings)
result = adapter.send(xmpp_message)
assert result is True
def test_sync_adapter_send_returns_false_on_error(
self, xmpp_settings: XmppSettings, xmpp_message: XmppMessage
) -> None:
"""Test que send retourne False en cas d'erreur.
Vérifie que la méthode ne lève pas d'exception non gérée et retourne False.
:param xmpp_settings: Paramètres XMPP valides.
:param xmpp_message: Message XMPP minimal.
"""
# Mock XmppChannel.send_async pour simuler une erreur
with patch("pronote_sync.channels.xmpp.XmppChannel") as MockChannel:
mock_instance = MockChannel.return_value
mock_instance.send_async = AsyncMock(side_effect=RuntimeError("Connexion impossible"))
mock_instance.settings = xmpp_settings
adapter = SyncXmppChannel(xmpp_settings)
result = adapter.send(xmpp_message)
assert result is False
def test_sync_adapter_satisfies_channel_protocol(self, xmpp_settings: XmppSettings) -> None:
"""Test que SyncXmppChannel satisfait le protocole Channel.
Vérifie que l'instance est reconnue comme implémentant le protocole.
:param xmpp_settings: Paramètres XMPP valides.
"""
channel = SyncXmppChannel(xmpp_settings)
assert isinstance(channel, Channel)
class TestSyncXmppChannelAsyncioRun:
"""Tests pour la gestion de l'event loop dans SyncXmppChannel selon D4.
Ces tests vérifient que SyncXmppChannel utilise asyncio.run() directement
sans créer de nouvelle event loop inutilement.
"""
def test_sync_adapter_uses_asyncio_run_when_no_loop(
self, xmpp_settings: XmppSettings, xmpp_message: XmppMessage
) -> None:
"""Test que send utilise asyncio.run() quand aucune boucle n'est en cours.
Vérifie que asyncio.run est appelé avec le coroutine send_async.
:param xmpp_settings: Paramètres XMPP valides.
:param xmpp_message: Message XMPP minimal.
"""
# Mock XmppChannel.send_async pour simuler un succès
with patch("pronote_sync.channels.xmpp.XmppChannel") as MockChannel:
mock_instance = MockChannel.return_value
mock_instance.send_async = AsyncMock(return_value=True)
mock_instance.settings = xmpp_settings
adapter = SyncXmppChannel(xmpp_settings)
# Patch asyncio.run pour vérifier son appel
with patch("pronote_sync.channels.xmpp.asyncio.run") as mock_asyncio_run:
mock_asyncio_run.return_value = True
result = adapter.send(xmpp_message)
# Vérifier que asyncio.run a été appelé
mock_asyncio_run.assert_called_once()
# Vérifier que le résultat est True
assert result is True
def test_sync_adapter_send_returns_false_when_asyncio_run_fails(
self, xmpp_settings: XmppSettings, xmpp_message: XmppMessage
) -> None:
"""Test que send retourne False quand asyncio.run échoue.
Vérifie que toute exception dans asyncio.run est capturée et retourne False.
:param xmpp_settings: Paramètres XMPP valides.
:param xmpp_message: Message XMPP minimal.
"""
# Mock XmppChannel.send_async pour simuler un succès
with patch("pronote_sync.channels.xmpp.XmppChannel") as MockChannel:
mock_instance = MockChannel.return_value
mock_instance.send_async = AsyncMock(return_value=True)
mock_instance.settings = xmpp_settings
adapter = SyncXmppChannel(xmpp_settings)
# Patch asyncio.run pour lever une exception
with patch("pronote_sync.channels.xmpp.asyncio.run") as mock_asyncio_run:
mock_asyncio_run.side_effect = RuntimeError("Erreur asyncio")
result = adapter.send(xmpp_message)
# Vérifier que le résultat est False
assert result is False
class TestSyncXmppChannelSecurity:
"""Tests de sécurité pour SyncXmppChannel (non-fuite de secrets).
Ces tests vérifient que les secrets (JID, mot de passe, destinataire)
ne sont jamais exposés dans les logs, messages d'erreur ou causes d'exceptions.
"""
def test_sync_adapter_no_secret_in_logs(
self, caplog: pytest.LogCaptureFixture, xmpp_settings: XmppSettings
) -> None:
"""Test que les sentinelles n'apparaissent pas dans les logs en cas d'erreur.
:param caplog: Fixture pytest pour capturer les logs.
:param xmpp_settings: Paramètres XMPP valides.
"""
# Créer des settings avec sentinelles
from pydantic import SecretStr
settings = XmppSettings(
enabled=True,
jid=BOT_SENTINEL_JID,
password=SecretStr(PASS_SENTINEL_123),
host="xmpp.example.com",
to=RECIPIENT_SENTINEL,
use_tls=True,
)
# Mock XmppChannel.send_async pour lever une erreur
with patch("pronote_sync.channels.xmpp.XmppChannel") as MockChannel:
mock_instance = MockChannel.return_value
mock_instance.send_async = AsyncMock(side_effect=RuntimeError("Connexion impossible"))
mock_instance.settings = settings
channel = SyncXmppChannel(settings)
msg = XmppMessage(
target_date=date(2025, 9, 7),
synthesis=None,
homeworks=(),
changes=(),
messages=(),
external_info=None,
)
# Appeler send qui doit retourner False sans lever d'exception
result = channel.send(msg)
# Vérifier que le résultat est False
assert result is False
# Vérifier que les sentinelles n'apparaissent pas dans les logs
logs = caplog.text
assert BOT_SENTINEL_JID not in logs
assert PASS_SENTINEL_123 not in logs
assert RECIPIENT_SENTINEL not in logs