AntoineVe/college-infos
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix(xmpp): corriger le timeout XMPP — TLS mode + timeouts de connexion/cleanup (#25) #26

Merged
AntoineVe merged 1 commits from fix/xmpp-tls-timeout-25 into main 2026-09-11 17:25:23 +02:00
Conversation 0 Commits 1 Files Changed 9 +1070 -105
9 changed files with 1070 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.env.example
View File

@@ -50,7 +50,11 @@ XMPP_HOST=example.com
XMPP_PORT=5222
XMPP_TO=parent@example.com
XMPP_RESOURCE=pronote-sync
XMPP_USE_TLS=true
# Mode TLS : "starttls" (port 5222, défaut), "direct" (port 5223) ou "disabled" (localhost uniquement)
XMPP_TLS_MODE=starttls
# XMPP_USE_TLS=true/false est obsolète mais encore supporté (alias de XMPP_TLS_MODE)
XMPP_CONNECT_TIMEOUT=15
XMPP_CLEANUP_TIMEOUT=10
XMPP_TIMEOUT=30
# --- IA (optionnelle) ---

16
.secrets.baseline
View File

@@ -139,9 +139,8 @@
"type": "Hex High Entropy String",
"filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": true,
"line_number": 5085,
"is_secret": false
"is_verified": false,
"line_number": 5103
}
],
"tests/unit/test_caldav_gateway.py": [
@@ -175,7 +174,16 @@
"is_verified": false,
"line_number": 28
}
],
"tests/unit/test_xmpp_channel.py": [
{
"type": "Secret Keyword",
"filename": "tests/unit/test_xmpp_channel.py",
"hashed_secret": "b01bcded8826562355b3e32a8b31e214ff533511",
"is_verified": false,
"line_number": 1011
}
]
},
"generated_at": "2026-09-11T09:57:08Z"
"generated_at": "2026-09-11T14:59:15Z"
}

34
GUIDE_DEV_PYTHON.md
View File

