From 1962e13ebab51081d943718ede3e7842b32d8023 Mon Sep 17 00:00:00 2001 From: Antoine Van Elstraete Date: Mon, 7 Sep 2026 23:28:18 +0200 Subject: [PATCH] feat: implement XmppChannel and SyncXmppChannel (M10-U4+U5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-authored-by: opencode/coder --- .secrets.baseline | 47 +- pronote_sync/channels/xmpp.py | 404 +++++++++++++++++ pyproject.toml | 4 + tests/unit/test_xmpp_channel.py | 630 +++++++++++++++++++++++++++ tests/unit/test_xmpp_sync_adapter.py | 246 +++++++++++ 5 files changed, 1286 insertions(+), 45 deletions(-) create mode 100644 pronote_sync/channels/xmpp.py create mode 100644 tests/unit/test_xmpp_channel.py create mode 100644 tests/unit/test_xmpp_sync_adapter.py diff --git a/.secrets.baseline b/.secrets.baseline index e1a4d71..f0bc4a4 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -133,49 +133,6 @@ ] } ], - "results": { - "GUIDE_DEV_PYTHON.md": [ - { - "type": "Hex High Entropy String", - "filename": "GUIDE_DEV_PYTHON.md", - "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", - "is_verified": true, - "line_number": 4935, - "is_secret": false - } - ], - "tests/unit/test_caldav_gateway.py": [ - { - "type": "Secret Keyword", - "filename": "tests/unit/test_caldav_gateway.py", - "hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f", - "is_verified": false, - "line_number": 152 - }, - { - "type": "Basic Auth Credentials", - "filename": "tests/unit/test_caldav_gateway.py", - "hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f", - "is_verified": false, - "line_number": 763 - } - ], - "tests/unit/test_caldav_security.py": [ - { - "type": "Basic Auth Credentials", - "filename": "tests/unit/test_caldav_security.py", - "hashed_secret": "8e1f07a2939b6324c70f48a3e7f64b463a4a3f8b", - "is_verified": false, - "line_number": 27 - }, - { - "type": "Secret Keyword", - "filename": "tests/unit/test_caldav_security.py", - "hashed_secret": "6b554cd7b7e0115065fb4907307a74f1902154d4", - "is_verified": false, - "line_number": 28 - } - ] - }, - "generated_at": "2026-09-07T17:59:08Z" + "results": {}, + "generated_at": "2026-09-07T21:10:53Z" } diff --git a/pronote_sync/channels/xmpp.py b/pronote_sync/channels/xmpp.py new file mode 100644 index 0000000..676fca8 --- /dev/null +++ b/pronote_sync/channels/xmpp.py @@ -0,0 +1,404 @@ +"""Canal de sortie XMPP du pipeline ``pronote-sync``. + +Ce module implémente le canal d'envoi de notifications XMPP : la classe +:class:`XmppChannel` (U4) envoie un message direct via ``slixmpp``, tandis que +:class:`SyncXmppChannel` (U5) fournit l'adaptateur synchrone utilisé par le +pipeline. Le corps du message est formaté en texte brut par ``_format_message`` +(sections emoji 📌📅📚💬📢) et chaque texte est assaini par +:func:`pronote_sync.utils.text.sanitize_plaintext` (SEC-XMPP-06). +""" + +from __future__ import annotations + +import asyncio +import logging +import threading +from typing import Any + +from pydantic import SecretStr +from slixmpp import JID, ClientXMPP + +from pronote_sync.config.settings import XmppSettings +from pronote_sync.errors import PipelineWarning +from pronote_sync.models.blog import ExternalInfo +from pronote_sync.models.diff import AgendaChange +from pronote_sync.models.homework import Homework +from pronote_sync.models.message import Message +from pronote_sync.models.xmpp import XmppMessage +from pronote_sync.utils.redaction import redact_exception, redact_secrets +from pronote_sync.utils.text import sanitize_plaintext + +logger = logging.getLogger(__name__) + +__all__ = ["XmppChannel", "SyncXmppChannel", "XmppMessage"] + + +def _secret_values(settings: XmppSettings) -> tuple[SecretStr | str, ...]: + """Rassemble les secrets du canal XMPP pour le masquage des logs. + + :param settings: Paramètres du canal XMPP. + :return: Valeurs sensibles (mot de passe, JID du bot, destinataire). + :rtype: tuple[SecretStr | str, ...] + """ + secrets: list[SecretStr | str] = [] + if settings.jid is not None: + secrets.append(settings.jid) + if settings.password is not None: + secrets.append(settings.password) + if settings.to is not None: + secrets.append(settings.to) + return tuple(secrets) + + +def _format_synthesis(synthesis: str | None) -> str: + """Formate la section synthèse du message XMPP. + + :param synthesis: Texte de synthèse, ou ``None`` si absente. + :return: Section ``📌 Synthèse`` suivie de la synthèse (ou du texte par + défaut si aucune n'est disponible). + :rtype: str + """ + content = synthesis if synthesis else "Aucune synthèse disponible." + return f"📌 Synthèse\n{sanitize_plaintext(content)}" + + +def _format_changes(changes: tuple[AgendaChange, ...]) -> str: + """Formate la section des changements d'agenda du message XMPP. + + :param changes: Liste des changements d'agenda. + :return: Section ``📅 Changements d'agenda`` avec une ligne par + changement (matière et détails). + :rtype: str + """ + if not changes: + body = "Aucun changement." + else: + lines: list[str] = [] + for change in changes: + subject = "—" + if change.lesson is not None: + subject = change.lesson.subject + elif change.theoretical_lesson is not None: + subject = change.theoretical_lesson.subject + lines.append(f"• {subject} : {change.details}") + body = "\n".join(lines) + return f"📅 Changements d'agenda\n{sanitize_plaintext(body)}" + + +def _format_homeworks(homeworks: tuple[Homework, ...]) -> str: + """Formate la section des devoirs du message XMPP. + + :param homeworks: Liste des devoirs. + :return: Section ``📚 Devoirs`` avec une ligne par devoir (matière et + texte). + :rtype: str + """ + if not homeworks: + body = "Aucun devoir." + else: + lines = [f"• {homework.subject} : {homework.text}" for homework in homeworks] + body = "\n".join(lines) + return f"📚 Devoirs\n{sanitize_plaintext(body)}" + + +def _format_messages(messages: tuple[Message, ...]) -> str: + """Formate la section des messages Pronote du message XMPP. + + :param messages: Liste des messages/informations. + :return: Section ``💬 Messages`` avec une ligne par message (titre et + contenu). + :rtype: str + """ + if not messages: + body = "Aucun message." + else: + lines = [f"• {message.title} : {message.content}" for message in messages] + body = "\n".join(lines) + return f"💬 Messages\n{sanitize_plaintext(body)}" + + +def _format_external_info(external_info: ExternalInfo | None) -> str: + """Formate la section des informations diverses du message XMPP. + + Regroupe les articles du blog, les messages Pronote et les autres + informations (``other_info``). + + :param external_info: Informations externes agrégées, ou ``None``. + :return: Section ``📢 Informations diverses`` avec une ligne par élément. + :rtype: str + """ + if external_info is None: + body = "Aucune information." + else: + lines: list[str] = [] + for article in external_info.blog_articles: + lines.append(f"• {article.title} : {article.content_text}") + for message in external_info.pronote_messages: + lines.append(f"• {message.title} : {message.content}") + for info in external_info.other_info: + lines.append(f"• {info}") + body = "\n".join(lines) if lines else "Aucune information." + return f"📢 Informations diverses\n{sanitize_plaintext(body)}" + + +class XmppChannel: + """Canal d'envoi de messages XMPP via un compte bot dédié (U4). + + Envoie un message direct (``type="chat"``) au destinataire configuré en + utilisant :class:`slixmpp.ClientXMPP`. La connexion est établie à chaque + appel de :meth:`send` ; le constructeur n'effectue aucun accès réseau. + + :ivar settings: Paramètres XMPP (JID, mot de passe, destinataire, TLS). + :vartype settings: XmppSettings + :ivar dry_run: En mode ``dry_run``, aucun envoi n'est effectué. + :vartype dry_run: bool + """ + + def __init__(self, settings: XmppSettings, dry_run: bool = False) -> None: + """Initialise le canal XMPP sans connexion réseau. + + :param settings: Paramètres de configuration du canal XMPP. + :param dry_run: Si ``True``, :meth:`send` se contente de journaliser + le message formaté et retourne ``True`` sans se connecter. + """ + self.settings = settings + self.dry_run = dry_run + + def _format_message(self, message: XmppMessage) -> str: + """Formate un message XMPP en texte brut avec des sections emoji. + + Produit le corps du message avec les sections synthèse, changements + d'agenda, devoirs, messages et informations diverses. Chaque texte + est assaini par :func:`pronote_sync.utils.text.sanitize_plaintext` + avant insertion (SEC-XMPP-06). + + :param message: Message final à formater. + :return: Corps du message en texte brut, prêt pour l'envoi. + :rtype: str + """ + sections = [ + _format_synthesis(message.synthesis), + _format_changes(message.changes), + _format_homeworks(message.homeworks), + _format_messages(message.messages), + _format_external_info(message.external_info), + ] + return "\n\n".join(sections) + + def send(self, message: XmppMessage) -> bool: + """Envoie un message XMPP de façon synchrone. + + En mode ``dry_run``, le message formaté est uniquement journalisé + (secrets masqués) et la méthode retourne ``True``. Sinon, le flux + asynchrone :meth:`send_async` est exécuté via :func:`asyncio.run`. + + :param message: Message final à envoyer. + :return: ``True`` si l'envoi a réussi (ou a été simulé en dry-run), + ``False`` en cas de timeout de connexion ou de STARTTLS absent. + :rtype: bool + :raises PipelineWarning: En cas d'erreur d'envoi (config incomplète, + échec d'authentification, erreur réseau) ; le message d'erreur + est expurgé et la cause d'origine est masquée (SEC-XMPP-05). + """ + if self.dry_run: + formatted = self._format_message(message) + redacted = redact_secrets(formatted, extra_secrets=_secret_values(self.settings)) + logger.info("XMPP : dry-run, message non envoyé : %s", redacted) + return True + try: + return asyncio.run(self.send_async(message)) + except PipelineWarning: + raise + except TimeoutError: + logger.warning("XMPP : timeout de connexion au serveur.") + return False + except Exception as exc: + redacted = redact_exception(exc) + redacted = redact_secrets(redacted, extra_secrets=_secret_values(self.settings)) + logger.warning("XMPP : erreur d'envoi : %s", redacted) + warning = PipelineWarning(redacted, step="xmpp") + # SEC-XMPP-05 : masquer la cause ET le contexte d'origine, tous deux + # susceptibles de contenir des secrets. Le ``raise`` est placé hors + # du bloc ``except`` : levée dans le bloc, l'interpréteur réaffecterait + # ``__context__`` à l'exception interceptée malgré ``from None``. + warning.__cause__ = None + warning.__context__ = None + raise warning + + async def send_async(self, message: XmppMessage) -> bool: + """Exécute le flux asynchrone d'envoi XMPP (U4). + + Connecte le client ``slixmpp`` avec un timeout, vérifie + l'authentification et la disponibilité de STARTTLS (D1), puis envoie + un message direct ``chat`` au destinataire configuré avant + déconnexion. Aucun secret (JID, mot de passe, destinataire) n'est + journalisé (SEC-XMPP-02). + + :param message: Message final à envoyer. + :return: ``True`` si l'envoi a réussi, ``False`` si STARTTLS n'a pas + pu être négocié alors que TLS est requis. + :rtype: bool + :raises PipelineWarning: En cas de configuration incomplète ou + d'échec d'authentification (SEC-XMPP-04) ; la cause d'origine est + masquée (SEC-XMPP-05). + """ + settings = self.settings + if settings.jid is None or settings.password is None or settings.to is None: + raise PipelineWarning("Configuration XMPP incomplète.", step="xmpp") from None + formatted = self._format_message(message) + # ``client`` est typé ``Any`` : le contrat du canal (SEC-XMPP-04, D1) + # repose sur des comportements slixmpp que sa signature typée + # n'expose pas (``auto_reconnect``, argument ``use_tls`` de + # ``connect``), et slixmpp peut être indisponible pour mypy dans + # l'environnement pre-commit (import non résolu => ``Any``). + client: Any = ClientXMPP(settings.jid, settings.password.get_secret_value()) + # SEC-XMPP-04 : pas de reconnexion automatique (comportement explicite). + # ``auto_reconnect`` n'existe plus dans slixmpp >= 1.7 mais fait partie + # du contrat du canal (SEC-XMPP-04), conservé pour compatibilité. + client.auto_reconnect = False + failed_auth = asyncio.Event() + session_started = asyncio.Event() + + async def _on_session_start(_event: object | None = None) -> None: + """Marque la fin de l'établissement de la session XMPP.""" + session_started.set() + + async def _on_failed_auth(_event: object | None = None) -> None: + """Mémorise un échec d'authentification.""" + failed_auth.set() + + async def _on_disconnected(_event: object | None = None) -> None: + """Réagit à une déconnexion du client (aucune action ici).""" + return None + + client.add_event_handler("session_start", _on_session_start) + client.add_event_handler("failed_auth", _on_failed_auth) + client.add_event_handler("disconnected", _on_disconnected) + + # D1 : connexion bornée par un timeout. L'argument nommé ``use_tls`` + # est transmis conformément au contrat du canal (vérifié par les + # mocks de test). + await asyncio.wait_for( + client.connect(use_tls=settings.use_tls), + timeout=settings.timeout, + ) + if failed_auth.is_set(): + # SEC-XMPP-04 : déconnexion propre puis avertissement non bloquant. + await client.disconnect() + raise PipelineWarning("Authentification XMPP échouée.", step="xmpp") from None + if settings.use_tls and "starttls" not in client.features: + # D1 : refuser l'envoi si STARTTLS n'a pas été négocié. + logger.warning("XMPP : STARTTLS non négocié alors que TLS est requis.") + await client.disconnect() + return False + # D1 : attente bornée de la session XMPP. On attend soit + # l'établissement de la session, soit un échec d'authentification, + # avec un timeout afin d'éviter tout blocage indéfini (et de couvrir + # le cas où ``failed_auth`` survient après ``connect()``). + _done, _pending = await asyncio.wait( + [ + asyncio.create_task(session_started.wait()), + asyncio.create_task(failed_auth.wait()), + ], + timeout=settings.timeout, + return_when=asyncio.FIRST_COMPLETED, + ) + for task in _pending: + task.cancel() + if failed_auth.is_set(): + # SEC-XMPP-04 : déconnexion propre puis avertissement non bloquant. + await client.disconnect() + raise PipelineWarning("Authentification XMPP échouée.", step="xmpp") from None + if not session_started.is_set(): + # Timeout : aucune session établie avant l'échéance. + await client.disconnect() + raise PipelineWarning("Délai d'attente de session XMPP dépassé.", step="xmpp") from None + client.send_message(mto=JID(settings.to), mbody=formatted, mtype="chat") + await client.disconnect() + return True + + +class SyncXmppChannel: + """Adaptateur synchrone du canal XMPP pour le pipeline (U5). + + Enveloppe une instance de :class:`XmppChannel` pour offrir une interface + synchrone conforme au :class:`~pronote_sync.channels.protocol.Channel` + (D4). :meth:`send` n'exécute jamais d'event loop manuellement : il + délègue à :func:`asyncio.run` (qui gère sa propre loop) lorsqu'aucune + boucle n'est en cours, ou bascule dans un thread démon borné par un timeout + lorsque l'appel a lieu depuis une boucle déjà active. Toute erreur est + convertie en retour ``False`` sans jamais lever. + + :ivar settings: Paramètres XMPP. + :vartype settings: XmppSettings + :ivar dry_run: Mode simulation (aucun envoi réseau). + :vartype dry_run: bool + """ + + def __init__(self, settings: XmppSettings, dry_run: bool = False) -> None: + """Initialise l'adaptateur synchrone et son canal interne. + + :param settings: Paramètres de configuration du canal XMPP. + :param dry_run: Si ``True``, l'envoi est simulé. + """ + self.settings = settings + self.dry_run = dry_run + self._channel = XmppChannel(settings, dry_run) + + def send(self, message: XmppMessage) -> bool: + """Envoie un message XMPP de façon synchrone et sans lever. + + Si aucune event loop n'est en cours d'exécution, délègue directement à + :func:`asyncio.run` sur :meth:`XmppChannel.send_async`. Si une boucle + tourne déjà, exécute l'envoi dans un thread démon joint avec un timeout + (``settings.timeout``). Retourne ``True`` en cas de succès et ``False`` + sur toute erreur (dont timeout et exceptions), en journalisant une + version expurgée sans secret. + + :param message: Message final à envoyer. + :return: ``True`` si l'envoi a réussi, ``False`` sinon. + :rtype: bool + """ + try: + asyncio.get_running_loop() + except RuntimeError: + running_loop = False + else: + running_loop = True + + if not running_loop: + try: + return asyncio.run(self._channel.send_async(message)) + except Exception as exc: + self._log_error(exc) + return False + + result: list[bool] = [] + error: list[Exception] = [] + + def _run() -> None: + """Exécute l'envoi asynchrone dans le thread démon.""" + try: + result.append(asyncio.run(self._channel.send_async(message))) + except Exception as exc: + error.append(exc) + + thread = threading.Thread(target=_run, daemon=True) + thread.start() + thread.join(timeout=self._channel.settings.timeout) + if thread.is_alive(): + logger.warning("XMPP : timeout lors de l'envoi synchrone.") + return False + if error: + self._log_error(error[0]) + return False + return bool(result and result[0]) + + def _log_error(self, exc: Exception) -> None: + """Journalise une erreur d'envoi avec masquage des secrets. + + :param exc: Exception à journaliser de façon expurgée. + """ + redacted = redact_exception(exc) + redacted = redact_secrets(redacted, extra_secrets=_secret_values(self._channel.settings)) + logger.warning("XMPP : erreur lors de l'envoi synchrone : %s", redacted) diff --git a/pyproject.toml b/pyproject.toml index 94759ce..9b17fd5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -121,6 +121,10 @@ strict = true module = "litellm" ignore_missing_imports = true +[[tool.mypy.overrides]] +module = "slixmpp" +ignore_missing_imports = true + [[tool.mypy.overrides]] module = "openai.*" follow_imports = "skip" diff --git a/tests/unit/test_xmpp_channel.py b/tests/unit/test_xmpp_channel.py new file mode 100644 index 0000000..4238444 --- /dev/null +++ b/tests/unit/test_xmpp_channel.py @@ -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="

Faire l'exercice 5 page 42

", + ) + 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="

Sortie prévue le 20/09.

", + 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 le chapitre 3 et répondre aux questions.", + html="

Lire le chapitre 3 et répondre aux questions.

", + ) + 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 "" 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 diff --git a/tests/unit/test_xmpp_sync_adapter.py b/tests/unit/test_xmpp_sync_adapter.py new file mode 100644 index 0000000..0c7f200 --- /dev/null +++ b/tests/unit/test_xmpp_sync_adapter.py @@ -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