fix(xmpp): remplacer use_tls par tls_mode et borner les phases de connexion
Cause racine (#25) : use_tls=True (défaut) activait le direct TLS sur le port 5222 (conventionnellement STARTTLS). Le client envoyait un ClientHello TLS sur un port attendant un stream XMPP en clair, le serveur ne voyait jamais l'identité configurée, et la session expirait après 30 s. Corrections : - Remplacer use_tls (bool) par tls_mode: Literal[direct|starttls|disabled] (défaut starttls, compatible port 5222). use_tls conservé comme alias déprécié avec DeprecationWarning. - Ajouter connect_timeout (15 s) et cleanup_timeout (10 s) distincts du timeout de session (30 s). - Gérer l'événement connection_failed de Slixmpp pour échouer rapidement au lieu d'attendre le timeout de session. - Borner await connect_future et await disconnect_future par leurs timeouts respectifs (anti-blocage). - Attendre connect_future et session_future conjointement (asyncio.wait, FIRST_COMPLETED) pour détecter connection_failed avant l'expiration du connect_timeout. - Annuler les tâches pending sur tous les chemins de retour, y compris CancelledError et Exception. - Redact tous les redact_exception avec extra_secrets=_secret_values(). - Construire ClientXMPP dans le try (contrat « never raises »). - Enrichir FakeClientXMPP avec modes connect/disconnect configurables. - 25 nouveaux tests (timeout connexion, connection_failed, cleanup bloqué, CancelledError, TLS mismatch, fuite secrets). Couverture 96 %. - Mettre à jour .env.example, GUIDE_DEV_PYTHON.md, README.LLM.md. Co-authored-by: OpenCode <opencode@antoineve.me>
This commit is contained in:
@@ -17,7 +17,7 @@ from unittest.mock import patch
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from pronote_sync.channels.xmpp import XmppChannel, XmppMessage
|
||||
from pronote_sync.channels.xmpp import SyncXmppChannel, XmppChannel, XmppMessage
|
||||
from pronote_sync.config.settings import XmppSettings
|
||||
from pronote_sync.models.agenda import Lesson, TheoreticalLesson
|
||||
from pronote_sync.models.blog import BlogArticle, ExternalInfo
|
||||
@@ -48,11 +48,15 @@ class FakeClientXMPP:
|
||||
self._should_disconnect_early = False
|
||||
self._host_used: str | None = None
|
||||
self._port_used: int | None = None
|
||||
# New modes for realistic failure simulation
|
||||
self._connect_mode: str = "ok" # "ok", "pending", "connection_failed"
|
||||
self._disconnect_mode: str = "ok" # "ok", "pending"
|
||||
self._fire_connection_failed: bool = False
|
||||
|
||||
def add_event_handler(
|
||||
self, name: str, pointer: Callable[..., object], disposable: bool = False
|
||||
) -> None:
|
||||
if name not in ("session_start", "failed_auth", "disconnected"):
|
||||
if name not in ("session_start", "failed_auth", "connection_failed", "disconnected"):
|
||||
raise AssertionError(f"Unsupported event: {name}")
|
||||
self.handlers.setdefault(name, []).append(pointer)
|
||||
|
||||
@@ -63,10 +67,24 @@ class FakeClientXMPP:
|
||||
self.connected = True
|
||||
self._host_used = host
|
||||
self._port_used = port
|
||||
# Schedule event handlers to fire after connect returns
|
||||
loop.call_soon(self._fire_events)
|
||||
future.set_result(True)
|
||||
return future
|
||||
|
||||
if self._connect_mode == "pending":
|
||||
# Never resolves - simulates connection timeout
|
||||
return future
|
||||
elif self._connect_mode == "connection_failed":
|
||||
# Resolves connect future but fires connection_failed event
|
||||
loop.call_soon(self._fire_connection_failed_event)
|
||||
future.set_result(True)
|
||||
return future
|
||||
else:
|
||||
# Schedule event handlers to fire after connect returns
|
||||
loop.call_soon(self._fire_events)
|
||||
future.set_result(True)
|
||||
return future
|
||||
|
||||
def _fire_connection_failed_event(self) -> None:
|
||||
"""Fire connection_failed event for testing."""
|
||||
self._fire("connection_failed")
|
||||
|
||||
def _fire_events(self) -> None:
|
||||
if self._should_disconnect_early:
|
||||
@@ -86,8 +104,13 @@ class FakeClientXMPP:
|
||||
loop = asyncio.get_event_loop()
|
||||
future: asyncio.Future[bool] = loop.create_future()
|
||||
self.disconnected = True
|
||||
future.set_result(True)
|
||||
return future
|
||||
|
||||
if self._disconnect_mode == "pending":
|
||||
# Never resolves - simulates cleanup timeout
|
||||
return future
|
||||
else:
|
||||
future.set_result(True)
|
||||
return future
|
||||
|
||||
def send_message(
|
||||
self, mto: object, mbody: str | None = None, mtype: str | None = None, **kwargs: object
|
||||
@@ -681,6 +704,339 @@ class TestXmppChannelSend:
|
||||
jid_arg = call_args.args[0]
|
||||
assert jid_arg == "bot@example.com/myresource"
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_connect_timeout_returns_false(self, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""Test que connect_timeout retourne False quand connect() ne résout pas.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
connect_timeout=0.05,
|
||||
timeout=30,
|
||||
cleanup_timeout=0.01,
|
||||
)
|
||||
|
||||
class PendingConnectClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
self._connect_mode = "pending"
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
result = await channel.send_async(msg)
|
||||
assert result is False
|
||||
# Vérifier que le log contient "connexion"
|
||||
logs = caplog.text
|
||||
assert "connexion" in logs.lower()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cancelled_error_cancels_pending_tasks(self) -> None:
|
||||
"""Test que l'annulation de send_async annule les tâches encore en attente.
|
||||
|
||||
Un ``connect()`` qui ne résout jamais et une annulation de la tâche
|
||||
appelante doivent entraîner la cancellation de la future de connexion
|
||||
(idempotence de ``_cancel_pending`` sur les chemins d'exception).
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
connect_timeout=15,
|
||||
timeout=30,
|
||||
cleanup_timeout=0.01,
|
||||
)
|
||||
|
||||
class PendingConnectClient(FakeClientXMPP):
|
||||
"""Client dont ``connect()`` retourne une future jamais résolue."""
|
||||
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
self._connect_mode = "pending"
|
||||
self.connect_future: asyncio.Future[bool] | None = None
|
||||
|
||||
def connect(
|
||||
self, host: str | None = None, port: int | None = None
|
||||
) -> asyncio.Future[bool]:
|
||||
future = super().connect(host, port)
|
||||
self.connect_future = future
|
||||
return future
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
client = PendingConnectClient("bot@example.com", "secret123")
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=lambda j, p: client):
|
||||
task = asyncio.ensure_future(channel.send_async(msg))
|
||||
# Laisse asyncio.wait démarrer et la future de connexion rester en attente
|
||||
await asyncio.sleep(0.05)
|
||||
task.cancel()
|
||||
result = await task
|
||||
assert result is False
|
||||
assert client.connect_future is not None
|
||||
assert client.connect_future.cancelled()
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_connection_failed_event_returns_false(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test que connection_failed event retourne False rapidement.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
connect_timeout=15,
|
||||
timeout=30,
|
||||
cleanup_timeout=0.01,
|
||||
)
|
||||
|
||||
class ConnectionFailedClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
self._connect_mode = "connection_failed"
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectionFailedClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
result = await channel.send_async(msg)
|
||||
assert result is False
|
||||
# Vérifier que le log contient une mention d'échec réseau
|
||||
logs = caplog.text
|
||||
assert "réseau" in logs.lower() or "connexion" in logs.lower()
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_cleanup_timeout_does_not_hang(self, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""Test que cleanup_timeout ne bloque pas quand disconnect() ne résout pas.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
import logging
|
||||
|
||||
caplog.set_level(logging.DEBUG)
|
||||
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
connect_timeout=0.05,
|
||||
timeout=30,
|
||||
cleanup_timeout=0.01,
|
||||
)
|
||||
|
||||
class HangingDisconnectClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
self._disconnect_mode = "pending"
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=HangingDisconnectClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
# Should complete quickly despite hanging disconnect
|
||||
result = await asyncio.wait_for(channel.send_async(msg), timeout=0.5)
|
||||
assert result is True
|
||||
# Vérifier que le log contient une mention de timeout de nettoyage
|
||||
logs = caplog.text
|
||||
assert "déconnexion" in logs.lower()
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_tls_mode_direct_config(self) -> None:
|
||||
"""Test que tls_mode='direct' configure enable_direct_tls=True et enable_starttls=False.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="direct",
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
class InspectClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls:
|
||||
mock_cls.return_value = InspectClient("bot@example.com", "secret123")
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
await channel.send_async(msg)
|
||||
client_instance = mock_cls.return_value
|
||||
assert client_instance.enable_direct_tls is True
|
||||
assert client_instance.enable_starttls is False
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_tls_mode_starttls_config(self) -> None:
|
||||
"""Test que tls_mode='starttls' configure enable_starttls=True et enable_direct_tls=False.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
class InspectClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls:
|
||||
mock_cls.return_value = InspectClient("bot@example.com", "secret123")
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
await channel.send_async(msg)
|
||||
client_instance = mock_cls.return_value
|
||||
assert client_instance.enable_starttls is True
|
||||
assert client_instance.enable_direct_tls is False
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_tls_mode_disabled_config(self) -> None:
|
||||
"""Test que tls_mode='disabled' avec host='127.0.0.1' configure TLS désactivé.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="127.0.0.1",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="disabled",
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
class InspectClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls:
|
||||
mock_cls.return_value = InspectClient("bot@example.com", "secret123")
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
await channel.send_async(msg)
|
||||
client_instance = mock_cls.return_value
|
||||
assert client_instance.enable_direct_tls is False
|
||||
assert client_instance.enable_starttls is False
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_secret_leak_in_connection_failure_log(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test qu'aucun secret ne fuit dans les logs en cas d'échec de connexion.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
sentinel_password = "SECRET_PASSWORD_XMPP_12345" # pragma: allowlist secret
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr(sentinel_password),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
class ConnectionFailedClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
self._connect_mode = "connection_failed"
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectionFailedClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
await channel.send_async(msg)
|
||||
|
||||
# Vérifier que le mot de passe sentinelle n'apparaît pas dans les logs
|
||||
logs = caplog.text
|
||||
assert sentinel_password not in logs
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_secret_leak_in_connect_timeout_log(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test qu'aucun secret ne fuit dans les logs en cas de timeout de connexion.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
sentinel_password = "SECRET_PASSWORD_XMPP_12345"
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr(sentinel_password),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
connect_timeout=0.05,
|
||||
timeout=30,
|
||||
cleanup_timeout=0.01,
|
||||
)
|
||||
|
||||
class PendingConnectClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
self._connect_mode = "pending"
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
await channel.send_async(msg)
|
||||
|
||||
# Vérifier que le mot de passe sentinelle n'apparaît pas dans les logs
|
||||
logs = caplog.text
|
||||
assert sentinel_password not in logs
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_async_tls_direct_config(self) -> None:
|
||||
@@ -816,6 +1172,328 @@ class TestXmppChannelSend:
|
||||
result = await channel.send_async(msg)
|
||||
assert result is True
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_async_connect_raises_exception_returns_false(self) -> None:
|
||||
"""Test que connect() levant une exception retourne False.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
class ConnectExceptionClient(FakeClientXMPP):
|
||||
def connect(
|
||||
self, host: str | None = None, port: int | None = None
|
||||
) -> asyncio.Future[bool]:
|
||||
raise ConnectionError("Network unreachable")
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectExceptionClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
result = await channel.send_async(msg)
|
||||
assert result is False
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_async_connect_future_error_returns_false(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test que la future de connect() résolue avec une exception retourne False.
|
||||
|
||||
Le message d'erreur doit être expurgé et ne pas contenir les secrets.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
sentinel_password = "SECRET_PASSWORD_XMPP_CONNECT_FUT" # pragma: allowlist secret
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr(sentinel_password),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
class ConnectFutureErrorClient(FakeClientXMPP):
|
||||
def connect(
|
||||
self, host: str | None = None, port: int | None = None
|
||||
) -> asyncio.Future[bool]:
|
||||
loop = asyncio.get_event_loop()
|
||||
future: asyncio.Future[bool] = loop.create_future()
|
||||
future.set_exception(ConnectionError("Network unreachable"))
|
||||
return future
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectFutureErrorClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
result = await channel.send_async(msg)
|
||||
assert result is False
|
||||
assert "Échec de connexion XMPP" in caplog.text
|
||||
assert sentinel_password not in caplog.text
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_async_session_failure_after_connect_returns_false(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test qu'un échec de session après connexion résolue retourne False.
|
||||
|
||||
Le connect future est résolu avec succès, puis l'événement
|
||||
``failed_auth`` arrive pendant l'attente de session.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
timeout=5,
|
||||
cleanup_timeout=0.01,
|
||||
)
|
||||
|
||||
class LateAuthFailClient(FakeClientXMPP):
|
||||
def connect(
|
||||
self, host: str | None = None, port: int | None = None
|
||||
) -> asyncio.Future[bool]:
|
||||
loop = asyncio.get_event_loop()
|
||||
future: asyncio.Future[bool] = loop.create_future()
|
||||
self.connected = True
|
||||
self._host_used = host
|
||||
self._port_used = port
|
||||
loop.call_later(0.01, lambda: self._fire("failed_auth"))
|
||||
future.set_result(True)
|
||||
return future
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=LateAuthFailClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
result = await channel.send_async(msg)
|
||||
assert result is False
|
||||
assert "authentification" in caplog.text.lower()
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_async_cancelled_returns_false(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test que l'annulation de la tâche retourne False sans lever.
|
||||
|
||||
Le contrat du canal impose un retour booléen : ``CancelledError`` doit
|
||||
être interceptée et convertie en ``False``.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
import logging
|
||||
|
||||
caplog.set_level(logging.DEBUG)
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
tls_mode="starttls",
|
||||
connect_timeout=15,
|
||||
timeout=30,
|
||||
cleanup_timeout=0.01,
|
||||
)
|
||||
|
||||
class PendingConnectClient(FakeClientXMPP):
|
||||
def __init__(self, jid: str, password: str) -> None:
|
||||
super().__init__(jid, password)
|
||||
self._connect_mode = "pending"
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient):
|
||||
channel = XmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
task = asyncio.ensure_future(channel.send_async(msg))
|
||||
await asyncio.sleep(0.001)
|
||||
task.cancel()
|
||||
result = await asyncio.wait_for(task, timeout=1.0)
|
||||
assert result is False
|
||||
assert "annulé" in caplog.text.lower()
|
||||
|
||||
|
||||
class TestSecretValues:
|
||||
"""Tests pour la fonction _secret_values."""
|
||||
|
||||
def test_secret_values_with_all_secrets(self) -> None:
|
||||
"""Test que _secret_values retourne tous les secrets.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
from pronote_sync.channels.xmpp import _secret_values
|
||||
|
||||
settings = XmppSettings(
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
to="parent@example.com",
|
||||
)
|
||||
secrets = _secret_values(settings)
|
||||
assert len(secrets) == 3
|
||||
assert "bot@example.com" in secrets
|
||||
assert settings.password in secrets
|
||||
assert "parent@example.com" in secrets
|
||||
|
||||
def test_secret_values_with_none_values(self) -> None:
|
||||
"""Test que _secret_values filtre les valeurs None.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
from pronote_sync.channels.xmpp import _secret_values
|
||||
|
||||
settings = XmppSettings(
|
||||
jid=None,
|
||||
password=None,
|
||||
to=None,
|
||||
)
|
||||
secrets = _secret_values(settings)
|
||||
assert len(secrets) == 0
|
||||
|
||||
def test_secret_values_with_some_none(self) -> None:
|
||||
"""Test que _secret_values gère les valeurs partiellement None.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
from pronote_sync.channels.xmpp import _secret_values
|
||||
|
||||
settings = XmppSettings(
|
||||
jid="bot@example.com",
|
||||
password=None,
|
||||
to="parent@example.com",
|
||||
)
|
||||
secrets = _secret_values(settings)
|
||||
assert len(secrets) == 2
|
||||
assert "bot@example.com" in secrets
|
||||
assert "parent@example.com" in secrets
|
||||
|
||||
|
||||
class TestSyncXmppChannel:
|
||||
"""Tests unitaires pour la classe SyncXmppChannel."""
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
def test_sync_send_dry_run_returns_true(self) -> None:
|
||||
"""Test que SyncXmppChannel en dry_run retourne True.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
use_tls=True,
|
||||
timeout=30,
|
||||
)
|
||||
channel = SyncXmppChannel(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=FakeClientXMPP)
|
||||
def test_sync_send_success_returns_true(self) -> None:
|
||||
"""Test que SyncXmppChannel.send retourne True en cas de succès.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
use_tls=True,
|
||||
timeout=30,
|
||||
)
|
||||
channel = SyncXmppChannel(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=FakeClientXMPP)
|
||||
def test_sync_send_exception_returns_false(self) -> None:
|
||||
"""Test que SyncXmppChannel.send retourne False en cas d'exception.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
use_tls=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
class ErrorClient(FakeClientXMPP):
|
||||
def connect(
|
||||
self, host: str | None = None, port: int | None = None
|
||||
) -> asyncio.Future[bool]:
|
||||
raise RuntimeError("Connection failed")
|
||||
|
||||
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ErrorClient):
|
||||
channel = SyncXmppChannel(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
|
||||
|
||||
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
|
||||
def test_sync_send_asyncio_run_error_returns_false(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test que send retourne False si asyncio.run lève une exception.
|
||||
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
"""
|
||||
settings = XmppSettings(
|
||||
enabled=True,
|
||||
jid="bot@example.com",
|
||||
password=SecretStr("secret123"),
|
||||
host="xmpp.example.com",
|
||||
port=5222,
|
||||
to="parent@example.com",
|
||||
resource="pronote-sync",
|
||||
use_tls=True,
|
||||
timeout=30,
|
||||
)
|
||||
channel = SyncXmppChannel(settings, dry_run=False)
|
||||
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
|
||||
with patch(
|
||||
"pronote_sync.channels.xmpp.asyncio.run",
|
||||
side_effect=RuntimeError("Boucle événementielle indisponible"),
|
||||
):
|
||||
result = channel.send(msg)
|
||||
assert result is False
|
||||
assert "erreur" in caplog.text.lower()
|
||||
|
||||
|
||||
class TestXmppChannelSecurity:
|
||||
"""Tests de sécurité pour XmppChannel (non-fuite de secrets).
|
||||
|
||||
Reference in New Issue
Block a user