@@ -275,7 +275,8 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
> ⚠️ **Décision d'implémentation** :
> `XMPP_RECIPIENT` a été renommé en `XMPP_TO` dans l'implémentation (aligné avec §10.2.1).
> Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`.
> Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_TLS_MODE`, `XMPP_TIMEOUT`, `XMPP_CONNECT_TIMEOUT`, `XMPP_CLEANUP_TIMEOUT`.
> `XMPP_USE_TLS` reste supporté mais est **obsolète** (remplacé par `XMPP_TLS_MODE`).
> Une section `BLOG_ENABLED` et `BLOG_RSS_URL` a été ajoutée dans `.env.example`.
Les variables Pronote sont obligatoires selon les sources activées :
@@ -4066,7 +4067,9 @@ Si le besoin évolue (ex: **plusieurs destinataires**), les étapes suivantes so
| `XMPP_PORT` | Port XMPP (5222 pour STARTTLS, 5223 pour TLS direct). | `5222` | `int` | ❌ Non |
| `XMPP_TO` | Destinataire unique (ex: `parent@exemple.org`). | `None` | `str` | ✅ Oui |
| `XMPP_RESOURCE` | Ressource XMPP (ex: `pronote-sync`). | `"pronote-sync"` | `str` | ❌ Non |
| `XMPP_USE_TLS` | Utiliser TLS pour la connexion. | `True` | `bool` | ❌ Non |
| `XMPP_TLS_MODE` | Mode TLS : `starttls` (défaut), `direct` ou `disabled` (localhost uniquement). | `"starttls"` | `Literal` | ❌ Non |
| `XMPP_CONNECT_TIMEOUT` | Timeout de connexion (secondes). | `15` | `float` | ❌ Non |
| `XMPP_CLEANUP_TIMEOUT` | Timeout de nettoyage (secondes). | `10` | `float` | ❌ Non |
| `XMPP_TIMEOUT` | Timeout de connexion (secondes). | `30` | `int` | ❌ Non |
**⚠️ Notes** :
@@ -4074,6 +4077,7 @@ Si le besoin évolue (ex: **plusieurs destinataires**), les étapes suivantes so
- **Pas de variables PubSub** : `XMPP_PUBSUB_NODE`, `XMPP_ROOM`, `XMPP_SUBSCRIBERS` **ne doivent pas être introduites** pour l'instant.
- **Sécurité** : `XMPP_JID`, `XMPP_PASSWORD` et `XMPP_TO` **ne doivent jamais apparaître** dans les logs, erreurs ou fixtures.
- **Standardisation** : `XMPP_TO` est mappé sur le champ `to` dans le modèle Pydantic.
- **`XMPP_USE_TLS` obsolète** : remplacée par `XMPP_TLS_MODE` (`starttls`, `direct` ou `disabled`) ; `XMPP_USE_TLS=true` → `direct` et `XMPP_USE_TLS=false` → `starttls`, avec un `DeprecationWarning`.
#### 10.2.2 Exemple de configuration dans `.env`
@@ -4086,13 +4090,19 @@ XMPP_HOST=exemple.org
XMPP_PORT=5222
XMPP_TO=parent@exemple.org
XMPP_RESOURCE=pronote-digest
XMPP_USE_TLS=true
# XMPP_TLS_MODE : "starttls" (défaut), "direct" (port 5223) ou "disabled" (localhost uniquement)
XMPP_TLS_MODE=starttls
# XMPP_USE_TLS=true/false est obsolète mais encore supporté (alias de XMPP_TLS_MODE)
XMPP_CONNECT_TIMEOUT=15
XMPP_CLEANUP_TIMEOUT=10
XMPP_TIMEOUT=30
```
#### 10.2.3 Modèle Pydantic pour la configuration XMPP
```python
from typing import Literal
from pydantic import SecretStr, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -4118,14 +4128,22 @@ class XmppSettings(BaseSettings):
port: int = Field(default=5222, ge=1, le=65535)
to: str | None = None
resource: str = "pronote-sync"
use_tls: bool = True
timeout: int = Field(default=30, gt=0)
tls_mode: Literal["direct", "starttls", "disabled"] = "starttls"
use_tls: bool | None = Field(
default=None,
deprecated="Utiliser tls_mode à la place (XMPP_USE_TLS obsolète).",
)
timeout: float = Field(default=30, gt=0)
connect_timeout: float = Field(default=15, gt=0)
cleanup_timeout: float = Field(default=10, gt=0)
```
> **⚠️ Mapping TLS** :
> - `use_tls=True` → **TLS direct** (port 5223, `enable_direct_tls=True`, `enable_starttls=False`).
> - `use_tls=False` → **STARTTLS** (port 5222, `enable_starttls=True`, `enable_direct_tls=False`).
> La validation refuse `use_tls=False` si `host` n'est pas un hôte de boucle locale (`localhost`, `127.0.0.1`, `::1`).
> - `tls_mode="starttls"` (défaut) → **STARTTLS** (port 5222, `enable_starttls=True`, `enable_direct_tls=False`).
> - `tls_mode="direct"` → **TLS direct** (port 5223, `enable_direct_tls=True`, `enable_starttls=False`).
> - `tls_mode="disabled"` → TLS désactivé ; **uniquement** autorisé sur un hôte de boucle locale (`localhost`, `127.0.0.1`, `::1`).
> - `XMPP_USE_TLS` (`use_tls`) est **obsolète** : `use_tls=True` → `tls_mode="direct"` et
> `use_tls=False` → `tls_mode="starttls"` (avec `DeprecationWarning`).
---

3
README.LLM.md
View File

@@ -71,7 +71,8 @@ The following variables can be safely pre-configured in `/etc/pronote-sync/prono
- `THEORETICAL_WEEK_ANCHOR_DATE`, `THEORETICAL_WEEK_ANCHOR_TYPE`
- **XMPP:**
- `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`, `XMPP_RESOURCE`
- `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_TLS_MODE`, `XMPP_TIMEOUT`, `XMPP_CONNECT_TIMEOUT`, `XMPP_CLEANUP_TIMEOUT`, `XMPP_RESOURCE`
- `XMPP_USE_TLS` is deprecated but still supported (aliased to `XMPP_TLS_MODE`)
- **AI:**
- `AI_ENABLED`, `AI_PROVIDER`, `AI_BASE_URL`, `AI_MODEL`, `AI_ALLOW_INSECURE_HTTP`

