Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
0363898669
|
13
.env.example
13
.env.example
@@ -11,6 +11,19 @@ PRONOTE_AGENDA_SOURCE=auto
|
||||
PRONOTE_HOMEWORK_SOURCE=auto
|
||||
PRONOTE_MESSAGES_SOURCE=pronotepy
|
||||
|
||||
# Mode d'authentification Pronote
|
||||
# "password" (défaut) : authentification classique URL + identifiant + mot de passe
|
||||
# "qr_token" : authentification par QR code puis token persistant
|
||||
PRONOTE_AUTH_MODE=password
|
||||
|
||||
# Fichier JSON du QR code Pronote (enrôlement initial, mode qr_token uniquement)
|
||||
# À générer depuis l'application mobile Pronote. Le QR code expire ~10 minutes.
|
||||
# PRONOTE_QR_CODE_FILE=/path/to/qr_code.json
|
||||
|
||||
# PIN à 4 chiffres pour l'enrôlement QR code (mode qr_token uniquement)
|
||||
# SENSIBLE : ne jamais committer cette valeur
|
||||
# PRONOTE_QR_PIN=1234
|
||||
|
||||
# --- CalDAV ---
|
||||
CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
|
||||
CALDAV_USERNAME=user@example.com
|
||||
|
||||
2
.gitignore
vendored
2
.gitignore
vendored
@@ -48,6 +48,8 @@ Thumbs.db
|
||||
# --- Project-specific state files ---
|
||||
.blog_rss_state.json
|
||||
.caldav_sync_state.json
|
||||
# État d'authentification pronotepy (QR code / token rotation)
|
||||
.pronote_auth_state.json
|
||||
*.state.json
|
||||
|
||||
# --- Local scratch / WIP files ---
|
||||
|
||||
25
AGENTS.md
25
AGENTS.md
@@ -146,6 +146,31 @@ pronote-sync --dry-run
|
||||
- Réutiliser un téléchargement/parsing iCal pour l'agenda et les devoirs pendant un même run, sans
|
||||
cache global ni persistant.
|
||||
|
||||
### Contrat d'authentification QR code / token
|
||||
|
||||
- Le mode d'authentification est sélectionné par `PRONOTE_AUTH_MODE` :
|
||||
- `password` (défaut) : authentification classique via URL, identifiant, mot de passe et ENT.
|
||||
- `qr_token` : authentification par QR code puis token persistant (pour les instances Pronote
|
||||
utilisant HubEduConnect/EduConnect où l'authentification par mot de passe échoue).
|
||||
- En mode `qr_token`, le premier login utilise `pronotepy.qrcode_login(qr_code, pin, uuid)` avec
|
||||
les paramètres `PRONOTE_QR_CODE_FILE` (chemin du JSON QR) et `PRONOTE_QR_PIN` (PIN SecretStr).
|
||||
- Après chaque login réussi, les credentials exportées par `pronotepy.export_credentials()` sont
|
||||
persistées dans `.pronote_auth_state.json` (permissions `0600`, format JSON versionné, écriture
|
||||
atomique). Le token rotate à chaque session — le fichier doit être mis à jour après chaque run.
|
||||
- Les logins suivants utilisent `pronotepy.token_login(**credentials)` avec le token persisté.
|
||||
- En cas d'échec de `token_login` (token expiré/invalide), une `PronoteAuthRotationError` est levée.
|
||||
Cette erreur se propage sans wrapping à travers `PronoteFetcher` et `fetch_step` jusqu'à
|
||||
`PipelineRunner.run()`, qui :
|
||||
- journalise l'erreur (expurgée) ;
|
||||
- envoie une notification XMPP actionnable si le canal est disponible et `dry_run` est inactif ;
|
||||
- retourne un résultat dégradé `(None, errors)`.
|
||||
- `PronoteAuthRotationError` est re-levée telle quelle (`except PronoteAuthRotationError: raise`)
|
||||
dans toutes les couches d'enveloppement du chemin critique (fetch_agenda, fetch_homework,
|
||||
fetch_step). Ne pas l'attraper avec `except Exception` sans la re-léver d'abord.
|
||||
- Le fichier `.pronote_auth_state.json` ne doit jamais être committé (couvert par `.gitignore`).
|
||||
Son contenu (token vivant) ne doit jamais apparaître dans les logs, les messages d'erreur ou
|
||||
les notifications XMPP.
|
||||
|
||||
### Contrat du provider `openai-compatible`
|
||||
- Le provider `openai-compatible` réutilise `OpenAISynthesisProvider` avec un `base_url` personnalisé ; aucun nouveau provider n'est créé.
|
||||
- `AI_BASE_URL` et `AI_MODEL` sont requis ; `AI_API_KEY` est requis (MVP).
|
||||
|
||||
@@ -42,6 +42,9 @@ class PronoteSettings(BaseSettings):
|
||||
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
||||
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
||||
messages_source: Literal["pronotepy"] = "pronotepy"
|
||||
auth_mode: Literal["password", "qr_token"] = "password"
|
||||
qr_code_file: str | None = None
|
||||
qr_pin: SecretStr | None = None
|
||||
|
||||
@field_serializer("ical_url")
|
||||
def _serialize_ical_url(self, value: SecretStr | None) -> str | None:
|
||||
@@ -55,6 +58,18 @@ class PronoteSettings(BaseSettings):
|
||||
return None
|
||||
return "**********"
|
||||
|
||||
@field_serializer("qr_pin")
|
||||
def _serialize_qr_pin(self, value: SecretStr | None) -> str | None:
|
||||
"""Masque le code PIN QR lors de la sérialisation (repr, str, JSON).
|
||||
|
||||
:param value: Valeur du champ ``qr_pin``.
|
||||
:return: ``"**********"`` si la valeur est définie, ``None`` sinon.
|
||||
:rtype: str | None
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
return "**********"
|
||||
|
||||
|
||||
class CalDAVSettings(BaseSettings):
|
||||
"""Paramètres d'accès au serveur CalDAV de destination.
|
||||
@@ -266,8 +281,9 @@ class Settings(BaseSettings):
|
||||
"""Énumère tous les secrets configurés pour la rédaction.
|
||||
|
||||
Collecte les valeurs :class:`pydantic.SecretStr` non vides présentes
|
||||
dans les sous-configurations (Pronote, CalDAV, XMPP, IA). Les valeurs
|
||||
vides ou ``None`` sont filtrées ; les doublons sont supprimés.
|
||||
dans les sous-configurations (URL iCal, mots de passe, code PIN QR et
|
||||
clé API IA). Les valeurs vides ou ``None`` sont filtrées ; les
|
||||
doublons sont supprimés.
|
||||
|
||||
:return: Tuple de secrets à masquer dans les messages d'erreur.
|
||||
:rtype: tuple[SecretStr, ...]
|
||||
@@ -275,6 +291,7 @@ class Settings(BaseSettings):
|
||||
secrets = [
|
||||
self.pronote.ical_url,
|
||||
self.pronote.password,
|
||||
self.pronote.qr_pin,
|
||||
self.caldav.url,
|
||||
self.caldav.password,
|
||||
self.xmpp.password,
|
||||
|
||||
@@ -21,6 +21,23 @@ class PronoteSyncError(Exception):
|
||||
self.message = message
|
||||
|
||||
|
||||
class PronoteAuthRotationError(PronoteSyncError):
|
||||
"""Erreur de rotation du token d'authentification pronotepy (QR code / token).
|
||||
|
||||
Levée quand le token persisté est invalide ou expiré et qu'un ré-enrôlement
|
||||
manuel (suppression du fichier d'état + nouveau QR code) est nécessaire.
|
||||
|
||||
:ivar message: Message décrivant l'action à effectuer, sans secret.
|
||||
"""
|
||||
|
||||
def __init__(self, message: str) -> None:
|
||||
"""Initialise l'erreur de rotation.
|
||||
|
||||
:param message: Message actionnable sans secret (PIN, token, URL).
|
||||
"""
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
class ErrorSeverity(StrEnum):
|
||||
"""Niveau de gravité d'une erreur produite par le pipeline."""
|
||||
|
||||
|
||||
@@ -11,7 +11,12 @@ from typing import Protocol, runtime_checkable
|
||||
from pronote_sync.channels import get_channel
|
||||
from pronote_sync.channels.protocol import Channel
|
||||
from pronote_sync.config.settings import Settings
|
||||
from pronote_sync.errors import PipelineCriticalError, PipelineError, PipelineWarning
|
||||
from pronote_sync.errors import (
|
||||
PipelineCriticalError,
|
||||
PipelineError,
|
||||
PipelineWarning,
|
||||
PronoteAuthRotationError,
|
||||
)
|
||||
from pronote_sync.models.blog import ExternalInfo
|
||||
from pronote_sync.models.pronote import PronoteData
|
||||
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
|
||||
@@ -26,6 +31,7 @@ from pronote_sync.pipeline.steps.send import send_step
|
||||
from pronote_sync.pipeline.steps.synthesis import synthesis_step
|
||||
from pronote_sync.sources.blog.rss import BlogRSSClient
|
||||
from pronote_sync.sources.blog.state import BlogRSSState
|
||||
from pronote_sync.sources.pronote.auth_state import PronoteAuthState
|
||||
from pronote_sync.sources.pronote.client import PronoteClient
|
||||
from pronote_sync.sources.pronote.fallback import PronoteFetcher, PronoteFetcherProtocol
|
||||
from pronote_sync.sources.theoretical import get_theoretical_provider
|
||||
@@ -133,7 +139,15 @@ class PipelineRunner:
|
||||
blog_state = BlogRSSState() if settings.blog.enabled else None
|
||||
return cls(
|
||||
settings=settings,
|
||||
pronote_fetcher=PronoteFetcher(settings, PronoteClient(settings.pronote)),
|
||||
pronote_fetcher=PronoteFetcher(
|
||||
settings,
|
||||
PronoteClient(
|
||||
settings.pronote,
|
||||
auth_state=(
|
||||
PronoteAuthState() if settings.pronote.auth_mode == "qr_token" else None
|
||||
),
|
||||
),
|
||||
),
|
||||
agenda_comparator=comparator,
|
||||
synthesis_provider=get_synthesis_provider(settings.ai),
|
||||
channel=get_channel(settings.xmpp, dry_run=effective_dry_run),
|
||||
@@ -190,6 +204,11 @@ class PipelineRunner:
|
||||
des étapes facultatives sont converties en :class:`PipelineWarning` afin
|
||||
que les étapes suivantes, notamment XMPP, restent exécutées.
|
||||
|
||||
Une :class:`PronoteAuthRotationError` interrompt également l'exécution :
|
||||
l'erreur est journalisée expurgée, une notification XMPP actionnable est
|
||||
envoyée (sauf en dry-run ou sans canal), puis un résultat dégradé est
|
||||
retourné.
|
||||
|
||||
:return: Données Pronote normalisées ou ``None``, puis erreurs et avertissements.
|
||||
:rtype: tuple[PronoteData | None, list[PipelineError]]
|
||||
"""
|
||||
@@ -271,6 +290,28 @@ class PipelineRunner:
|
||||
except Exception as exc:
|
||||
self._warn("send", self._redact(exc))
|
||||
return data, [*self._errors, *self._warnings]
|
||||
except PronoteAuthRotationError as exc:
|
||||
error = PipelineCriticalError(self._redact(exc), step="pronote")
|
||||
logger.error("Erreur critique du pipeline : %s", error.message)
|
||||
if self._channel is not None and not self._dry_run:
|
||||
message = XmppMessage(
|
||||
target_date=now.date(),
|
||||
synthesis=(
|
||||
"⚠️ Rotation du token Pronote échouée. Le token d'authentification est "
|
||||
"expiré ou invalide. Action requise : supprimez le fichier "
|
||||
".pronote_auth_state.json et relancez le pipeline avec un nouveau QR "
|
||||
"code (PRONOTE_QR_CODE_FILE + PRONOTE_QR_PIN)."
|
||||
),
|
||||
external_info=None,
|
||||
)
|
||||
try:
|
||||
if not send_step(self._channel, message):
|
||||
self._warn("send", "Le canal XMPP a refusé l'envoi")
|
||||
except Exception as send_exc:
|
||||
# L'envoi de la notification est un dernier avertissement : son échec
|
||||
# ne doit pas masquer l'erreur de rotation, déjà critique.
|
||||
self._warn("send", self._redact(send_exc))
|
||||
self._errors.append(error)
|
||||
except PipelineCriticalError as exc:
|
||||
logger.error("Erreur critique du pipeline : %s", exc.message)
|
||||
self._errors.append(exc)
|
||||
|
||||
@@ -5,7 +5,7 @@ from __future__ import annotations
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
|
||||
from pronote_sync.errors import PipelineCriticalError, PipelineWarning
|
||||
from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError
|
||||
from pronote_sync.models.agenda import Lesson, SchoolEvent
|
||||
from pronote_sync.models.homework import Homework
|
||||
from pronote_sync.models.message import Message
|
||||
@@ -101,6 +101,8 @@ def fetch_step(
|
||||
:return: Données récupérées et avertissements non critiques.
|
||||
:rtype: tuple[FetchedPronoteData, list[PipelineWarning]]
|
||||
:raises PipelineCriticalError: Si l'agenda ou les devoirs ne sont pas disponibles.
|
||||
:raises PronoteAuthRotationError: Si une rotation du token d'authentification
|
||||
pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline.
|
||||
"""
|
||||
try:
|
||||
lessons, school_events = fetcher.fetch_agenda()
|
||||
@@ -108,6 +110,8 @@ def fetch_step(
|
||||
homeworks = fetcher.fetch_homework(target_date)
|
||||
except PipelineCriticalError:
|
||||
raise
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
raise PipelineCriticalError(
|
||||
f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
|
||||
|
||||
195
pronote_sync/sources/pronote/auth_state.py
Normal file
195
pronote_sync/sources/pronote/auth_state.py
Normal file
@@ -0,0 +1,195 @@
|
||||
"""Persistance des credentials d'authentification par token pronotepy.
|
||||
|
||||
Ce module fournit :class:`PronoteAuthState`, qui stocke et charge les credentials
|
||||
d'authentification par QR code / token entre les exécutions du pipeline. Le token
|
||||
pronotepy rotate à chaque session : le fichier d'état doit être mis à jour après
|
||||
chaque login réussi via :meth:`PronoteAuthState.save`.
|
||||
|
||||
Le fichier d'état est créé avec des permissions ``0600`` car il contient un token
|
||||
d'authentification vivant. Son contenu n'est jamais journalisé.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from pronote_sync.errors import PronoteSyncError
|
||||
from pronote_sync.utils.redaction import redact_exception, redact_secrets
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_STATE_VERSION = 1
|
||||
|
||||
|
||||
class PronoteAuthState:
|
||||
"""Persiste les credentials d'authentification par token pronotepy entre
|
||||
les exécutions du pipeline.
|
||||
|
||||
Le fichier d'état contient un dict au format :
|
||||
{"version": 1, "credentials": {"pronote_url": "...", "username": "...", "password": "<token>", "uuid": "..."}}
|
||||
|
||||
Les credentials sont le retour de pronotepy.Client.export_credentials(), utilisé tel quel
|
||||
pour token_login(**credentials). Le token rotate à chaque session — le fichier doit être
|
||||
mis à jour après chaque login réussi.
|
||||
|
||||
:param state_file: Chemin du fichier d'état JSON (``str`` ou
|
||||
:class:`~pathlib.Path`). ``".pronote_auth_state.json"`` par défaut.
|
||||
"""
|
||||
|
||||
def __init__(self, state_file: Path | str = ".pronote_auth_state.json") -> None:
|
||||
"""Initialise le gestionnaire d'état d'authentification Pronote.
|
||||
|
||||
Le fichier d'état n'est pas créé à l'initialisation : il n'est écrit
|
||||
qu'à la première sauvegarde réussie via :meth:`save`.
|
||||
|
||||
:param state_file: Chemin du fichier d'état JSON (``str`` ou
|
||||
:class:`~pathlib.Path`). ``".pronote_auth_state.json"`` par défaut.
|
||||
"""
|
||||
self._state_file = Path(state_file)
|
||||
|
||||
def load(self) -> dict[str, str] | None:
|
||||
"""Charge les credentials d'authentification depuis le fichier d'état.
|
||||
|
||||
Un fichier absent renvoie ``None`` (journalisé en debug). Un fichier
|
||||
corrompu, une version absente ou non supportée, ou un champ
|
||||
``credentials`` invalide renvoient ``None`` avec un avertissement.
|
||||
Le contenu des credentials n'est jamais journalisé.
|
||||
|
||||
:return: Dict des credentials (``pronote_url``, ``username``,
|
||||
``password``, ``uuid``) prêt pour
|
||||
``pronotepy.Client.token_login(**credentials)``, ou ``None`` si
|
||||
aucun état valide n'est disponible.
|
||||
:rtype: dict[str, str] | None
|
||||
"""
|
||||
if not self._state_file.exists():
|
||||
logger.debug(
|
||||
"Fichier d'état d'authentification Pronote %s absent, aucun token à charger.",
|
||||
redact_secrets(str(self._state_file)),
|
||||
)
|
||||
return None
|
||||
try:
|
||||
data: Any = json.loads(self._state_file.read_text(encoding="utf-8"))
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"Impossible de charger le fichier d'état d'authentification Pronote %s : %s, "
|
||||
"aucun token chargé.",
|
||||
redact_secrets(str(self._state_file)),
|
||||
redact_exception(exc),
|
||||
)
|
||||
return None
|
||||
if not isinstance(data, dict) or data.get("version") != _STATE_VERSION:
|
||||
logger.warning(
|
||||
"Fichier d'état d'authentification Pronote %s : version absente ou non supportée, "
|
||||
"aucun token chargé.",
|
||||
redact_secrets(str(self._state_file)),
|
||||
)
|
||||
return None
|
||||
credentials_data = data.get("credentials")
|
||||
if not isinstance(credentials_data, dict):
|
||||
logger.warning(
|
||||
"Fichier d'état d'authentification Pronote %s : champ credentials absent ou invalide, "
|
||||
"aucun token chargé.",
|
||||
redact_secrets(str(self._state_file)),
|
||||
)
|
||||
return None
|
||||
credentials: dict[str, str] = {}
|
||||
for key, value in credentials_data.items():
|
||||
if not isinstance(key, str) or not isinstance(value, str):
|
||||
logger.warning(
|
||||
"Fichier d'état d'authentification Pronote %s : champ credentials invalide, "
|
||||
"aucun token chargé.",
|
||||
redact_secrets(str(self._state_file)),
|
||||
)
|
||||
return None
|
||||
credentials[key] = value
|
||||
return credentials
|
||||
|
||||
def save(self, credentials: dict[str, str]) -> None:
|
||||
"""Sauvegarde les credentials dans le fichier d'état, de manière atomique.
|
||||
|
||||
Le fichier contient ``{"version": 1, "credentials": ...}``. Le JSON est
|
||||
d'abord écrit dans un fichier temporaire du même répertoire, créé avec
|
||||
les permissions ``0600`` (lecture seule pour le propriétaire) dès son
|
||||
ouverture via :func:`os.open` (avec ``O_EXCL`` et ``O_NOFOLLOW`` pour
|
||||
résister aux attaques par lien symbolique), puis verrouillé via
|
||||
:func:`os.fchmod` avant toute écriture ; le fichier temporaire remplace
|
||||
ensuite atomiquement le fichier d'état via :func:`os.replace`. Un
|
||||
éventuel fichier temporaire stale d'une exécution interrompue est
|
||||
supprimé avant l'ouverture. Les credentials ne sont jamais journalisés.
|
||||
|
||||
:param credentials: Dict des credentials pronotepy, tel que retourné
|
||||
par ``pronotepy.Client.export_credentials()``.
|
||||
:raises PronoteSyncError: Si l'écriture ou le remplacement du fichier
|
||||
échoue.
|
||||
"""
|
||||
payload: dict[str, Any] = {
|
||||
"version": _STATE_VERSION,
|
||||
"credentials": credentials,
|
||||
}
|
||||
tmp_file = self._state_file.with_suffix(".tmp")
|
||||
fd: int | None = None
|
||||
try:
|
||||
# Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue.
|
||||
if tmp_file.exists():
|
||||
try:
|
||||
tmp_file.unlink()
|
||||
except OSError:
|
||||
logger.debug(
|
||||
"Impossible de supprimer le fichier temporaire stale %s, "
|
||||
"l'ouverture en O_EXCL échouera.",
|
||||
redact_secrets(str(tmp_file)),
|
||||
)
|
||||
# O_EXCL empêche de créer par-dessus un fichier existant (attaque par lien
|
||||
# symbolique) et O_NOFOLLOW refuse de suivre un lien symbolique.
|
||||
fd = os.open(
|
||||
str(tmp_file),
|
||||
os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
|
||||
0o600,
|
||||
)
|
||||
# Verrouille les permissions en 0600 avant toute écriture, indépendamment de l'umask.
|
||||
os.fchmod(fd, 0o600)
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump(payload, handle, indent=2)
|
||||
os.replace(tmp_file, self._state_file)
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Impossible d'écrire le fichier d'état d'authentification Pronote %s : %s.",
|
||||
redact_secrets(str(self._state_file)),
|
||||
redact_exception(exc),
|
||||
)
|
||||
if fd is not None:
|
||||
try:
|
||||
os.close(fd)
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
tmp_file.unlink(missing_ok=True)
|
||||
except Exception as cleanup_exc:
|
||||
logger.debug(
|
||||
"Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s",
|
||||
redact_exception(cleanup_exc),
|
||||
)
|
||||
raise PronoteSyncError(
|
||||
f"Impossible d'écrire le fichier d'état d'authentification Pronote "
|
||||
f"{redact_secrets(str(self._state_file))}."
|
||||
) from None
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Supprime le fichier d'état d'authentification.
|
||||
|
||||
Si le fichier n'existe pas, la méthode ne fait rien et aucune erreur
|
||||
n'est levée.
|
||||
|
||||
:raises OSError: Si la suppression du fichier existant échoue.
|
||||
"""
|
||||
if not self._state_file.exists():
|
||||
return
|
||||
logger.debug(
|
||||
"Suppression du fichier d'état d'authentification Pronote %s.",
|
||||
redact_secrets(str(self._state_file)),
|
||||
)
|
||||
self._state_file.unlink()
|
||||
@@ -10,19 +10,24 @@ des cours et des devoirs se propagent pour déclencher le repli iCal.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Any, Protocol
|
||||
from uuid import uuid4
|
||||
|
||||
import pronotepy
|
||||
import pronotepy.ent as pronotepy_ent
|
||||
import requests
|
||||
|
||||
from pronote_sync.config.settings import PronoteSettings
|
||||
from pronote_sync.errors import PronoteAuthRotationError
|
||||
from pronote_sync.models.agenda import Lesson, LessonStatus
|
||||
from pronote_sync.models.homework import Homework
|
||||
from pronote_sync.models.message import Message, MessageType
|
||||
from pronote_sync.utils.redaction import redact_exception
|
||||
from pronote_sync.sources.pronote.auth_state import PronoteAuthState
|
||||
from pronote_sync.utils.redaction import redact_exception, redact_secrets
|
||||
from pronote_sync.utils.uid import generate_deterministic_uid, normalize_pronote_uid
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -91,6 +96,51 @@ def _resolve_ent(ent_name: str) -> Any:
|
||||
return resolver
|
||||
|
||||
|
||||
def _collect_auth_secrets(client: PronoteClient) -> list[str]:
|
||||
"""Collecte toutes les valeurs sensibles d'authentification pour la redaction.
|
||||
|
||||
Rassemble le mot de passe, le PIN QR, le contenu du fichier QR (jeton,
|
||||
login, url) et les credentials persistés (token, username) afin de les
|
||||
transmettre comme ``extra_secrets`` aux fonctions de masquage. Une valeur
|
||||
vide ou ``None`` est ignorée.
|
||||
|
||||
:param client: Le client Pronote dont on collecte les secrets.
|
||||
:return: Liste des valeurs sensibles à expurger des logs.
|
||||
:rtype: list[str]
|
||||
"""
|
||||
secrets: list[str] = []
|
||||
settings = client._settings
|
||||
# Mot de passe
|
||||
if settings.password is not None:
|
||||
secrets.append(settings.password.get_secret_value())
|
||||
# PIN QR
|
||||
if settings.qr_pin is not None:
|
||||
secrets.append(settings.qr_pin.get_secret_value())
|
||||
# Contenu du fichier QR (jeton, login, url)
|
||||
if settings.qr_code_file is not None:
|
||||
try:
|
||||
qr_path = Path(settings.qr_code_file)
|
||||
qr_data: Any = json.loads(qr_path.read_text(encoding="utf-8"))
|
||||
for key in ("jeton", "login", "url"):
|
||||
val = qr_data.get(key)
|
||||
if isinstance(val, str):
|
||||
secrets.append(val)
|
||||
except Exception as exc:
|
||||
logger.debug(
|
||||
"Impossible de lire le fichier QR %s : %s",
|
||||
redact_secrets(settings.qr_code_file),
|
||||
redact_exception(exc),
|
||||
)
|
||||
# Credentials persistés (token, username du fichier d'état)
|
||||
if client._auth_state is not None:
|
||||
creds = client._auth_state.load()
|
||||
if creds is not None:
|
||||
for val in creds.values():
|
||||
if isinstance(val, str):
|
||||
secrets.append(val)
|
||||
return [s for s in secrets if s]
|
||||
|
||||
|
||||
class PronoteClientProtocol(Protocol):
|
||||
"""Interface du client Pronote consommée par la logique de repli."""
|
||||
|
||||
@@ -143,24 +193,56 @@ class PronoteClient:
|
||||
exceptions se propager pour déclencher le repli iCal.
|
||||
"""
|
||||
|
||||
def __init__(self, settings: PronoteSettings) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
settings: PronoteSettings,
|
||||
auth_state: PronoteAuthState | None = None,
|
||||
) -> None:
|
||||
"""Initialise le client Pronote sans se connecter.
|
||||
|
||||
:param settings: Paramètres d'accès à Pronote (username, password, ent).
|
||||
:param settings: Paramètres d'accès à Pronote (username, password, ent,
|
||||
mode d'authentification, fichier QR et PIN).
|
||||
:param auth_state: Gestionnaire de persistance du token
|
||||
d'authentification (optionnel ; requis en mode ``qr_token`` pour
|
||||
conserver le token entre les exécutions).
|
||||
"""
|
||||
self._settings: PronoteSettings = settings
|
||||
self._auth_state: PronoteAuthState | None = auth_state
|
||||
self._client: pronotepy.Client | None = None
|
||||
|
||||
def _connect(self) -> pronotepy.Client:
|
||||
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse).
|
||||
|
||||
Le client est créé une seule fois puis réutilisé pour les appels
|
||||
suivants. Le nom d'ENT, s'il est configuré, est résolu via
|
||||
:func:`_resolve_ent` ; en l'absence d'ENT, ``ent=None`` est transmis
|
||||
à ``pronotepy`` pour une connexion directe. Le type de compte
|
||||
(``student`` ou ``parent``) détermine la classe de client utilisée.
|
||||
L'erreur de connexion est relancée sans journalisation, la méthode
|
||||
publique appelante étant responsable de la journaliser.
|
||||
En mode ``password``, utilise l'authentification classique (URL,
|
||||
username, password, ENT). En mode ``qr_token``, utilise le token
|
||||
persisté via :class:`PronoteAuthState`, ou procède à l'enrôlement
|
||||
initial par QR code si aucun token n'est présent.
|
||||
|
||||
:return: Le client ``pronotepy`` connecté.
|
||||
:rtype: pronotepy.Client
|
||||
:raises ValueError: Si les credentials requis sont manquants.
|
||||
:raises PronoteAuthRotationError: Si le token persisté est invalide
|
||||
(rotation requise) ou si l'enrôlement QR échoue.
|
||||
:raises pronotepy.PronoteAPIError: Si la connexion échoue.
|
||||
"""
|
||||
if self._client is not None:
|
||||
return self._client
|
||||
|
||||
if self._settings.auth_mode == "qr_token":
|
||||
self._client = self._connect_qr_token()
|
||||
else:
|
||||
self._client = self._connect_password()
|
||||
return self._client
|
||||
|
||||
def _connect_password(self) -> pronotepy.Client:
|
||||
"""Connecte le client ``pronotepy`` en mode ``password``.
|
||||
|
||||
Le nom d'ENT, s'il est configuré, est résolu via :func:`_resolve_ent` ;
|
||||
en l'absence d'ENT, ``ent=None`` est transmis à ``pronotepy`` pour une
|
||||
connexion directe. Le type de compte (``student`` ou ``parent``)
|
||||
détermine la classe de client utilisée. L'erreur de connexion est
|
||||
relancée sans journalisation, la méthode publique appelante étant
|
||||
responsable de la journaliser.
|
||||
|
||||
:return: Le client ``pronotepy`` connecté.
|
||||
:rtype: pronotepy.Client
|
||||
@@ -168,27 +250,153 @@ class PronoteClient:
|
||||
est manquant, ou si l'ENT fourni est inconnu.
|
||||
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
|
||||
"""
|
||||
if self._client is None:
|
||||
url = self._settings.url
|
||||
username = self._settings.username
|
||||
password = self._settings.password
|
||||
ent = self._settings.ent
|
||||
if url is None or username is None or password is None:
|
||||
raise ValueError("url, username et password sont requis pour pronotepy")
|
||||
resolver = _resolve_ent(ent) if ent is not None else None
|
||||
client_class: type[pronotepy.Client] = (
|
||||
pronotepy.ParentClient
|
||||
if self._settings.account_type == "parent"
|
||||
else pronotepy.Client
|
||||
)
|
||||
self._client = client_class(
|
||||
pronote_url=url,
|
||||
username=username,
|
||||
password=password.get_secret_value(),
|
||||
ent=resolver,
|
||||
)
|
||||
url = self._settings.url
|
||||
username = self._settings.username
|
||||
password = self._settings.password
|
||||
ent = self._settings.ent
|
||||
if url is None or username is None or password is None:
|
||||
raise ValueError("url, username et password sont requis pour pronotepy")
|
||||
resolver = _resolve_ent(ent) if ent is not None else None
|
||||
client_class: type[pronotepy.Client] = (
|
||||
pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client
|
||||
)
|
||||
self._client = client_class(
|
||||
pronote_url=url,
|
||||
username=username,
|
||||
password=password.get_secret_value(),
|
||||
ent=resolver,
|
||||
)
|
||||
return self._client
|
||||
|
||||
def _connect_qr_token(self) -> pronotepy.Client:
|
||||
"""Connecte via token persisté ou enrôlement par QR code.
|
||||
|
||||
En premier lieu, les credentials persistés (``pronote_url``, username,
|
||||
``password``/token, ``uuid``) sont rejoués via
|
||||
``pronotepy.Client.token_login`` si :class:`PronoteAuthState` est
|
||||
disponible et fournit un état. En cas d'échec du login par token
|
||||
(exception ou client non connecté), une :class:`PronoteAuthRotationError`
|
||||
est levée immédiatement, sans repli vers l'enrôlement QR : la rotation
|
||||
du token doit être déclenchée par l'opérateur. L'enrôlement par QR code
|
||||
n'est tenté que lorsqu'aucun credential n'est persisté (premier login) ;
|
||||
le nouveau token est ensuite persisté immédiatement.
|
||||
|
||||
:return: Le client ``pronotepy`` connecté.
|
||||
:rtype: pronotepy.Client
|
||||
:raises PronoteAuthRotationError: Si le token persisté est invalide
|
||||
(expiré ou refusé par Pronote), ou si l'enrôlement QR échoue
|
||||
(fichier QR ou PIN manquant, fichier QR invalide ou expiré).
|
||||
"""
|
||||
client_class: type[pronotepy.Client] = (
|
||||
pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client
|
||||
)
|
||||
|
||||
# Login par token avec les credentials persistés
|
||||
if self._auth_state is not None:
|
||||
creds = self._auth_state.load()
|
||||
if creds is not None:
|
||||
try:
|
||||
client = client_class.token_login(**creds)
|
||||
if client.logged_in:
|
||||
self._auth_state.save(client.export_credentials())
|
||||
return client
|
||||
# logged_in est False — le token est invalide
|
||||
raise PronoteAuthRotationError(
|
||||
"Le token d'authentification Pronote est invalide (non connecté). "
|
||||
"Action requise : supprimez le fichier .pronote_auth_state.json "
|
||||
"et relancez avec un nouveau QR code."
|
||||
) from None
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec du login par token pronotepy : %s",
|
||||
redact_exception(exc, extra_secrets=_collect_auth_secrets(self)),
|
||||
)
|
||||
# Token expiré/invalide — pas de repli vers l'enrôlement QR
|
||||
raise PronoteAuthRotationError(
|
||||
"Le token d'authentification Pronote est expiré ou invalide. "
|
||||
"Action requise : supprimez le fichier .pronote_auth_state.json "
|
||||
"et relancez avec un nouveau QR code (PRONOTE_QR_CODE_FILE + "
|
||||
"PRONOTE_QR_PIN)."
|
||||
) from None
|
||||
|
||||
# Enrôlement : premier login via QR code (aucun credential persisté)
|
||||
client = self._enroll_qr_code(client_class)
|
||||
# Persister le token rotaté immédiatement
|
||||
if self._auth_state is not None:
|
||||
self._auth_state.save(client.export_credentials())
|
||||
return client
|
||||
|
||||
def _enroll_qr_code(self, client_class: type[pronotepy.Client]) -> pronotepy.Client:
|
||||
"""Procède à l'enrôlement initial via QR code pronotepy.
|
||||
|
||||
Le fichier QR JSON doit contenir les clés ``login``, ``jeton`` et
|
||||
``url``. Le PIN et le contenu du fichier ne sont jamais journalisés ;
|
||||
les erreurs propagées sont expurgées.
|
||||
|
||||
:param client_class: Classe de client pronotepy à utiliser.
|
||||
:return: Le client ``pronotepy`` connecté après enrôlement.
|
||||
:rtype: pronotepy.Client
|
||||
:raises PronoteAuthRotationError: Si le fichier QR ou le PIN est
|
||||
manquant, si le fichier QR est illisible ou incomplet, ou si le
|
||||
login par QR code échoue (PIN invalide ou QR code expiré).
|
||||
"""
|
||||
qr_file = self._settings.qr_code_file
|
||||
qr_pin = self._settings.qr_pin
|
||||
|
||||
if qr_file is None or qr_pin is None:
|
||||
raise PronoteAuthRotationError(
|
||||
"Enrôlement QR requis : PRONOTE_QR_CODE_FILE et PRONOTE_QR_PIN sont "
|
||||
"nécessaires pour le premier login en mode qr_token. Supprimez le "
|
||||
"fichier .pronote_auth_state.json si présent et relancez avec un "
|
||||
"QR code frais."
|
||||
) from None
|
||||
|
||||
# Read and validate QR code JSON
|
||||
try:
|
||||
qr_path = Path(qr_file)
|
||||
qr_data: Any = json.loads(qr_path.read_text(encoding="utf-8"))
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Fichier QR invalide %s : %s",
|
||||
redact_secrets(qr_file, extra_secrets=_collect_auth_secrets(self)),
|
||||
redact_exception(exc, extra_secrets=_collect_auth_secrets(self)),
|
||||
)
|
||||
raise PronoteAuthRotationError(
|
||||
"Impossible de lire le fichier QR code : "
|
||||
f"{redact_secrets(qr_file, extra_secrets=_collect_auth_secrets(self))}"
|
||||
) from None
|
||||
|
||||
# Validate required keys
|
||||
for key in ("login", "jeton", "url"):
|
||||
if key not in qr_data:
|
||||
raise PronoteAuthRotationError(
|
||||
f"Le fichier QR code ne contient pas la clé requise : {key}"
|
||||
) from None
|
||||
|
||||
pin_value = qr_pin.get_secret_value()
|
||||
app_uuid = f"pronote-sync-{uuid4().hex}"
|
||||
|
||||
try:
|
||||
client = client_class.qrcode_login(
|
||||
qr_code=qr_data,
|
||||
pin=pin_value,
|
||||
uuid=app_uuid,
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de l'enrôlement QR : %s",
|
||||
redact_exception(exc, extra_secrets=_collect_auth_secrets(self)),
|
||||
)
|
||||
raise PronoteAuthRotationError(
|
||||
"Échec de l'enrôlement par QR code : PIN invalide ou QR code expiré. "
|
||||
"Générez un nouveau QR code dans l'application Pronote et mettez à "
|
||||
"jour PRONOTE_QR_CODE_FILE."
|
||||
) from None
|
||||
|
||||
return client
|
||||
|
||||
def get_messages(self) -> list[Message]:
|
||||
"""Récupère les messages des discussions Pronote.
|
||||
|
||||
@@ -284,6 +492,8 @@ class PronoteClient:
|
||||
:param end: Date de fin de la fenêtre (incluse).
|
||||
:return: Liste des cours.
|
||||
:rtype: list[Lesson]
|
||||
:raises PronoteAuthRotationError: Si le token persisté est invalide et
|
||||
qu'aucun ré-enrôlement n'est possible (fichier QR ou PIN manquant).
|
||||
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
|
||||
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
||||
:raises requests.RequestException: Si une requête réseau échoue.
|
||||
@@ -334,6 +544,8 @@ class PronoteClient:
|
||||
:param end: Date de fin de la fenêtre (incluse).
|
||||
:return: Liste des devoirs.
|
||||
:rtype: list[Homework]
|
||||
:raises PronoteAuthRotationError: Si le token persisté est invalide et
|
||||
qu'aucun ré-enrôlement n'est possible (fichier QR ou PIN manquant).
|
||||
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
|
||||
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
||||
:raises requests.RequestException: Si une requête réseau échoue.
|
||||
|
||||
@@ -23,7 +23,7 @@ from enum import StrEnum
|
||||
from typing import Literal, Protocol
|
||||
|
||||
from pronote_sync.config.settings import Settings
|
||||
from pronote_sync.errors import PipelineCriticalError
|
||||
from pronote_sync.errors import PipelineCriticalError, PronoteAuthRotationError
|
||||
from pronote_sync.models.agenda import Lesson, SchoolEvent
|
||||
from pronote_sync.models.homework import Homework
|
||||
from pronote_sync.models.message import Message
|
||||
@@ -147,13 +147,19 @@ class PronoteFetcher:
|
||||
return self._settings.pronote.ical_url is not None
|
||||
|
||||
def _is_pronotepy_configured(self) -> bool:
|
||||
"""Vérifie que la source pronotepy est entièrement configurée.
|
||||
"""Vérifie si la source pronotepy est utilisable selon le mode d'authentification.
|
||||
|
||||
:return: ``True`` si ``url``, ``username`` et ``password``
|
||||
sont tous définis, ``False`` sinon.
|
||||
:return: ``True`` si pronotepy est configuré pour le mode
|
||||
d'authentification actif, ``False`` sinon.
|
||||
:rtype: bool
|
||||
"""
|
||||
pronote = self._settings.pronote
|
||||
if pronote.auth_mode == "qr_token":
|
||||
# En mode qr_token, seul PRONOTE_URL est requis.
|
||||
# Le QR code et le PIN ne sont nécessaires que pour l'enrôlement initial.
|
||||
# Les exécutions suivantes utilisent le token persisté.
|
||||
return pronote.url is not None
|
||||
# En mode password, URL + identifiant + mot de passe sont requis.
|
||||
return (
|
||||
pronote.url is not None
|
||||
and pronote.username is not None
|
||||
@@ -249,10 +255,14 @@ class PronoteFetcher:
|
||||
:return: Tuple ``(cours, événements scolaires)``.
|
||||
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
||||
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
|
||||
:raises PronoteAuthRotationError: Si une rotation du token d'authentification
|
||||
pronotepy est nécessaire : propagée telle quelle, sans repli.
|
||||
"""
|
||||
primary, fallback = self._agenda_sources()
|
||||
try:
|
||||
return self._fetch_agenda_source(primary)
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de la récupération %s pour l'agenda : %s",
|
||||
@@ -266,6 +276,8 @@ class PronoteFetcher:
|
||||
logger.info("Repli sur %s pour l'agenda.", fallback)
|
||||
try:
|
||||
lessons, school_events = self._fetch_agenda_source(fallback)
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de la récupération %s pour l'agenda : %s",
|
||||
@@ -370,10 +382,14 @@ class PronoteFetcher:
|
||||
:return: Liste des devoirs.
|
||||
:rtype: list[Homework]
|
||||
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
|
||||
:raises PronoteAuthRotationError: Si une rotation du token d'authentification
|
||||
pronotepy est nécessaire : propagée telle quelle, sans repli.
|
||||
"""
|
||||
primary, fallback = self._homework_sources()
|
||||
try:
|
||||
return self._fetch_homework_source(primary, target_date)
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de la récupération %s pour les devoirs : %s",
|
||||
@@ -387,6 +403,8 @@ class PronoteFetcher:
|
||||
logger.info("Repli sur %s pour les devoirs.", fallback)
|
||||
try:
|
||||
homeworks = self._fetch_homework_source(fallback, target_date)
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de la récupération %s pour les devoirs : %s",
|
||||
|
||||
@@ -9,17 +9,19 @@ import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from pronote_sync.config.settings import AISettings, AppSettings, PronoteSettings, Settings
|
||||
from pronote_sync.errors import PipelineCriticalError, PipelineWarning
|
||||
from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError
|
||||
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent
|
||||
from pronote_sync.models.blog import BlogArticle
|
||||
from pronote_sync.models.diff import AgendaDiff
|
||||
from pronote_sync.models.homework import Homework
|
||||
from pronote_sync.models.message import Message
|
||||
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
|
||||
from pronote_sync.models.xmpp import XmppMessage
|
||||
from pronote_sync.pipeline.run import PipelineRunner
|
||||
from pronote_sync.sources.blog.result import BlogRSSFetchResult
|
||||
from pronote_sync.sources.blog.rss import BlogRSSClient
|
||||
from pronote_sync.sources.blog.state import BlogRSSState
|
||||
from pronote_sync.sources.pronote.auth_state import PronoteAuthState
|
||||
from pronote_sync.sources.pronote.fallback import PronoteFetcher
|
||||
from pronote_sync.sync.diff import AgendaComparator
|
||||
|
||||
@@ -393,6 +395,60 @@ def test_from_settings_with_theoretical_agenda_instantiates_comparator(
|
||||
assert isinstance(runner._agenda_comparator, RecordingComparator)
|
||||
|
||||
|
||||
def test_from_settings_password_mode_passes_auth_state_none(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Password mode (default) constructs PronoteClient with auth_state=None."""
|
||||
import pronote_sync.pipeline.run as run_module
|
||||
|
||||
constructed: list[tuple[object, object]] = []
|
||||
|
||||
class RecordingClient:
|
||||
"""PronoteClient constructor recording the supplied auth_state."""
|
||||
|
||||
def __init__(self, settings: PronoteSettings, *, auth_state: object) -> None:
|
||||
"""Record the constructor arguments used by the composition root.
|
||||
|
||||
:param settings: Pronote settings supplied by the composition root.
|
||||
:param auth_state: Auth state handler supplied by the composition root.
|
||||
"""
|
||||
constructed.append((settings, auth_state))
|
||||
|
||||
monkeypatch.setattr(run_module, "PronoteClient", RecordingClient)
|
||||
|
||||
PipelineRunner.from_settings(Settings())
|
||||
|
||||
assert len(constructed) == 1
|
||||
assert constructed[0][1] is None
|
||||
|
||||
|
||||
def test_from_settings_qr_token_mode_passes_auth_state_instance(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""QR-token mode constructs PronoteClient with a PronoteAuthState instance."""
|
||||
import pronote_sync.pipeline.run as run_module
|
||||
|
||||
constructed: list[tuple[object, object]] = []
|
||||
|
||||
class RecordingClient:
|
||||
"""PronoteClient constructor recording the supplied auth_state."""
|
||||
|
||||
def __init__(self, settings: PronoteSettings, *, auth_state: object) -> None:
|
||||
"""Record the constructor arguments used by the composition root.
|
||||
|
||||
:param settings: Pronote settings supplied by the composition root.
|
||||
:param auth_state: Auth state handler supplied by the composition root.
|
||||
"""
|
||||
constructed.append((settings, auth_state))
|
||||
|
||||
monkeypatch.setattr(run_module, "PronoteClient", RecordingClient)
|
||||
|
||||
PipelineRunner.from_settings(Settings(pronote=PronoteSettings(auth_mode="qr_token")))
|
||||
|
||||
assert len(constructed) == 1
|
||||
assert isinstance(constructed[0][1], PronoteAuthState)
|
||||
|
||||
|
||||
def test_runner_reuses_ical_download_and_parse_within_one_run(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
pipeline_inputs: tuple[Lesson, Homework],
|
||||
@@ -1063,3 +1119,338 @@ def test_runner_ical_cache_cleanup_on_second_run(
|
||||
"https://pronote.example.test/calendar.ics",
|
||||
"https://pronote.example.test/calendar.ics",
|
||||
]
|
||||
|
||||
|
||||
def test_rotation_error_sends_xmpp_notification() -> None:
|
||||
"""Une PronoteAuthRotationError envoie une notification XMPP puis retourne un résultat dégradé.
|
||||
|
||||
Ce test vérifie que l'erreur de rotation se propage à travers le pipeline réel
|
||||
(PronoteFetcher → fetch_step → PipelineRunner.run) et déclenche une notification XMPP
|
||||
avec un message actionnable.
|
||||
"""
|
||||
calls: list[str] = []
|
||||
channel = StubChannel(calls)
|
||||
|
||||
# Créer un client Pronote qui lève PronoteAuthRotationError
|
||||
class RotatingPronoteClient:
|
||||
"""Client Pronote qui simule une erreur de rotation de token."""
|
||||
|
||||
def get_lessons(self, start: date, end: date) -> list[Lesson]:
|
||||
"""Lève l'erreur de rotation lors de la récupération des cours.
|
||||
|
||||
:param start: Début de la fenêtre (ignoré).
|
||||
:param end: Fin de la fenêtre (ignoré).
|
||||
:return: Ne retourne jamais.
|
||||
:raises PronoteAuthRotationError: Toujours.
|
||||
"""
|
||||
del start, end
|
||||
raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis")
|
||||
|
||||
def get_homeworks(self, start: date, end: date) -> list[Homework]:
|
||||
"""Ne devrait pas être appelé si fetch_agenda échoue.
|
||||
|
||||
:param start: Début de la fenêtre (ignoré).
|
||||
:param end: Fin de la fenêtre (ignoré).
|
||||
:return: Liste vide.
|
||||
:rtype: list[Homework]
|
||||
"""
|
||||
del start, end
|
||||
return []
|
||||
|
||||
def get_messages(self) -> list[Message]:
|
||||
"""Ne devrait pas être appelé si fetch_agenda échoue.
|
||||
|
||||
:return: Liste vide.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
def get_informations(self) -> list[Message]:
|
||||
"""Ne devrait pas être appelé si fetch_agenda échoue.
|
||||
|
||||
:return: Liste vide.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
settings = Settings(
|
||||
pronote=PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="test",
|
||||
password=SecretStr("test_password"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
)
|
||||
)
|
||||
|
||||
runner = PipelineRunner(
|
||||
settings=settings,
|
||||
pronote_fetcher=PronoteFetcher(settings, RotatingPronoteClient()),
|
||||
channel=channel,
|
||||
now_provider=lambda: datetime(2026, 9, 8, 7, 0),
|
||||
)
|
||||
|
||||
data, errors = runner.run()
|
||||
|
||||
assert data is None
|
||||
assert len(errors) == 1
|
||||
assert isinstance(errors[0], PipelineCriticalError)
|
||||
assert len(channel.messages) == 1
|
||||
message = channel.messages[0]
|
||||
assert isinstance(message, XmppMessage)
|
||||
assert message.target_date == date(2026, 9, 8)
|
||||
assert message.synthesis is not None
|
||||
assert "Rotation" in message.synthesis
|
||||
assert "token" in message.synthesis
|
||||
assert "QR code" in message.synthesis
|
||||
|
||||
|
||||
def test_rotation_error_no_channel_no_xmpp_send() -> None:
|
||||
"""Sans canal XMPP, l'erreur de rotation ne tente aucun envoi.
|
||||
|
||||
Ce test vérifie que même sans canal XMPP configuré, l'erreur de rotation
|
||||
est correctement capturée et retournée dans la liste des erreurs.
|
||||
"""
|
||||
|
||||
class RotatingPronoteClient:
|
||||
"""Client Pronote qui simule une erreur de rotation de token."""
|
||||
|
||||
def get_lessons(self, start: date, end: date) -> list[Lesson]:
|
||||
"""Lève l'erreur de rotation lors de la récupération des cours.
|
||||
|
||||
:param start: Début de la fenêtre (ignoré).
|
||||
:param end: Fin de la fenêtre (ignoré).
|
||||
:return: Ne retourne jamais.
|
||||
:raises PronoteAuthRotationError: Toujours.
|
||||
"""
|
||||
del start, end
|
||||
raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis")
|
||||
|
||||
def get_homeworks(self, start: date, end: date) -> list[Homework]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:param start: Début de la fenêtre (ignoré).
|
||||
:param end: Fin de la fenêtre (ignoré).
|
||||
:return: Liste vide.
|
||||
:rtype: list[Homework]
|
||||
"""
|
||||
del start, end
|
||||
return []
|
||||
|
||||
def get_messages(self) -> list[Message]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:return: Liste vide.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
def get_informations(self) -> list[Message]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:return: Liste vide.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
settings = Settings(
|
||||
pronote=PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="test",
|
||||
password=SecretStr("test_password"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
)
|
||||
)
|
||||
|
||||
runner = PipelineRunner(
|
||||
settings=settings,
|
||||
pronote_fetcher=PronoteFetcher(settings, RotatingPronoteClient()),
|
||||
channel=None,
|
||||
now_provider=lambda: datetime(2026, 9, 8, 7, 0),
|
||||
)
|
||||
|
||||
data, errors = runner.run()
|
||||
|
||||
assert data is None
|
||||
assert len(errors) == 1
|
||||
assert isinstance(errors[0], PipelineCriticalError)
|
||||
|
||||
|
||||
def test_rotation_error_dry_run_no_xmpp_send() -> None:
|
||||
"""En dry-run, l'erreur de rotation n'envoie aucune notification XMPP.
|
||||
|
||||
Ce test vérifie que même en mode dry-run, l'erreur de rotation est correctement
|
||||
capturée et retournée, mais aucune notification XMPP n'est envoyée.
|
||||
"""
|
||||
|
||||
class RotatingPronoteClient:
|
||||
"""Client Pronote qui simule une erreur de rotation de token."""
|
||||
|
||||
def get_lessons(self, start: date, end: date) -> list[Lesson]:
|
||||
"""Lève l'erreur de rotation lors de la récupération des cours.
|
||||
|
||||
:param start: Début de la fenêtre (ignoré).
|
||||
:param end: Fin de la fenêtre (ignoré).
|
||||
:return: Ne retourne jamais.
|
||||
:raises PronoteAuthRotationError: Toujours.
|
||||
"""
|
||||
del start, end
|
||||
raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis")
|
||||
|
||||
def get_homeworks(self, start: date, end: date) -> list[Homework]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:param start: Début de la fenêtre (ignoré).
|
||||
:param end: Fin de la fenêtre (ignoré).
|
||||
:return: Liste vide.
|
||||
:rtype: list[Homework]
|
||||
"""
|
||||
del start, end
|
||||
return []
|
||||
|
||||
def get_messages(self) -> list[Message]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:return: Liste vide.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
def get_informations(self) -> list[Message]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:return: Liste vide.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
calls: list[str] = []
|
||||
channel = StubChannel(calls)
|
||||
|
||||
settings = Settings(
|
||||
pronote=PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="test",
|
||||
password=SecretStr("test_password"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
messages_source="pronotepy",
|
||||
auth_mode="password",
|
||||
qr_code_file=None,
|
||||
qr_pin=None,
|
||||
ical_url=None,
|
||||
)
|
||||
)
|
||||
|
||||
runner = PipelineRunner(
|
||||
settings=settings,
|
||||
pronote_fetcher=PronoteFetcher(settings, RotatingPronoteClient()),
|
||||
channel=channel,
|
||||
dry_run=True,
|
||||
now_provider=lambda: datetime(2026, 9, 8, 7, 0),
|
||||
)
|
||||
|
||||
data, errors = runner.run()
|
||||
|
||||
assert data is None
|
||||
assert len(errors) == 1
|
||||
assert isinstance(errors[0], PipelineCriticalError)
|
||||
assert channel.messages == []
|
||||
assert "send" not in calls
|
||||
|
||||
|
||||
def test_no_secrets_in_xmpp_message() -> None:
|
||||
"""La synthèse XMPP de rotation ne contient aucun secret (token, PIN, URL).
|
||||
|
||||
Ce test vérifie que le message XMPP généré pour une erreur de rotation
|
||||
ne contient aucun secret sensible, même si l'erreur originale en contenait.
|
||||
"""
|
||||
|
||||
class RotatingPronoteClient:
|
||||
"""Client Pronote qui simule une erreur de rotation avec secrets dans message."""
|
||||
|
||||
def get_lessons(self, start: date, end: date) -> list[Lesson]:
|
||||
"""Lève l'erreur de rotation avec message contenant des secrets.
|
||||
|
||||
:param start: Début de la fenêtre (ignoré).
|
||||
:param end: Fin de la fenêtre (ignoré).
|
||||
:return: Ne retourne jamais.
|
||||
:raises PronoteAuthRotationError: Toujours, avec des secrets dans le message.
|
||||
"""
|
||||
del start, end
|
||||
raise PronoteAuthRotationError(
|
||||
"Token sk-sentinel-token-987654 invalide et PIN 000000 pour "
|
||||
"https://pronote.sentinel.example/icalsecurise"
|
||||
)
|
||||
|
||||
def get_homeworks(self, start: date, end: date) -> list[Homework]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:param start: Début de la fenêtre (ignoré).
|
||||
:param end: Fin de la fenêtre (ignoré).
|
||||
:return: Liste vide.
|
||||
:rtype: list[Homework]
|
||||
"""
|
||||
del start, end
|
||||
return []
|
||||
|
||||
def get_messages(self) -> list[Message]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:return: Liste vide.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
def get_informations(self) -> list[Message]:
|
||||
"""Ne devrait pas être appelé.
|
||||
|
||||
:return: Liste vide.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
channel = StubChannel([])
|
||||
|
||||
settings = Settings(
|
||||
pronote=PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="test",
|
||||
password=SecretStr("test_password"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
)
|
||||
)
|
||||
|
||||
runner = PipelineRunner(
|
||||
settings=settings,
|
||||
pronote_fetcher=PronoteFetcher(settings, RotatingPronoteClient()),
|
||||
channel=channel,
|
||||
now_provider=lambda: datetime(2026, 9, 8, 7, 0),
|
||||
)
|
||||
|
||||
data, errors = runner.run()
|
||||
|
||||
assert data is None
|
||||
assert len(errors) == 1
|
||||
assert len(channel.messages) == 1
|
||||
message = channel.messages[0]
|
||||
assert isinstance(message, XmppMessage)
|
||||
assert message.synthesis is not None
|
||||
# Vérifier que les secrets ne sont pas dans le message final
|
||||
assert "sk-sentinel-token-987654" not in message.synthesis
|
||||
assert "000000" not in message.synthesis
|
||||
assert "pronote.sentinel.example" not in message.synthesis
|
||||
# Vérifier que le message contient les instructions actionnables
|
||||
assert ".pronote_auth_state.json" in message.synthesis
|
||||
assert "PRONOTE_QR_CODE_FILE" in message.synthesis
|
||||
assert "PRONOTE_QR_PIN" in message.synthesis
|
||||
|
||||
@@ -131,4 +131,80 @@ def test_url_from_pronote_url_env_var(monkeypatch: MonkeyPatch) -> None:
|
||||
assert settings.pronote.url == test_url
|
||||
|
||||
|
||||
def test_auth_mode_default_password() -> None:
|
||||
"""Vérifie que ``auth_mode`` vaut ``"password"`` par défaut.
|
||||
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings()
|
||||
assert settings.auth_mode == "password"
|
||||
|
||||
|
||||
def test_auth_mode_env_qr_token(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``PRONOTE_AUTH_MODE=qr_token`` est chargé correctement.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("PRONOTE_AUTH_MODE", "qr_token")
|
||||
settings = load_settings()
|
||||
assert settings.pronote.auth_mode == "qr_token"
|
||||
|
||||
|
||||
def test_qr_pin_loaded_as_secretstr_and_masked(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``PRONOTE_QR_PIN`` est chargé en ``SecretStr`` et masqué.
|
||||
|
||||
Le PIN ne doit apparaître nulle part dans les représentations textuelles
|
||||
(str, repr, JSON) : seul le masque ``**********`` est visible.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("PRONOTE_QR_PIN", "123456")
|
||||
settings = load_settings()
|
||||
assert isinstance(settings.pronote.qr_pin, SecretStr)
|
||||
assert settings.pronote.qr_pin.get_secret_value() == "123456"
|
||||
|
||||
str_repr = str(settings)
|
||||
assert "123456" not in str_repr
|
||||
assert "**********" in str_repr
|
||||
|
||||
repr_repr = repr(settings)
|
||||
assert "123456" not in repr_repr
|
||||
assert "**********" in repr_repr
|
||||
|
||||
json_str = settings.model_dump_json()
|
||||
assert "123456" not in json_str
|
||||
assert "**********" in json_str
|
||||
|
||||
|
||||
def test_qr_code_file_loaded_as_plain_string(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``PRONOTE_QR_CODE_FILE`` est chargé comme chaîne simple.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("PRONOTE_QR_CODE_FILE", "/data/qr_code.png")
|
||||
settings = load_settings()
|
||||
assert isinstance(settings.pronote.qr_code_file, str)
|
||||
assert settings.pronote.qr_code_file == "/data/qr_code.png"
|
||||
|
||||
|
||||
def test_qr_pin_in_redaction_secrets(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que le PIN QR est collecté pour la rédaction des secrets.
|
||||
|
||||
Le ``SecretStr`` du PIN doit figurer dans ``redaction_secrets()`` et sa
|
||||
représentation textuelle doit rester masquée.
|
||||
|
||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
||||
:return: None
|
||||
"""
|
||||
monkeypatch.setenv("PRONOTE_QR_PIN", "654321")
|
||||
settings = load_settings()
|
||||
secrets = settings.redaction_secrets()
|
||||
assert settings.pronote.qr_pin in secrets
|
||||
assert "654321" not in repr(settings.pronote.qr_pin)
|
||||
assert "**********" in repr(settings.pronote.qr_pin)
|
||||
|
||||
|
||||
# Ensure trailing newline
|
||||
|
||||
@@ -1315,6 +1315,110 @@ def test_is_pronotepy_configured_without_ent_returns_true() -> None:
|
||||
assert fetcher._is_pronotepy_configured() is True
|
||||
|
||||
|
||||
def test_is_pronotepy_configured_password_mode_all_set() -> None:
|
||||
"""Test _is_pronotepy_configured() en mode password avec tous les champs définis.
|
||||
|
||||
URL, identifiant et mot de passe sont présents : la fonction retourne True.
|
||||
|
||||
:return: None
|
||||
:rtype: None
|
||||
"""
|
||||
settings = Settings(
|
||||
pronote=PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
),
|
||||
app=Settings().app,
|
||||
)
|
||||
|
||||
client: _MockPronoteClientProtocol = MagicMock()
|
||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
||||
|
||||
assert fetcher._is_pronotepy_configured() is True
|
||||
|
||||
|
||||
def test_is_pronotepy_configured_password_mode_missing_password() -> None:
|
||||
"""Test _is_pronotepy_configured() en mode password sans mot de passe.
|
||||
|
||||
Le mot de passe est None : la fonction retourne False.
|
||||
|
||||
:return: None
|
||||
:rtype: None
|
||||
"""
|
||||
settings = Settings(
|
||||
pronote=PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=None,
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
),
|
||||
app=Settings().app,
|
||||
)
|
||||
|
||||
client: _MockPronoteClientProtocol = MagicMock()
|
||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
||||
|
||||
assert fetcher._is_pronotepy_configured() is False
|
||||
|
||||
|
||||
def test_is_pronotepy_configured_qr_token_mode_url_only() -> None:
|
||||
"""Test _is_pronotepy_configured() en mode qr_token avec URL uniquement.
|
||||
|
||||
En mode qr_token, seul l'URL est requis : l'identifiant et le mot de
|
||||
passe peuvent être absents, la fonction retourne True.
|
||||
|
||||
:return: None
|
||||
:rtype: None
|
||||
"""
|
||||
settings = Settings(
|
||||
pronote=PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username=None,
|
||||
password=None,
|
||||
auth_mode="qr_token",
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
),
|
||||
app=Settings().app,
|
||||
)
|
||||
|
||||
client: _MockPronoteClientProtocol = MagicMock()
|
||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
||||
|
||||
assert fetcher._is_pronotepy_configured() is True
|
||||
|
||||
|
||||
def test_is_pronotepy_configured_qr_token_mode_no_url() -> None:
|
||||
"""Test _is_pronotepy_configured() en mode qr_token sans URL.
|
||||
|
||||
L'URL est None : la fonction retourne False, même si le mode qr_token
|
||||
ne requiert que PRONOTE_URL.
|
||||
|
||||
:return: None
|
||||
:rtype: None
|
||||
"""
|
||||
settings = Settings(
|
||||
pronote=PronoteSettings(
|
||||
url=None,
|
||||
username=None,
|
||||
password=None,
|
||||
auth_mode="qr_token",
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
),
|
||||
app=Settings().app,
|
||||
)
|
||||
|
||||
client: _MockPronoteClientProtocol = MagicMock()
|
||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
||||
|
||||
assert fetcher._is_pronotepy_configured() is False
|
||||
|
||||
|
||||
def test_agenda_sources_auto_without_ical_and_without_ent_returns_pronotepy() -> None:
|
||||
"""Test _agenda_sources() en mode AUTO sans iCal URL et sans ent retourne pronotepy.
|
||||
|
||||
|
||||
208
tests/unit/test_pronote_auth_state.py
Normal file
208
tests/unit/test_pronote_auth_state.py
Normal file
@@ -0,0 +1,208 @@
|
||||
"""Tests unitaires pour le gestionnaire d'état d'authentification Pronote.
|
||||
|
||||
Ce module valide le comportement de :class:`PronoteAuthState` dans
|
||||
:mod:`pronote_sync.sources.pronote.auth_state`. Les tests couvrent :
|
||||
|
||||
- Le chargement des credentials (absent, corrompu, version invalide),
|
||||
- La persistance et le rechargement des credentials,
|
||||
- Les permissions ``0600`` du fichier d'état,
|
||||
- La suppression via :meth:`clear`,
|
||||
- L'absence de fuite des credentials dans les journaux.
|
||||
|
||||
Tous les tests utilisent des fichiers temporaires via la fixture ``tmp_path``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from pronote_sync.sources.pronote.auth_state import PronoteAuthState
|
||||
|
||||
|
||||
def test_load_no_file_returns_none(tmp_path: Path) -> None:
|
||||
"""Vérifie qu'un fichier d'état absent renvoie ``None``.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:return: None
|
||||
"""
|
||||
state = PronoteAuthState(tmp_path / "missing.json")
|
||||
|
||||
assert state.load() is None
|
||||
|
||||
|
||||
def test_save_then_load_roundtrip(tmp_path: Path) -> None:
|
||||
"""Vérifie que des credentials sauvegardés sont rechargés à l'identique.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "auth.json"
|
||||
credentials = {
|
||||
"pronote_url": "https://example.com/pronote",
|
||||
"username": "parent-1",
|
||||
"password": "token-123", # pragma: allowlist secret
|
||||
"uuid": "uuid-456",
|
||||
}
|
||||
|
||||
state = PronoteAuthState(state_file)
|
||||
state.save(credentials)
|
||||
loaded = PronoteAuthState(state_file).load()
|
||||
|
||||
assert loaded == credentials
|
||||
|
||||
|
||||
def test_load_corrupted_json_returns_none(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""Vérifie qu'un fichier JSON corrompu renvoie ``None`` et journalise un avertissement.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "corrupt.json"
|
||||
state_file.write_text("not json{", encoding="utf-8")
|
||||
|
||||
with caplog.at_level("WARNING"):
|
||||
result = PronoteAuthState(state_file).load()
|
||||
|
||||
assert result is None
|
||||
assert "Impossible de charger le fichier d'état d'authentification Pronote" in caplog.text
|
||||
|
||||
|
||||
def test_load_wrong_version_returns_none(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""Vérifie qu'une version non supportée renvoie ``None`` et journalise un avertissement.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "wrong_version.json"
|
||||
state_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"version": 2,
|
||||
"credentials": {
|
||||
"pronote_url": "https://example.com",
|
||||
"username": "u",
|
||||
"password": "t",
|
||||
"uuid": "i",
|
||||
},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
with caplog.at_level("WARNING"):
|
||||
result = PronoteAuthState(state_file).load()
|
||||
|
||||
assert result is None
|
||||
assert "version absente ou non supportée" in caplog.text
|
||||
|
||||
|
||||
def test_load_missing_version_returns_none(
|
||||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Vérifie qu'un fichier sans champ version renvoie ``None``.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "missing_version.json"
|
||||
state_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"credentials": {
|
||||
"pronote_url": "https://example.com",
|
||||
"username": "u",
|
||||
"password": "t",
|
||||
"uuid": "i",
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
with caplog.at_level("WARNING"):
|
||||
result = PronoteAuthState(state_file).load()
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_clear_removes_file(tmp_path: Path) -> None:
|
||||
"""Vérifie que clear supprime le fichier d'état existant.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "auth.json"
|
||||
state = PronoteAuthState(state_file)
|
||||
state.save({"pronote_url": "u", "username": "u", "password": "t", "uuid": "i"})
|
||||
|
||||
assert state_file.exists()
|
||||
state.clear()
|
||||
|
||||
assert not state_file.exists()
|
||||
|
||||
|
||||
def test_clear_no_file_noop(tmp_path: Path) -> None:
|
||||
"""Vérifie que clear ne fait rien quand le fichier n'existe pas.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:return: None
|
||||
"""
|
||||
state = PronoteAuthState(tmp_path / "missing.json")
|
||||
|
||||
state.clear()
|
||||
|
||||
|
||||
def test_save_creates_file_with_0600_permissions(tmp_path: Path) -> None:
|
||||
"""Vérifie que le fichier d'état est créé avec les permissions ``0600``.
|
||||
|
||||
Le fichier contient un token vivant : il doit être lisible uniquement
|
||||
par le propriétaire.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "auth.json"
|
||||
state = PronoteAuthState(state_file)
|
||||
state.save({"pronote_url": "u", "username": "u", "password": "t", "uuid": "i"})
|
||||
|
||||
assert os.stat(state_file).st_mode & 0o777 == 0o600
|
||||
|
||||
|
||||
def test_no_credentials_in_logs(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""Vérifie qu'aucun contenu des credentials n'apparaît dans les journaux.
|
||||
|
||||
Des sentinelles distinctes sont utilisées pour ``pronote_url``,
|
||||
``username``, ``password`` et ``uuid`` ; aucun de ces marqueurs ne doit
|
||||
apparaître dans les messages journalisés lors d'une sauvegarde, d'un
|
||||
chargement et d'une suppression.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:param caplog: Fixture pytest pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "auth.json"
|
||||
credentials = {
|
||||
"pronote_url": "https://SENTINEL_URL_ZZZ.example/pronote",
|
||||
"username": "SENTINEL_USER_ZZZ",
|
||||
"password": "SENTINEL_PASSWORD_ZZZ", # pragma: allowlist secret
|
||||
"uuid": "SENTINEL_UUID_ZZZ",
|
||||
}
|
||||
|
||||
state = PronoteAuthState(state_file)
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
state.save(credentials)
|
||||
state.load()
|
||||
state.clear()
|
||||
|
||||
assert "SENTINEL_URL_ZZZ" not in caplog.text
|
||||
assert "SENTINEL_USER_ZZZ" not in caplog.text
|
||||
assert "SENTINEL_PASSWORD_ZZZ" not in caplog.text
|
||||
assert "SENTINEL_UUID_ZZZ" not in caplog.text
|
||||
@@ -7,7 +7,10 @@ utilisent des mocks pour éviter tout accès réseau réel à Pronote.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pronotepy
|
||||
import pytest
|
||||
@@ -15,9 +18,11 @@ import pytest_mock
|
||||
from pydantic import SecretStr
|
||||
|
||||
from pronote_sync.config.settings import PronoteSettings
|
||||
from pronote_sync.errors import PronoteAuthRotationError
|
||||
from pronote_sync.models.agenda import Lesson, LessonStatus
|
||||
from pronote_sync.models.homework import Homework
|
||||
from pronote_sync.models.message import Message, MessageType
|
||||
from pronote_sync.sources.pronote.auth_state import PronoteAuthState
|
||||
from pronote_sync.sources.pronote.client import PronoteClient, PronoteClientProtocol
|
||||
|
||||
# --- Protocol tests ---
|
||||
@@ -613,4 +618,585 @@ def test_get_informations_degraded_on_error(
|
||||
assert messages == []
|
||||
|
||||
|
||||
# --- QR code / token authentication tests ---
|
||||
|
||||
|
||||
def test_connect_password_mode_unchanged(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
) -> None:
|
||||
"""Vérifie que le mode password conserve le comportement historique.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param pronote_settings: Paramètres Pronote valides en mode password.
|
||||
:return: None
|
||||
"""
|
||||
from unittest.mock import Mock
|
||||
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_client_class = Mock(return_value=mock_client)
|
||||
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
|
||||
mocker.patch("pronotepy.Client")
|
||||
|
||||
client = PronoteClient(pronote_settings, auth_state=None)
|
||||
connected = client._connect()
|
||||
|
||||
assert connected is mock_client
|
||||
mock_client_class.assert_called_once_with(
|
||||
pronote_url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password="testpass", # pragma: allowlist secret
|
||||
ent=mocker.ANY,
|
||||
)
|
||||
# Connexion paresseuse : un second appel réutilise le client déjà créé
|
||||
client._connect()
|
||||
assert mock_client_class.call_count == 1
|
||||
|
||||
|
||||
def test_connect_qr_token_with_persisted_creds(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie le login par token persisté en mode qr_token.
|
||||
|
||||
Les credentials chargés depuis :class:`PronoteAuthState` sont rejoués via
|
||||
``token_login`` et le token rotate est resauvegardé.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
creds = {
|
||||
"pronote_url": "https://pronote.example.com",
|
||||
"username": "testuser",
|
||||
"password": "persisted-token", # pragma: allowlist secret
|
||||
"uuid": "persisted-uuid",
|
||||
}
|
||||
rotated_creds = {**creds, "uuid": "rotated-uuid"}
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = creds
|
||||
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_client.logged_in = True
|
||||
mock_client.export_credentials.return_value = rotated_creds
|
||||
mocker.patch("pronotepy.ParentClient.token_login", return_value=mock_client)
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
connected = client._connect()
|
||||
|
||||
assert connected is mock_client
|
||||
pronotepy.ParentClient.token_login.assert_called_once_with(**creds) # type: ignore[attr-defined]
|
||||
auth_state.save.assert_called_once_with(rotated_creds)
|
||||
|
||||
|
||||
def test_connect_qr_token_no_creds_with_qr_code(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Vérifie l'enrôlement initial par QR code quand aucun token n'est persisté.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:return: None
|
||||
"""
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"login": "testuser",
|
||||
"jeton": "qr-jeton",
|
||||
"url": "https://pronote.example.com",
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = None
|
||||
|
||||
creds = {
|
||||
"pronote_url": "https://pronote.example.com",
|
||||
"username": "testuser",
|
||||
"password": "new-token", # pragma: allowlist secret
|
||||
"uuid": "new-uuid",
|
||||
}
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_client.logged_in = True
|
||||
mock_client.export_credentials.return_value = creds
|
||||
mocker.patch("pronotepy.ParentClient.qrcode_login", return_value=mock_client)
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr("123456"),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
connected = client._connect()
|
||||
|
||||
assert connected is mock_client
|
||||
qrcode_login = pronotepy.ParentClient.qrcode_login
|
||||
qrcode_login.assert_called_once() # type: ignore[attr-defined]
|
||||
kwargs = qrcode_login.call_args.kwargs # type: ignore[attr-defined]
|
||||
assert kwargs["pin"] == "123456"
|
||||
assert kwargs["qr_code"] == {
|
||||
"login": "testuser",
|
||||
"jeton": "qr-jeton",
|
||||
"url": "https://pronote.example.com",
|
||||
}
|
||||
assert kwargs["uuid"].startswith("pronote-sync-")
|
||||
auth_state.save.assert_called_once_with(creds)
|
||||
|
||||
|
||||
def test_connect_qr_token_token_login_fails_raises_rotation_error(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Vérifie la levée de PronoteAuthRotationError quand le token persisté est invalide.
|
||||
|
||||
En cas d'échec du login par token, aucun repli vers l'enrôlement QR
|
||||
n'est tenté : l'erreur de rotation est levée immédiatement, même si un
|
||||
fichier QR est disponible.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:return: None
|
||||
"""
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"login": "testuser",
|
||||
"jeton": "qr-jeton",
|
||||
"url": "https://pronote.example.com",
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = {
|
||||
"pronote_url": "https://pronote.example.com",
|
||||
"username": "testuser",
|
||||
"password": "expired-token", # pragma: allowlist secret
|
||||
"uuid": "old-uuid",
|
||||
}
|
||||
|
||||
token_login = mocker.patch("pronotepy.ParentClient.token_login")
|
||||
token_login.side_effect = pronotepy.PronoteAPIError("token invalide")
|
||||
qrcode_login = mocker.patch("pronotepy.ParentClient.qrcode_login")
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr("123456"),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
token_login.assert_called_once()
|
||||
qrcode_login.assert_not_called()
|
||||
auth_state.save.assert_not_called()
|
||||
message = str(exc_info.value)
|
||||
assert "expiré ou invalide" in message
|
||||
assert ".pronote_auth_state.json" in message
|
||||
assert "PRONOTE_QR_CODE_FILE" in message
|
||||
|
||||
|
||||
def test_token_login_failure_raises_rotation_not_enroll(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Vérifie qu'un login par token non connecté lève PronoteAuthRotationError sans enrôlement QR.
|
||||
|
||||
``token_login`` retourne un client non connecté (``logged_in`` False) :
|
||||
l'erreur de rotation est levée immédiatement et ``qrcode_login`` n'est
|
||||
jamais appelé, même avec un QR code disponible.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:return: None
|
||||
"""
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"login": "testuser",
|
||||
"jeton": "qr-jeton",
|
||||
"url": "https://pronote.example.com",
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = {
|
||||
"pronote_url": "https://pronote.example.com",
|
||||
"username": "testuser",
|
||||
"password": "expired-token", # pragma: allowlist secret
|
||||
"uuid": "old-uuid",
|
||||
}
|
||||
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_client.logged_in = False
|
||||
token_login = mocker.patch("pronotepy.ParentClient.token_login", return_value=mock_client)
|
||||
qrcode_login = mocker.patch("pronotepy.ParentClient.qrcode_login")
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr("123456"),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
token_login.assert_called_once()
|
||||
qrcode_login.assert_not_called()
|
||||
auth_state.save.assert_not_called()
|
||||
message = str(exc_info.value)
|
||||
assert "non connecté" in message
|
||||
assert ".pronote_auth_state.json" in message
|
||||
|
||||
|
||||
def test_connect_qr_token_no_creds_no_qr_raises_rotation_error(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie la levée de PronoteAuthRotationError sans token persisté ni QR code.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = None
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
message = str(exc_info.value)
|
||||
assert "PRONOTE_QR_CODE_FILE" in message
|
||||
assert "PRONOTE_QR_PIN" in message
|
||||
assert ".pronote_auth_state.json" in message
|
||||
|
||||
|
||||
def test_connect_qr_token_token_login_fails_no_qr_raises_rotation_error(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie la levée de PronoteAuthRotationError quand le token échoue sans QR.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = {
|
||||
"pronote_url": "https://pronote.example.com",
|
||||
"username": "testuser",
|
||||
"password": "expired-token", # pragma: allowlist secret
|
||||
"uuid": "old-uuid",
|
||||
}
|
||||
token_login = mocker.patch("pronotepy.ParentClient.token_login")
|
||||
token_login.side_effect = pronotepy.PronoteAPIError("token invalide")
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
message = str(exc_info.value)
|
||||
assert "expiré ou invalide" in message
|
||||
assert ".pronote_auth_state.json" in message
|
||||
assert "PRONOTE_QR_CODE_FILE" in message
|
||||
|
||||
|
||||
def test_connect_qr_token_invalid_qr_json_raises_rotation_error(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Vérifie la levée de PronoteAuthRotationError pour un fichier QR illisible.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:return: None
|
||||
"""
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text("{json invalide", encoding="utf-8")
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = None
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr("123456"),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
assert "Impossible de lire le fichier QR code" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_connect_qr_token_missing_qr_key_raises_rotation_error(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Vérifie la levée de PronoteAuthRotationError quand une clé QR requise manque.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:return: None
|
||||
"""
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text(
|
||||
json.dumps({"login": "testuser", "url": "https://pronote.example.com"}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = None
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr("123456"),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
assert "jeton" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_connect_qr_token_qrcode_login_fails_raises_rotation_error(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Vérifie la levée de PronoteAuthRotationError quand le login QR échoue.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:return: None
|
||||
"""
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"login": "testuser",
|
||||
"jeton": "qr-jeton",
|
||||
"url": "https://pronote.example.com",
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = None
|
||||
mocker.patch(
|
||||
"pronotepy.ParentClient.qrcode_login",
|
||||
side_effect=pronotepy.exceptions.QRCodeDecryptError("PIN incorrect"),
|
||||
)
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr("123456"),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
message = str(exc_info.value)
|
||||
assert "PIN invalide ou QR code expiré" in message
|
||||
assert "PRONOTE_QR_CODE_FILE" in message
|
||||
|
||||
|
||||
def test_no_secrets_in_rotation_error_messages(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie qu'aucun secret ne fuit dans les erreurs ni les logs de rotation.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:param caplog: Fixture pytest de capture des logs.
|
||||
:return: None
|
||||
"""
|
||||
sentinel_pin = "SENTINEL_PIN_42"
|
||||
sentinel_token = "SENTINEL_TOKEN_7"
|
||||
sentinel_url = "https://sentinel-url.pronote.example.com"
|
||||
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"login": "testuser",
|
||||
"jeton": sentinel_token,
|
||||
"url": sentinel_url,
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = None
|
||||
|
||||
mocker.patch(
|
||||
"pronotepy.ParentClient.qrcode_login",
|
||||
side_effect=pronotepy.exceptions.QRCodeDecryptError(
|
||||
f"token: {sentinel_token} password: {sentinel_pin}"
|
||||
),
|
||||
)
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr(sentinel_pin),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with caplog.at_level(logging.ERROR, logger="pronote_sync.sources.pronote.client"):
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
message = str(exc_info.value)
|
||||
assert sentinel_pin not in message
|
||||
assert sentinel_token not in message
|
||||
assert "sentinel-url" not in message
|
||||
assert caplog.text
|
||||
assert sentinel_pin not in caplog.text
|
||||
assert sentinel_token not in caplog.text
|
||||
assert "sentinel-url" not in caplog.text
|
||||
|
||||
|
||||
def test_no_raw_secrets_in_logs(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie l'expurgation de secrets bruts sans motif reconnaissable dans les logs.
|
||||
|
||||
Des sentinelles distinctes pour le token persisté, le PIN QR et le jeton
|
||||
QR sont injectées dans le message d'exception de ``token_login`` sans
|
||||
motif ``cle=valeur`` ni format d'URL ; elles ne doivent apparaître ni
|
||||
dans les logs ni dans l'erreur de rotation levée.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:param caplog: Fixture pytest de capture des logs.
|
||||
:return: None
|
||||
"""
|
||||
sentinel_token = "SENTINEL_RAW_TOKEN_ALPHA"
|
||||
sentinel_pin = "SENTINEL_RAW_PIN_BRAVO"
|
||||
sentinel_jeton = "SENTINEL_RAW_JETON_CHARLIE"
|
||||
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"login": "testuser",
|
||||
"jeton": sentinel_jeton,
|
||||
"url": "https://pronote.example.com",
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load.return_value = {
|
||||
"pronote_url": "https://pronote.example.com",
|
||||
"username": "testuser",
|
||||
"password": sentinel_token,
|
||||
"uuid": "old-uuid",
|
||||
}
|
||||
|
||||
mocker.patch(
|
||||
"pronotepy.ParentClient.token_login",
|
||||
side_effect=pronotepy.PronoteAPIError(
|
||||
f"login refusé {sentinel_token} puis {sentinel_pin} puis {sentinel_jeton}"
|
||||
),
|
||||
)
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr(sentinel_pin),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
with caplog.at_level(logging.ERROR, logger="pronote_sync.sources.pronote.client"):
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
client._connect()
|
||||
|
||||
message = str(exc_info.value)
|
||||
assert sentinel_token not in message
|
||||
assert sentinel_pin not in message
|
||||
assert sentinel_jeton not in message
|
||||
assert caplog.text
|
||||
assert sentinel_token not in caplog.text
|
||||
assert sentinel_pin not in caplog.text
|
||||
assert sentinel_jeton not in caplog.text
|
||||
|
||||
|
||||
# Ensure trailing newline
|
||||
|
||||
180
tests/unit/test_rotation_propagation.py
Normal file
180
tests/unit/test_rotation_propagation.py
Normal file
@@ -0,0 +1,180 @@
|
||||
"""Unit tests for PronoteAuthRotationError propagation through each layer.
|
||||
|
||||
These tests verify that the rotation error propagates correctly through the
|
||||
real call chain without being wrapped in PipelineCriticalError at any layer.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr
|
||||
|
||||
from pronote_sync.config.settings import PronoteSettings, Settings
|
||||
from pronote_sync.errors import PipelineCriticalError, PronoteAuthRotationError
|
||||
from pronote_sync.models.agenda import Lesson, SchoolEvent
|
||||
from pronote_sync.models.homework import Homework
|
||||
from pronote_sync.models.message import Message
|
||||
from pronote_sync.pipeline.steps.fetch import fetch_step
|
||||
from pronote_sync.sources.pronote.fallback import PronoteFetcher
|
||||
|
||||
|
||||
class StubPronoteClientWithRotationError:
|
||||
"""Stub PronoteClient that raises PronoteAuthRotationError from its methods."""
|
||||
|
||||
def get_lessons(self, start: date, end: date) -> list[Lesson]:
|
||||
"""Raise rotation error when fetching lessons.
|
||||
|
||||
:param start: Start date (unused).
|
||||
:param end: End date (unused).
|
||||
:return: Never returns.
|
||||
:raises PronoteAuthRotationError: Always.
|
||||
"""
|
||||
del start, end
|
||||
raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis")
|
||||
|
||||
def get_homeworks(self, start: date, end: date) -> list[Homework]:
|
||||
"""Raise rotation error when fetching homeworks.
|
||||
|
||||
:param start: Start date (unused).
|
||||
:param end: End date (unused).
|
||||
:return: Never returns.
|
||||
:raises PronoteAuthRotationError: Always.
|
||||
"""
|
||||
del start, end
|
||||
raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis")
|
||||
|
||||
def get_messages(self) -> list[Message]:
|
||||
"""Return empty messages list.
|
||||
|
||||
:return: Empty list.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
def get_informations(self) -> list[Message]:
|
||||
"""Return empty information messages list.
|
||||
|
||||
:return: Empty list.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
|
||||
class StubSettings:
|
||||
"""Minimal settings stub for PronoteFetcher."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize with minimal configuration."""
|
||||
self.pronote = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="test",
|
||||
password=SecretStr("test_password"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
agenda_source="pronotepy",
|
||||
homework_source="pronotepy",
|
||||
messages_source="pronotepy",
|
||||
auth_mode="password",
|
||||
qr_code_file=None,
|
||||
qr_pin=None,
|
||||
ical_url=None,
|
||||
)
|
||||
self.app = type("AppSettings", (), {"sync_past_days": 7, "sync_future_days": 7})()
|
||||
|
||||
|
||||
class StubFetcherWithRotationError:
|
||||
"""Stub PronoteFetcher that raises PronoteAuthRotationError from its methods."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the stub fetcher."""
|
||||
self._settings = StubSettings()
|
||||
self._client = StubPronoteClientWithRotationError()
|
||||
|
||||
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||
"""Raise rotation error when fetching agenda.
|
||||
|
||||
:return: Never returns.
|
||||
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
||||
:raises PronoteAuthRotationError: Always.
|
||||
"""
|
||||
raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis")
|
||||
|
||||
def fetch_homework(self, target_date: date) -> list[Homework]:
|
||||
"""Raise rotation error when fetching homework.
|
||||
|
||||
:param target_date: Target date (unused).
|
||||
:return: Never returns.
|
||||
:rtype: list[Homework]
|
||||
:raises PronoteAuthRotationError: Always.
|
||||
"""
|
||||
del target_date
|
||||
raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis")
|
||||
|
||||
def fetch_messages(self) -> list[Message]:
|
||||
"""Return empty messages list.
|
||||
|
||||
:return: Empty list.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
def fetch_informations(self) -> list[Message]:
|
||||
"""Return empty information messages list.
|
||||
|
||||
:return: Empty list.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
return []
|
||||
|
||||
|
||||
def test_pronote_fetcher_fetch_agenda_propagates_rotation_error() -> None:
|
||||
"""PronoteFetcher.fetch_agenda() propagates PronoteAuthRotationError without wrapping.
|
||||
|
||||
This test verifies that when the underlying PronoteClient raises
|
||||
PronoteAuthRotationError, the fetcher propagates it directly without
|
||||
converting it to PipelineCriticalError.
|
||||
"""
|
||||
settings = Settings(pronote=StubSettings().pronote)
|
||||
fetcher = PronoteFetcher(settings, StubPronoteClientWithRotationError())
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
fetcher.fetch_agenda()
|
||||
|
||||
assert "Token persisté expiré" in str(exc_info.value)
|
||||
assert not isinstance(exc_info.value, PipelineCriticalError)
|
||||
|
||||
|
||||
def test_pronote_fetcher_fetch_homework_propagates_rotation_error() -> None:
|
||||
"""PronoteFetcher.fetch_homework() propagates PronoteAuthRotationError without wrapping.
|
||||
|
||||
This test verifies that when the underlying PronoteClient raises
|
||||
PronoteAuthRotationError, the fetcher propagates it directly without
|
||||
converting it to PipelineCriticalError.
|
||||
"""
|
||||
settings = Settings(pronote=StubSettings().pronote)
|
||||
fetcher = PronoteFetcher(settings, StubPronoteClientWithRotationError())
|
||||
target_date = date(2026, 9, 9)
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
fetcher.fetch_homework(target_date)
|
||||
|
||||
assert "Token persisté expiré" in str(exc_info.value)
|
||||
assert not isinstance(exc_info.value, PipelineCriticalError)
|
||||
|
||||
|
||||
def test_fetch_step_propagates_rotation_error() -> None:
|
||||
"""fetch_step() propagates PronoteAuthRotationError without wrapping.
|
||||
|
||||
This test verifies that the pipeline step fetch_step() propagates
|
||||
PronoteAuthRotationError directly from the fetcher without converting
|
||||
it to PipelineCriticalError.
|
||||
"""
|
||||
fetcher = StubFetcherWithRotationError()
|
||||
|
||||
with pytest.raises(PronoteAuthRotationError) as exc_info:
|
||||
fetch_step(fetcher, today=date(2026, 9, 8))
|
||||
|
||||
assert "Token persisté expiré" in str(exc_info.value)
|
||||
assert not isinstance(exc_info.value, PipelineCriticalError)
|
||||
Reference in New Issue
Block a user