Antoine Van Elstraete
b2106e75ac
Fix all 8 findings from the independent review (FIXME_M10.md): #1 Transport compatible with slixmpp 1.17.0 (D5): - Use real ClientXMPP type (remove Any), JID with resource - connect(host, port) explicit, no use_tls kwarg - enable_direct_tls/enable_starttls configured before connect - Single timeout via asyncio.Future for session_start/failed_auth/disconnected - Remove premature 'starttls' in features check, remove auto_reconnect - try/finally guarantees disconnect on all paths (#4) #2 Factory dry_run no longer bypassed (D6): - Single send() entry point in SyncXmppChannel - dry_run check before any ClientXMPP creation - Remove XmppChannel.send() dual implementation #3 Thread daemon removed — single asyncio.run(), documented limitation #5 Richer message format: - Target date header, change type [Ajouté/Supprimé/Modifié] - Lesson times, homework due date, message author - No pronote_messages duplication (external_info = blog + other_info only) #6 Error contract unified (D6): - Channel.send() -> bool never raises PipelineWarning - Errors logged with redaction, returns False - PipelineWarning(step='xmpp') will be created by pipeline M11 #7 Tests faithful to slixmpp 1.17.0 API: - FakeClientXMPP with real connect(host,port)/disconnect() signatures - Assertions on host, port, resource, mtype='chat' - No RuntimeWarning from unawaited coroutines #8 .secrets.baseline restored from main Coverage: 96.44% on channels/, 600 tests pass, pre-commit all-files green. Co-authored-by: opencode/coder <coder@agents.invalid> Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
325 lines
10 KiB
Python
325 lines
10 KiB
Python
"""Tests unitaires pour la factory get_channel des canaux XMPP.
|
|
|
|
Ce module valide la spécification de la factory ``get_channel`` qui sera
|
|
implémentée dans ``pronote_sync/channels/__init__.py``.
|
|
|
|
Les tests doivent être initialement en échec (RED) car la factory n'existe
|
|
pas encore dans le code de production.
|
|
|
|
Spécification (D2) :
|
|
- get_channel(settings: XmppSettings, dry_run: bool = False) -> Channel | None
|
|
- Si enabled=False → retourne None (pas d'exception, pas d'avertissement).
|
|
- Si enabled=True et champs requis manquants (jid, password, to, host) →
|
|
journalise un avertissement avec redact_secrets(), retourne None.
|
|
- Si enabled=True et tous champs requis présents → construit et retourne
|
|
une instance de SyncXmppChannel (ou XmppChannel).
|
|
- La factory n'élève jamais d'exception.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from pydantic import SecretStr
|
|
|
|
from pronote_sync.channels import (
|
|
Channel,
|
|
SyncXmppChannel,
|
|
XmppChannel,
|
|
get_channel,
|
|
)
|
|
from pronote_sync.config.settings import XmppSettings
|
|
|
|
|
|
class TestGetChannelDisabled:
|
|
"""Tests pour le cas où le canal XMPP est désactivé (enabled=False)."""
|
|
|
|
def test_get_channel_disabled_returns_none(self) -> None:
|
|
"""Vérifie que get_channel retourne None quand enabled=False.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(enabled=False)
|
|
result = get_channel(settings)
|
|
assert result is None
|
|
|
|
|
|
class TestGetChannelEnabledComplete:
|
|
"""Tests pour le cas où le canal est activé avec une configuration complète."""
|
|
|
|
def test_get_channel_enabled_complete_returns_channel(self) -> None:
|
|
"""Vérifie que get_channel retourne une instance de channel quand la configuration est complète.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(
|
|
enabled=True,
|
|
jid="bot@example.com",
|
|
password=SecretStr("pass"),
|
|
host="example.com",
|
|
to="parent@example.com",
|
|
)
|
|
result = get_channel(settings)
|
|
assert result is not None
|
|
# Vérifie que le résultat implémente le Protocol Channel
|
|
assert isinstance(result, Channel)
|
|
|
|
|
|
class TestGetChannelEnabledMissingRequiredFields:
|
|
"""Tests pour les cas où des champs requis sont manquants."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"settings_kwargs",
|
|
[
|
|
{
|
|
"enabled": True,
|
|
"jid": None,
|
|
"password": SecretStr("pass"),
|
|
"host": "example.com",
|
|
"to": "parent@example.com",
|
|
},
|
|
{
|
|
"enabled": True,
|
|
"jid": "bot@example.com",
|
|
"password": None,
|
|
"host": "example.com",
|
|
"to": "parent@example.com",
|
|
},
|
|
{
|
|
"enabled": True,
|
|
"jid": "bot@example.com",
|
|
"password": SecretStr("pass"),
|
|
"host": "",
|
|
"to": "parent@example.com",
|
|
},
|
|
{
|
|
"enabled": True,
|
|
"jid": "bot@example.com",
|
|
"password": SecretStr("pass"),
|
|
"host": "example.com",
|
|
"to": None,
|
|
},
|
|
],
|
|
ids=["missing_jid", "missing_password", "missing_host", "missing_to"],
|
|
)
|
|
def test_get_channel_enabled_missing_required_field_returns_none(
|
|
self,
|
|
settings_kwargs: dict[str, Any],
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Vérifie que get_channel retourne None quand un champ requis est manquant.
|
|
|
|
:param settings_kwargs: Paramètres pour XmppSettings avec un champ manquant.
|
|
:param caplog: Fixture pour capturer les logs.
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(**settings_kwargs)
|
|
result = get_channel(settings)
|
|
assert result is None
|
|
# Vérifie qu'un avertissement a été journalisé
|
|
assert len(caplog.records) > 0
|
|
assert any(record.levelname == "WARNING" for record in caplog.records)
|
|
|
|
def test_get_channel_enabled_missing_jid_returns_none(self) -> None:
|
|
"""Vérifie que get_channel retourne None quand jid est None.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(
|
|
enabled=True,
|
|
jid=None,
|
|
password=SecretStr("pass"),
|
|
host="example.com",
|
|
to="parent@example.com",
|
|
)
|
|
result = get_channel(settings)
|
|
assert result is None
|
|
|
|
def test_get_channel_enabled_missing_password_returns_none(self) -> None:
|
|
"""Vérifie que get_channel retourne None quand password est None.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(
|
|
enabled=True,
|
|
jid="bot@example.com",
|
|
password=None,
|
|
host="example.com",
|
|
to="parent@example.com",
|
|
)
|
|
result = get_channel(settings)
|
|
assert result is None
|
|
|
|
def test_get_channel_enabled_missing_to_returns_none(self) -> None:
|
|
"""Vérifie que get_channel retourne None quand to est None.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(
|
|
enabled=True,
|
|
jid="bot@example.com",
|
|
password=SecretStr("pass"),
|
|
host="example.com",
|
|
to=None,
|
|
)
|
|
result = get_channel(settings)
|
|
assert result is None
|
|
|
|
def test_get_channel_enabled_missing_host_returns_none(self) -> None:
|
|
"""Vérifie que get_channel retourne None quand host est vide.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(
|
|
enabled=True,
|
|
jid="bot@example.com",
|
|
password=SecretStr("pass"),
|
|
host="",
|
|
to="parent@example.com",
|
|
)
|
|
result = get_channel(settings)
|
|
assert result is None
|
|
|
|
|
|
class TestGetChannelNoExceptionOnMisconfiguration:
|
|
"""Tests pour vérifier que la factory ne lève jamais d'exception."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"settings_kwargs",
|
|
[
|
|
{"enabled": True, "jid": None},
|
|
{"enabled": True, "password": None},
|
|
{"enabled": True, "to": None},
|
|
{"enabled": True, "host": ""},
|
|
{"enabled": True, "jid": None, "password": None, "to": None, "host": ""},
|
|
{"enabled": False},
|
|
],
|
|
ids=[
|
|
"missing_jid_only",
|
|
"missing_password_only",
|
|
"missing_to_only",
|
|
"missing_host_only",
|
|
"all_missing",
|
|
"disabled",
|
|
],
|
|
)
|
|
def test_get_channel_no_exception_on_misconfiguration(
|
|
self,
|
|
settings_kwargs: dict[str, Any],
|
|
) -> None:
|
|
"""Vérifie que get_channel ne lève jamais d'exception sur une configuration invalide.
|
|
|
|
:param settings_kwargs: Paramètres pour XmppSettings potentiellement invalides.
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(**settings_kwargs)
|
|
# Ne doit jamais lever d'exception
|
|
result = get_channel(settings)
|
|
assert result is None
|
|
|
|
|
|
class TestGetChannelNoSecretInWarningLog:
|
|
"""Tests pour vérifier que les secrets ne fuient pas dans les logs."""
|
|
|
|
def test_get_channel_no_secret_in_warning_log(self, caplog: pytest.LogCaptureFixture) -> None:
|
|
"""Vérifie que les valeurs sentinelles ne apparaissent pas dans les logs.
|
|
|
|
Utilise des valeurs sentinelles pour éviter toute fuite de secrets réels.
|
|
|
|
:param caplog: Fixture pour capturer les logs.
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
sentinel_jid = "JID_SENTINEL@example.com"
|
|
sentinel_password = SecretStr("PASS_SENTINEL")
|
|
sentinel_to = "TO_SENTINEL@example.com"
|
|
|
|
# Configuration incomplète : host manquant -> get_channel journalise
|
|
# un avertissement expurgé et retourne None.
|
|
settings = XmppSettings(
|
|
enabled=True,
|
|
jid=sentinel_jid,
|
|
password=sentinel_password,
|
|
host="",
|
|
to=sentinel_to,
|
|
)
|
|
result = get_channel(settings)
|
|
assert result is None
|
|
|
|
# Vérifie qu'un avertissement a été journalisé
|
|
assert len(caplog.records) > 0
|
|
assert any(record.levelname == "WARNING" for record in caplog.records)
|
|
|
|
# Vérifie que les valeurs sentinelles n'apparaissent pas dans les logs
|
|
log_text = "".join(record.message for record in caplog.records)
|
|
assert sentinel_jid not in log_text
|
|
assert sentinel_password.get_secret_value() not in log_text
|
|
assert sentinel_to not in log_text
|
|
|
|
|
|
class TestGetChannelDryRun:
|
|
"""Tests pour le flag dry_run."""
|
|
|
|
def test_get_channel_dry_run(self) -> None:
|
|
"""Vérifie que le flag dry_run est passé à travers et retourne un channel.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(
|
|
enabled=True,
|
|
jid="bot@example.com",
|
|
password=SecretStr("pass"),
|
|
host="example.com",
|
|
to="parent@example.com",
|
|
)
|
|
result = get_channel(settings, dry_run=True)
|
|
assert result is not None
|
|
assert isinstance(result, Channel)
|
|
|
|
def test_get_channel_dry_run_no_connection(self) -> None:
|
|
"""Vérifie que dry_run=True ne crée pas de ClientXMPP.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
settings = XmppSettings(
|
|
enabled=True,
|
|
jid="bot@example.com",
|
|
password=SecretStr("pass"),
|
|
host="example.com",
|
|
to="parent@example.com",
|
|
)
|
|
with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls:
|
|
result = get_channel(settings, dry_run=True)
|
|
assert result is not None
|
|
assert isinstance(result, Channel)
|
|
# ClientXMPP ne doit pas être instancié en dry_run
|
|
assert not mock_cls.called
|
|
|
|
|
|
class TestChannelImportsFromInit:
|
|
"""Tests pour vérifier que les exports depuis __init__.py fonctionnent."""
|
|
|
|
def test_channel_imports_from_init(self) -> None:
|
|
"""Vérifie que Channel, XmppChannel, SyncXmppChannel sont importables depuis pronote_sync.channels.
|
|
|
|
:return: None
|
|
:rtype: None
|
|
"""
|
|
# Ces imports doivent réussir
|
|
assert Channel is not None
|
|
assert XmppChannel is not None
|
|
assert SyncXmppChannel is not None
|
|
assert get_channel is not None
|