get_channel(settings, dry_run=False) -> Channel | None with: - enabled=False → None (no warning, no exception) - enabled=True + missing jid/password/to/host → redacted warning log, None - enabled=True + complete config → SyncXmppChannel instance - Factory never raises exceptions (D2 non-blocking degradation) - redact_secrets with extra_secrets=[password, jid, to] on warning logs Re-exports Channel, XmppChannel, SyncXmppChannel from channels package. 19 unit tests covering disabled, misconfigured, complete, dry-run, and secret-safe warning log scenarios. Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid> Co-authored-by: opencode/coder <coder@agents.invalid>
304 lines
9.7 KiB
Python
304 lines
9.7 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
|
|
|
|
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)
|
|
|
|
|
|
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
|