206
pronote_sync/channels/xmpp.py
View File

@@ -36,6 +36,24 @@ logger = logging.getLogger(__name__)
__all__ = ["XmppChannel", "SyncXmppChannel", "XmppMessage"]
async def _cancel_pending(
*futures: asyncio.Future[bool],
) -> None:
"""Annule les futures/tâches encore en attente et supprime le bruit.
À appeler avant chaque retour anticipé de :meth:`XmppChannel.send_async`
afin qu'aucune tentative de connexion ne survive au retour de la méthode.
:param futures: Futures ou tâches à annuler (les déjà terminées sont
ignorées pour la cancellation mais attendues pour purger l'attente).
:rtype: None
"""
for future in futures:
if not future.done():
future.cancel()
await asyncio.gather(*futures, return_exceptions=True)
def _secret_values(settings: XmppSettings) -> tuple[SecretStr | str, ...]:
"""Rassemble les secrets du canal XMPP pour le masquage des logs.
@@ -220,11 +238,14 @@ class XmppChannel:
"""Exécute le flux asynchrone d'envoi XMPP (U2).
Connecte le client ``slixmpp`` avec un hôte et un port explicites,
configure TLS avant la connexion, puis attend l'un des événements
``session_start``, ``failed_auth`` ou ``disconnected`` sous un
timeout unique avant d'envoyer un message direct ``chat`` au
configure TLS avant la connexion selon ``tls_mode`` (``direct``,
``starttls`` ou ``disabled``), attend la connexion sous
``connect_timeout`` puis l'un des événements ``session_start``,
``failed_auth``, ``connection_failed`` ou ``disconnected`` sous
``timeout`` avant d'envoyer un message direct ``chat`` au
destinataire configuré. La déconnexion est garantie par un bloc
``try/finally``. Aucun secret n'est journalisé (SEC-XMPP-02).
``try/finally`` borné par ``cleanup_timeout``. Aucun secret n'est
journalisé (SEC-XMPP-02).
:param message: Message final à envoyer.
:return: ``True`` si l'envoi a réussi (ou a été simulé en dry-run),
@@ -244,55 +265,132 @@ class XmppChannel:
logger.warning("Destinataire XMPP manquant.")
return False
# Create typed client
client = ClientXMPP(
jid_str,
self.settings.password.get_secret_value() if self.settings.password else "",
)
# Configure TLS BEFORE connect
if self.settings.use_tls:
# TLS direct (port 5223 typically)
client.enable_direct_tls = True
client.enable_starttls = False
else:
# STARTTLS (port 5222 typically)
client.enable_starttls = True
client.enable_direct_tls = False
# Register handlers
# Client typed lazily: the construction is done inside the try block so that
# any error is caught and converted to ``False`` (channel contract: never raise)
client: ClientXMPP | None = None
session_future: asyncio.Future[bool] = asyncio.get_event_loop().create_future()
failure_kind = "disconnected"
# Declared before the ``try`` so the exception handlers (CancelledError and
# Exception) can cancel any task still pending from ``asyncio.wait()``
connect_future: asyncio.Future[bool] | None = None
session_task: asyncio.Future[bool] | None = None
def on_session_start(event: object) -> None:
if not session_future.done():
session_future.set_result(True)
def on_failed_auth(event: object) -> None:
nonlocal failure_kind
if not session_future.done():
failure_kind = "failed_auth"
session_future.set_result(False)
def on_connection_failed(event: object) -> None:
nonlocal failure_kind
if not session_future.done():
failure_kind = "connection_failed"
session_future.set_result(False)
def on_disconnected(event: object) -> None:
if not session_future.done():
session_future.set_result(False)
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)
try:
# Create typed client
client = ClientXMPP(
jid_str,
self.settings.password.get_secret_value() if self.settings.password else "",
)
# Configure TLS BEFORE connect (canonical tls_mode)
match self.settings.tls_mode:
case "direct":
client.enable_direct_tls = True
client.enable_starttls = False
case "starttls":
client.enable_direct_tls = False
client.enable_starttls = True
case "disabled":
client.enable_direct_tls = False
client.enable_starttls = False
# Register handlers
client.add_event_handler("session_start", on_session_start)
client.add_event_handler("failed_auth", on_failed_auth)
client.add_event_handler("connection_failed", on_connection_failed)
client.add_event_handler("disconnected", on_disconnected)
# Connect with explicit host and port
connect_future = client.connect(self.settings.host, self.settings.port)
await connect_future # connect() returns a Future, not a coroutine
connect_future = asyncio.ensure_future(
client.connect(self.settings.host, self.settings.port)
)
# Wait for one of the three events under a single timeout
try:
success = await asyncio.wait_for(session_future, timeout=self.settings.timeout)
except TimeoutError:
logger.warning("Délai d'attente de session XMPP dépassé.")
return False
# Wrap the session future in a task so that cancelling pending tasks
# during the concurrent wait never cancels ``session_future`` itself
async def _await_session() -> bool:
return await session_future
if not success:
logger.warning("Échec d'authentification ou déconnexion XMPP.")
session_task = asyncio.ensure_future(_await_session())
# Wait for the connection and the session event concurrently, bounded by
# connect_timeout as the global time limit: a ``connection_failed`` event
# can thus trigger an early return before the connect timeout expires
done, _pending = await asyncio.wait(
{connect_future, session_task},
timeout=self.settings.connect_timeout,
return_when=asyncio.FIRST_COMPLETED,
)
if session_task in done:
if not session_future.result():
# ``connection_failed``/``failed_auth``/``disconnected`` fired
# before the connection was resolved: immediate failure (fail fast)
if failure_kind == "connection_failed":
logger.warning("Échec de connexion réseau XMPP.")
else:
logger.warning("Échec d'authentification ou déconnexion XMPP.")
await _cancel_pending(connect_future, session_task)
return False
# ``session_start`` fired: the connection succeeded even if the
# connect future is still pending; proceed to send the message
await _cancel_pending(connect_future, session_task)
elif connect_future in done:
# The connection resolved: surface a connect error (redacted) if any
if not connect_future.cancelled():
connect_exc = connect_future.exception()
if connect_exc is not None and isinstance(connect_exc, Exception):
logger.warning(
"Échec de connexion XMPP : %s",
redact_exception(
connect_exc,
extra_secrets=_secret_values(self.settings),
),
)
await _cancel_pending(connect_future, session_task)
return False
# Connection established: wait for a session event under ``timeout``
try:
success = await asyncio.wait_for(
asyncio.shield(session_future), timeout=self.settings.timeout
)
except TimeoutError:
logger.warning("Délai d'attente de session XMPP dépassé.")
await _cancel_pending(connect_future, session_task)
return False
if not success:
if failure_kind == "connection_failed":
logger.warning("Échec de connexion réseau XMPP.")
else:
logger.warning("Échec d'authentification ou déconnexion XMPP.")
await _cancel_pending(connect_future, session_task)
return False
else:
# connect_timeout expired: cancel everything and fail fast
logger.warning(
"Délai de connexion XMPP dépassé (%ss).", self.settings.connect_timeout
)
await _cancel_pending(connect_future, session_task)
return False
# Send the message
@@ -300,19 +398,39 @@ class XmppChannel:
client.send_message(mto=JID(self.settings.to), mbody=formatted, mtype="chat")
return True
except asyncio.CancelledError:
# Contrat du canal : toujours retourner un booléen, même en cas
# d'annulation de la tâche appelante (cleanup exécuté par le finally).
logger.debug("Envoi XMPP annulé avant la fin de l'opération.")
pending = [f for f in (connect_future, session_task) if f is not None]
if pending:
await _cancel_pending(*pending)
return False
except Exception as exc:
redacted = redact_exception(exc)
redacted = redact_exception(exc, extra_secrets=_secret_values(self.settings))
extra = _secret_values(self.settings)
logger.warning("Erreur XMPP: %s", redact_secrets(redacted, extra_secrets=extra))
pending = [f for f in (connect_future, session_task) if f is not None]
if pending:
await _cancel_pending(*pending)
return False
finally:
try:
disconnect_future = client.disconnect()
await disconnect_future
except Exception as cleanup_exc:
logger.debug(
"Erreur lors de la déconnexion XMPP: %s", redact_exception(cleanup_exc)
)
if client is not None:
try:
disconnect_future = client.disconnect()
await asyncio.wait_for(disconnect_future, timeout=self.settings.cleanup_timeout)
except asyncio.CancelledError:
logger.debug("Déconnexion XMPP annulée.")
except TimeoutError:
logger.debug(
"Délai de déconnexion XMPP dépassé (%ss), abandon.",
self.settings.cleanup_timeout,
)
except Exception as cleanup_exc:
logger.debug(
"Erreur lors de la déconnexion XMPP : %s",
redact_exception(cleanup_exc, extra_secrets=_secret_values(self.settings)),
)
class SyncXmppChannel:
@@ -363,7 +481,7 @@ class SyncXmppChannel:
try:
return asyncio.run(self._channel.send_async(message))
except Exception as exc:
redacted = redact_exception(exc)
redacted = redact_exception(exc, extra_secrets=_secret_values(self.settings))
redacted = redact_secrets(redacted, extra_secrets=_secret_values(self.settings))
logger.warning("XMPP : erreur lors de l'envoi synchrone : %s", redacted)
return False

