college-infos/tests/unit/test_xmpp_sync_adapter.py

"""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))
- 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 MagicMock, patch

import pytest
from pydantic import SecretStr

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
    """
    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.
    """

    @patch("pronote_sync.channels.xmpp.asyncio.run")
    def test_sync_adapter_send_returns_true_on_success(
        self, mock_asyncio_run: MagicMock, xmpp_settings: XmppSettings, xmpp_message: XmppMessage
    ) -> None:
        """Test que send retourne True en cas de succès.

        :param mock_asyncio_run: Mock de asyncio.run
        :param xmpp_settings: Paramètres XMPP valides.
        :param xmpp_message: Message XMPP minimal.
        """
        mock_asyncio_run.side_effect = lambda coro: coro.close() or True
        channel = SyncXmppChannel(xmpp_settings)
        result = channel.send(xmpp_message)
        assert result is True

    @patch("pronote_sync.channels.xmpp.asyncio.run")
    def test_sync_adapter_send_returns_false_on_error(
        self, mock_run: MagicMock, 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 mock_run: Mock de asyncio.run
        :param xmpp_settings: Paramètres XMPP valides.
        :param xmpp_message: Message XMPP minimal.
        """

        # Simuler une erreur dans asyncio.run
        def _run_with_error(coro: object) -> bool:
            """Ferme la coroutine non exécutée puis lève l'erreur simulée."""
            close = getattr(coro, "close", None)
            if close is not None:
                close()
            raise RuntimeError("Connexion impossible")

        mock_run.side_effect = _run_with_error
        channel = SyncXmppChannel(xmpp_settings)
        result = channel.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)

    def test_sync_adapter_dry_run_does_not_create_client(
        self, xmpp_settings: XmppSettings, xmpp_message: XmppMessage
    ) -> None:
        """Test que dry_run=True ne crée jamais ClientXMPP.

        :param xmpp_settings: Paramètres XMPP valides.
        :param xmpp_message: Message XMPP minimal.
        """
        channel = SyncXmppChannel(xmpp_settings, dry_run=True)
        with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls:
            result = channel.send(xmpp_message)
            assert result is True
            # ClientXMPP ne doit pas être instancié en dry_run
            assert not mock_cls.called

    def test_sync_adapter_dry_run_returns_true(
        self, xmpp_settings: XmppSettings, xmpp_message: XmppMessage
    ) -> None:
        """Test que dry_run=True retourne True sans se connecter.

        :param xmpp_settings: Paramètres XMPP valides.
        :param xmpp_message: Message XMPP minimal.
        """
        channel = SyncXmppChannel(xmpp_settings, dry_run=True)
        result = channel.send(xmpp_message)
        assert result is True

    @patch("pronote_sync.channels.xmpp.asyncio.run")
    def test_sync_adapter_never_raises(
        self, mock_run: MagicMock, xmpp_settings: XmppSettings, xmpp_message: XmppMessage
    ) -> None:
        """Test que send ne lève jamais d'exception.

        :param mock_run: Mock de asyncio.run
        :param xmpp_settings: Paramètres XMPP valides.
        :param xmpp_message: Message XMPP minimal.
        """

        # Simuler une erreur quelconque
        def _run_with_error(coro: object) -> bool:
            """Ferme la coroutine non exécutée puis lève l'erreur simulée."""
            close = getattr(coro, "close", None)
            if close is not None:
                close()
            raise Exception("Any error")

        mock_run.side_effect = _run_with_error
        channel = SyncXmppChannel(xmpp_settings)
        result = channel.send(xmpp_message)
        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
        settings = XmppSettings(
            enabled=True,
            jid=BOT_SENTINEL_JID,
            password=SecretStr(PASS_SENTINEL_123),
            host="localhost",
            port=5222,
            to=RECIPIENT_SENTINEL,
            use_tls=False,
        )

        channel = SyncXmppChannel(settings)
        msg = XmppMessage(
            target_date=date(2025, 9, 7),
            synthesis=None,
            homeworks=(),
            changes=(),
            messages=(),
            external_info=None,
        )

        # Simuler une erreur dans asyncio.run
        with patch("pronote_sync.channels.xmpp.asyncio.run") as mock_run:

            def _run_with_error(coro: object) -> bool:
                """Ferme la coroutine non exécutée puis lève l'erreur simulée."""
                close = getattr(coro, "close", None)
                if close is not None:
                    close()
                raise RuntimeError("Connexion impossible")

            mock_run.side_effect = _run_with_error
            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