84
pronote_sync/config/settings.py
View File

@@ -8,6 +8,7 @@ depuis les variables d'environnement (préfixées par groupe) et le fichier
from __future__ import annotations
import warnings
from datetime import date
from typing import Literal
from urllib.parse import urlparse
@@ -18,6 +19,7 @@ from pydantic import (
ValidationInfo,
field_serializer,
field_validator,
model_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -158,10 +160,16 @@ class XmppSettings(BaseSettings):
Contraintes de champs : ``port`` est borné entre 1 et 65535 et ``timeout``
doit être strictement positif.
Politique TLS : la désactivation de TLS (``use_tls`` à ``False``) n'est
autorisée que sur un hôte de boucle locale (``localhost``, ``127.0.0.1``,
``::1``). Dans tout autre cas, une erreur de validation est levée,
indépendamment de l'état du champ ``enabled``.
Politique TLS : le mode ``tls_mode`` détermine la négociation TLS
(``direct``, ``starttls`` ou ``disabled``). Le mode ``disabled`` n'est
autorisé que sur un hôte de boucle locale (``localhost``, ``127.0.0.1``,
``::1``) ; ``starttls`` et ``direct`` sont permis pour tous les hôtes.
Compatibilité : le champ historique ``use_tls`` (booléen) est un alias
obsolète ; ``use_tls=True`` mappe vers ``tls_mode="direct"`` et
``use_tls=False`` vers ``tls_mode="starttls"``, avec un
:pyexc:`DeprecationWarning`. La valeur brute fournie reste lisible via
``use_tls`` (``None`` si non fournie).
"""
model_config = SettingsConfigDict(
@@ -178,33 +186,61 @@ class XmppSettings(BaseSettings):
port: int = Field(default=5222, ge=1, le=65535)
to: str | None = None
resource: str = "pronote-sync"
use_tls: bool = True
timeout: int = Field(default=30, gt=0)
tls_mode: Literal["direct", "starttls", "disabled"] = "starttls"
use_tls: bool | None = Field(
default=None,
deprecated="Utiliser tls_mode à la place (XMPP_USE_TLS obsolète).",
)
timeout: float = Field(default=30, gt=0)
connect_timeout: float = Field(default=15, gt=0)
cleanup_timeout: float = Field(default=10, gt=0)
@field_validator("use_tls")
@model_validator(mode="before")
@classmethod
def _validate_tls_policy(cls, v: bool, info: ValidationInfo) -> bool:
"""Refuse la désactivation de TLS hors des hôtes de boucle locale.
def _migrate_use_tls(cls, data: object) -> object:
"""Mappe l'alias obsolète ``use_tls`` vers le mode canonique ``tls_mode``.
La règle s'applique quel que soit l'état du champ ``enabled``. Le
message d'erreur ne contient aucune valeur sensible (``jid``,
``use_tls=True`` devient ``tls_mode="direct"`` et ``use_tls=False``
devient ``tls_mode="starttls"`` ; un :pyexc:`DeprecationWarning` est
émis à chaque usage explicite de l'alias. ``tls_mode`` fourni
explicitement prend le pas sur l'alias.
:param data: Données d'entrée du modèle (dict ou autre).
:return: Données d'entrée avec ``tls_mode`` dérivé de ``use_tls``.
:rtype: object
"""
if not isinstance(data, dict) or "use_tls" not in data:
return data
warnings.warn(
"XMPP_USE_TLS est obsolète : utiliser XMPP_TLS_MODE "
"('direct', 'starttls' ou 'disabled').",
DeprecationWarning,
stacklevel=2,
)
if data.get("tls_mode") is None:
data["tls_mode"] = "direct" if data["use_tls"] else "starttls"
return data
@model_validator(mode="after")
def _validate_tls_policy(self) -> XmppSettings:
"""Refuse le mode ``disabled`` hors des hôtes de boucle locale.
La règle s'applique quel que soit l'état du champ ``enabled``. Les
modes ``starttls`` et ``direct`` sont autorisés pour tous les hôtes.
Le message d'erreur ne contient aucune valeur sensible (``jid``,
``password``, ``to``).
:param v: Valeur du champ ``use_tls`` à valider.
:param info: Contexte de validation (accès aux autres champs).
:return: La valeur validée inchangée.
:rtype: bool
:raises ValueError: Si ``use_tls`` est ``False`` et que ``host``
:return: L'instance validée inchangée.
:rtype: XmppSettings
:raises ValueError: Si ``tls_mode`` est ``disabled`` et que ``host``
n'est pas un hôte de boucle locale.
"""
if v is False:
host = info.data.get("host", "")
if host not in _XMPP_LOOPBACK_HOSTS:
raise ValueError(
"TLS désactivé n'est autorisé que sur les hôtes de loopback "
"(localhost, 127.0.0.1, ::1)."
) from None
return v
if self.tls_mode == "disabled" and self.host not in _XMPP_LOOPBACK_HOSTS:
raise ValueError(
"TLS désactivé n'est autorisé que sur les hôtes de loopback "
"(localhost, 127.0.0.1, ::1)."
) from None
return self
class AISettings(BaseSettings):

2
tests/integration/test_xmpp_integration.py
View File

@@ -67,7 +67,7 @@ class FakeClientXMPP:
:param handler: Fonction gestionnaire.
:raises: AssertionError si l'événement n'est pas supporté.
"""
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(handler)

694
tests/unit/test_xmpp_channel.py
View File

@@ -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).

130
tests/unit/test_xmpp_settings.py
View File

@@ -99,19 +99,52 @@ class TestTimeoutConstraints:
class TestTlsPolicy:
"""Tests de la politique TLS pour le champ use_tls."""
def test_use_tls_false_with_remote_host_rejected(self) -> None:
"""Vérifie que use_tls=False avec un hôte distant est rejeté.
def test_tls_disabled_with_remote_host_rejected(self) -> None:
"""Vérifie que tls_mode="disabled" avec un hôte distant est rejeté.
:raises ValidationError: Si use_tls=False et host n'est pas une boucle locale.
:raises ValidationError: Si tls_mode="disabled" et host n'est pas une boucle locale.
"""
with pytest.raises(ValidationError) as exc_info:
XmppSettings(use_tls=False, host="talk.example.com")
assert "use_tls" in str(exc_info.value).lower() or "tls" in str(exc_info.value).lower()
XmppSettings(tls_mode="disabled", host="talk.example.com")
assert "tls" in str(exc_info.value).lower()
def test_use_tls_false_with_remote_host_rejected_when_enabled(self) -> None:
def test_tls_disabled_with_remote_host_rejected_when_enabled(self) -> None:
"""Vérifie que la politique TLS s'applique même quand le canal est activé."""
with pytest.raises(ValidationError):
XmppSettings(enabled=True, use_tls=False, host="talk.example.com")
XmppSettings(enabled=True, tls_mode="disabled", host="talk.example.com")
def test_use_tls_false_maps_to_starttls(self) -> None:
"""Vérifie que l'alias obsolète use_tls=False mappe vers tls_mode='starttls'.
:return: Vérifie le mapping et l'émission d'un DeprecationWarning.
:rtype: None
"""
with pytest.warns(DeprecationWarning):
settings = XmppSettings(use_tls=False, host="talk.example.com")
assert settings.tls_mode == "starttls"
assert settings.host == "talk.example.com"
def test_use_tls_true_maps_to_direct(self) -> None:
"""Vérifie que l'alias obsolète use_tls=True mappe vers tls_mode='direct'.
:return: Vérifie le mapping et l'émission d'un DeprecationWarning.
:rtype: None
"""
with pytest.warns(DeprecationWarning):
settings = XmppSettings(use_tls=True, host="talk.example.com")
assert settings.tls_mode == "direct"
assert settings.host == "talk.example.com"
def test_default_tls_mode_is_starttls(self) -> None:
"""Vérifie que tls_mode vaut 'starttls' par défaut avec les timeouts par défaut.
:return: Vérifie tls_mode='starttls', connect_timeout=15, cleanup_timeout=10.
:rtype: None
"""
settings = XmppSettings()
assert settings.tls_mode == "starttls"
assert settings.connect_timeout == 15
assert settings.cleanup_timeout == 10
def test_use_tls_false_with_localhost_allowed(self) -> None:
"""Vérifie que use_tls=False avec localhost est autorisé.
@@ -120,7 +153,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=False, host="localhost")
assert settings.use_tls is False
assert settings.tls_mode == "starttls"
assert settings.host == "localhost"
def test_use_tls_false_with_127_allowed(self) -> None:
@@ -130,7 +163,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=False, host="127.0.0.1")
assert settings.use_tls is False
assert settings.tls_mode == "starttls"
assert settings.host == "127.0.0.1"
def test_use_tls_false_with_ipv6_loopback_allowed(self) -> None:
@@ -140,7 +173,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=False, host="::1")
assert settings.use_tls is False
assert settings.tls_mode == "starttls"
assert settings.host == "::1"
def test_use_tls_true_with_remote_host_allowed(self) -> None:
@@ -150,7 +183,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=True, host="talk.example.com")
assert settings.use_tls is True
assert settings.tls_mode == "direct"
assert settings.host == "talk.example.com"
def test_use_tls_true_with_empty_host_allowed(self) -> None:
@@ -160,7 +193,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=True, host="")
assert settings.use_tls is True
assert settings.tls_mode == "direct"
assert settings.host == ""
@@ -173,7 +206,7 @@ class TestNoSecretInErrorMessages:
Crée une instance avec des valeurs sensibles et vérifie que l'erreur de validation
ne contient pas ces valeurs dans son message.
:raises ValidationError: Si use_tls=False avec un hôte non-local.
:raises ValidationError: Si tls_mode="disabled" avec un hôte non-local.
:return: Vérifie que le message d'erreur ne contient pas les secrets.
:rtype: None
"""
@@ -184,7 +217,7 @@ class TestNoSecretInErrorMessages:
with pytest.raises(ValidationError) as exc_info:
XmppSettings(
use_tls=False,
tls_mode="disabled",
host="talk.example.com",
jid=sentinel_jid,
password=sentinel_password,
@@ -197,3 +230,72 @@ class TestNoSecretInErrorMessages:
assert "test_password_123" not in error_message
assert "test_to@example.com" not in error_message
assert "secret" not in error_message
class TestXmppSettingsTlsMode:
"""Tests supplémentaires pour les validateurs de XmppSettings."""
def test_use_tls_true_maps_to_direct_with_warning(self) -> None:
"""Vérifie que use_tls=True mappe vers tls_mode='direct' avec DeprecationWarning.
:return: Vérifie le mapping et l'émission d'un DeprecationWarning.
:rtype: None
"""
with pytest.warns(DeprecationWarning):
settings = XmppSettings(use_tls=True, host="talk.example.com")
assert settings.tls_mode == "direct"
def test_use_tls_false_maps_to_starttls_with_warning(self) -> None:
"""Vérifie que use_tls=False mappe vers tls_mode='starttls' avec DeprecationWarning.
:return: Vérifie le mapping et l'émission d'un DeprecationWarning.
:rtype: None
"""
with pytest.warns(DeprecationWarning):
settings = XmppSettings(use_tls=False, host="talk.example.com")
assert settings.tls_mode == "starttls"
def test_tls_mode_disabled_rejected_for_remote(self) -> None:
"""Vérifie que tls_mode='disabled' avec un hôte distant est rejeté.
:raises ValidationError: Si tls_mode="disabled" et host n'est pas une boucle locale.
"""
with pytest.raises(ValidationError) as exc_info:
XmppSettings(tls_mode="disabled", host="example.com")
assert "tls" in str(exc_info.value).lower()
def test_tls_mode_disabled_allowed_for_loopback(self) -> None:
"""Vérifie que tls_mode='disabled' avec host='127.0.0.1' est valide.
:return: Vérifie que XmppSettings(tls_mode='disabled', host='127.0.0.1') est valide.
:rtype: None
"""
settings = XmppSettings(tls_mode="disabled", host="127.0.0.1")
assert settings.tls_mode == "disabled"
assert settings.host == "127.0.0.1"
def test_connect_timeout_must_be_positive(self) -> None:
"""Vérifie que connect_timeout doit être strictement positif.
:raises ValidationError: Si connect_timeout <= 0.
"""
with pytest.raises(ValidationError) as exc_info:
XmppSettings(connect_timeout=0)
assert "connect_timeout" in str(exc_info.value).lower()
with pytest.raises(ValidationError) as exc_info:
XmppSettings(connect_timeout=-1)
assert "connect_timeout" in str(exc_info.value).lower()
def test_cleanup_timeout_must_be_positive(self) -> None:
"""Vérifie que cleanup_timeout doit être strictement positif.
:raises ValidationError: Si cleanup_timeout <= 0.
"""
with pytest.raises(ValidationError) as exc_info:
XmppSettings(cleanup_timeout=0)
assert "cleanup_timeout" in str(exc_info.value).lower()
with pytest.raises(ValidationError) as exc_info:
XmppSettings(cleanup_timeout=-1)
assert "cleanup_timeout" in str(exc_info.value).lower()