Compare commits
2 Commits
docs/fix-e
...
feat/qr-to
| Author | SHA1 | Date | |
|---|---|---|---|
|
0363898669
|
|||
| 4a6207f716 |
13
.env.example
13
.env.example
@@ -11,6 +11,19 @@ PRONOTE_AGENDA_SOURCE=auto
|
|||||||
PRONOTE_HOMEWORK_SOURCE=auto
|
PRONOTE_HOMEWORK_SOURCE=auto
|
||||||
PRONOTE_MESSAGES_SOURCE=pronotepy
|
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 ---
|
||||||
CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
|
CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
|
||||||
CALDAV_USERNAME=user@example.com
|
CALDAV_USERNAME=user@example.com
|
||||||
|
|||||||
2
.gitignore
vendored
2
.gitignore
vendored
@@ -48,6 +48,8 @@ Thumbs.db
|
|||||||
# --- Project-specific state files ---
|
# --- Project-specific state files ---
|
||||||
.blog_rss_state.json
|
.blog_rss_state.json
|
||||||
.caldav_sync_state.json
|
.caldav_sync_state.json
|
||||||
|
# État d'authentification pronotepy (QR code / token rotation)
|
||||||
|
.pronote_auth_state.json
|
||||||
*.state.json
|
*.state.json
|
||||||
|
|
||||||
# --- Local scratch / WIP files ---
|
# --- 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
|
- Réutiliser un téléchargement/parsing iCal pour l'agenda et les devoirs pendant un même run, sans
|
||||||
cache global ni persistant.
|
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`
|
### Contrat du provider `openai-compatible`
|
||||||
- Le provider `openai-compatible` réutilise `OpenAISynthesisProvider` avec un `base_url` personnalisé ; aucun nouveau provider n'est créé.
|
- 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).
|
- `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"
|
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
||||||
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
||||||
messages_source: Literal["pronotepy"] = "pronotepy"
|
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")
|
@field_serializer("ical_url")
|
||||||
def _serialize_ical_url(self, value: SecretStr | None) -> str | None:
|
def _serialize_ical_url(self, value: SecretStr | None) -> str | None:
|
||||||
@@ -55,6 +58,18 @@ class PronoteSettings(BaseSettings):
|
|||||||
return None
|
return None
|
||||||
return "**********"
|
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):
|
class CalDAVSettings(BaseSettings):
|
||||||
"""Paramètres d'accès au serveur CalDAV de destination.
|
"""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.
|
"""Énumère tous les secrets configurés pour la rédaction.
|
||||||
|
|
||||||
Collecte les valeurs :class:`pydantic.SecretStr` non vides présentes
|
Collecte les valeurs :class:`pydantic.SecretStr` non vides présentes
|
||||||
dans les sous-configurations (Pronote, CalDAV, XMPP, IA). Les valeurs
|
dans les sous-configurations (URL iCal, mots de passe, code PIN QR et
|
||||||
vides ou ``None`` sont filtrées ; les doublons sont supprimés.
|
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.
|
:return: Tuple de secrets à masquer dans les messages d'erreur.
|
||||||
:rtype: tuple[SecretStr, ...]
|
:rtype: tuple[SecretStr, ...]
|
||||||
@@ -275,6 +291,7 @@ class Settings(BaseSettings):
|
|||||||
secrets = [
|
secrets = [
|
||||||
self.pronote.ical_url,
|
self.pronote.ical_url,
|
||||||
self.pronote.password,
|
self.pronote.password,
|
||||||
|
self.pronote.qr_pin,
|
||||||
self.caldav.url,
|
self.caldav.url,
|
||||||
self.caldav.password,
|
self.caldav.password,
|
||||||
self.xmpp.password,
|
self.xmpp.password,
|
||||||
|
|||||||
@@ -21,6 +21,23 @@ class PronoteSyncError(Exception):
|
|||||||
self.message = message
|
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):
|
class ErrorSeverity(StrEnum):
|
||||||
"""Niveau de gravité d'une erreur produite par le pipeline."""
|
"""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 import get_channel
|
||||||
from pronote_sync.channels.protocol import Channel
|
from pronote_sync.channels.protocol import Channel
|
||||||
from pronote_sync.config.settings import Settings
|
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.blog import ExternalInfo
|
||||||
from pronote_sync.models.pronote import PronoteData
|
from pronote_sync.models.pronote import PronoteData
|
||||||
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
|
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.pipeline.steps.synthesis import synthesis_step
|
||||||
from pronote_sync.sources.blog.rss import BlogRSSClient
|
from pronote_sync.sources.blog.rss import BlogRSSClient
|
||||||
from pronote_sync.sources.blog.state import BlogRSSState
|
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.client import PronoteClient
|
||||||
from pronote_sync.sources.pronote.fallback import PronoteFetcher, PronoteFetcherProtocol
|
from pronote_sync.sources.pronote.fallback import PronoteFetcher, PronoteFetcherProtocol
|
||||||
from pronote_sync.sources.theoretical import get_theoretical_provider
|
from pronote_sync.sources.theoretical import get_theoretical_provider
|
||||||
@@ -133,7 +139,15 @@ class PipelineRunner:
|
|||||||
blog_state = BlogRSSState() if settings.blog.enabled else None
|
blog_state = BlogRSSState() if settings.blog.enabled else None
|
||||||
return cls(
|
return cls(
|
||||||
settings=settings,
|
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,
|
agenda_comparator=comparator,
|
||||||
synthesis_provider=get_synthesis_provider(settings.ai),
|
synthesis_provider=get_synthesis_provider(settings.ai),
|
||||||
channel=get_channel(settings.xmpp, dry_run=effective_dry_run),
|
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
|
des étapes facultatives sont converties en :class:`PipelineWarning` afin
|
||||||
que les étapes suivantes, notamment XMPP, restent exécutées.
|
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.
|
:return: Données Pronote normalisées ou ``None``, puis erreurs et avertissements.
|
||||||
:rtype: tuple[PronoteData | None, list[PipelineError]]
|
:rtype: tuple[PronoteData | None, list[PipelineError]]
|
||||||
"""
|
"""
|
||||||
@@ -271,6 +290,28 @@ class PipelineRunner:
|
|||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
self._warn("send", self._redact(exc))
|
self._warn("send", self._redact(exc))
|
||||||
return data, [*self._errors, *self._warnings]
|
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:
|
except PipelineCriticalError as exc:
|
||||||
logger.error("Erreur critique du pipeline : %s", exc.message)
|
logger.error("Erreur critique du pipeline : %s", exc.message)
|
||||||
self._errors.append(exc)
|
self._errors.append(exc)
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ from __future__ import annotations
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from datetime import date
|
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.agenda import Lesson, SchoolEvent
|
||||||
from pronote_sync.models.homework import Homework
|
from pronote_sync.models.homework import Homework
|
||||||
from pronote_sync.models.message import Message
|
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.
|
:return: Données récupérées et avertissements non critiques.
|
||||||
:rtype: tuple[FetchedPronoteData, list[PipelineWarning]]
|
:rtype: tuple[FetchedPronoteData, list[PipelineWarning]]
|
||||||
:raises PipelineCriticalError: Si l'agenda ou les devoirs ne sont pas disponibles.
|
: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:
|
try:
|
||||||
lessons, school_events = fetcher.fetch_agenda()
|
lessons, school_events = fetcher.fetch_agenda()
|
||||||
@@ -108,6 +110,8 @@ def fetch_step(
|
|||||||
homeworks = fetcher.fetch_homework(target_date)
|
homeworks = fetcher.fetch_homework(target_date)
|
||||||
except PipelineCriticalError:
|
except PipelineCriticalError:
|
||||||
raise
|
raise
|
||||||
|
except PronoteAuthRotationError:
|
||||||
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
raise PipelineCriticalError(
|
raise PipelineCriticalError(
|
||||||
f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
|
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
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
import logging
|
import logging
|
||||||
from datetime import date
|
from datetime import date
|
||||||
|
from pathlib import Path
|
||||||
from typing import Any, Protocol
|
from typing import Any, Protocol
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
import pronotepy
|
import pronotepy
|
||||||
import pronotepy.ent as pronotepy_ent
|
import pronotepy.ent as pronotepy_ent
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
from pronote_sync.config.settings import PronoteSettings
|
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.agenda import Lesson, LessonStatus
|
||||||
from pronote_sync.models.homework import Homework
|
from pronote_sync.models.homework import Homework
|
||||||
from pronote_sync.models.message import Message, MessageType
|
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
|
from pronote_sync.utils.uid import generate_deterministic_uid, normalize_pronote_uid
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -91,6 +96,51 @@ def _resolve_ent(ent_name: str) -> Any:
|
|||||||
return resolver
|
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):
|
class PronoteClientProtocol(Protocol):
|
||||||
"""Interface du client Pronote consommée par la logique de repli."""
|
"""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.
|
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.
|
"""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._settings: PronoteSettings = settings
|
||||||
|
self._auth_state: PronoteAuthState | None = auth_state
|
||||||
self._client: pronotepy.Client | None = None
|
self._client: pronotepy.Client | None = None
|
||||||
|
|
||||||
def _connect(self) -> pronotepy.Client:
|
def _connect(self) -> pronotepy.Client:
|
||||||
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse).
|
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse).
|
||||||
|
|
||||||
Le client est créé une seule fois puis réutilisé pour les appels
|
En mode ``password``, utilise l'authentification classique (URL,
|
||||||
suivants. Le nom d'ENT, s'il est configuré, est résolu via
|
username, password, ENT). En mode ``qr_token``, utilise le token
|
||||||
:func:`_resolve_ent` ; en l'absence d'ENT, ``ent=None`` est transmis
|
persisté via :class:`PronoteAuthState`, ou procède à l'enrôlement
|
||||||
à ``pronotepy`` pour une connexion directe. Le type de compte
|
initial par QR code si aucun token n'est présent.
|
||||||
(``student`` ou ``parent``) détermine la classe de client utilisée.
|
|
||||||
L'erreur de connexion est relancée sans journalisation, la méthode
|
:return: Le client ``pronotepy`` connecté.
|
||||||
publique appelante étant responsable de la journaliser.
|
: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é.
|
:return: Le client ``pronotepy`` connecté.
|
||||||
:rtype: pronotepy.Client
|
:rtype: pronotepy.Client
|
||||||
@@ -168,27 +250,153 @@ class PronoteClient:
|
|||||||
est manquant, ou si l'ENT fourni est inconnu.
|
est manquant, ou si l'ENT fourni est inconnu.
|
||||||
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
|
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
|
||||||
"""
|
"""
|
||||||
if self._client is None:
|
url = self._settings.url
|
||||||
url = self._settings.url
|
username = self._settings.username
|
||||||
username = self._settings.username
|
password = self._settings.password
|
||||||
password = self._settings.password
|
ent = self._settings.ent
|
||||||
ent = self._settings.ent
|
if url is None or username is None or password is None:
|
||||||
if url is None or username is None or password is None:
|
raise ValueError("url, username et password sont requis pour pronotepy")
|
||||||
raise ValueError("url, username et password sont requis pour pronotepy")
|
resolver = _resolve_ent(ent) if ent is not None else None
|
||||||
resolver = _resolve_ent(ent) if ent is not None else None
|
client_class: type[pronotepy.Client] = (
|
||||||
client_class: type[pronotepy.Client] = (
|
pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client
|
||||||
pronotepy.ParentClient
|
)
|
||||||
if self._settings.account_type == "parent"
|
self._client = client_class(
|
||||||
else pronotepy.Client
|
pronote_url=url,
|
||||||
)
|
username=username,
|
||||||
self._client = client_class(
|
password=password.get_secret_value(),
|
||||||
pronote_url=url,
|
ent=resolver,
|
||||||
username=username,
|
)
|
||||||
password=password.get_secret_value(),
|
|
||||||
ent=resolver,
|
|
||||||
)
|
|
||||||
return self._client
|
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]:
|
def get_messages(self) -> list[Message]:
|
||||||
"""Récupère les messages des discussions Pronote.
|
"""Récupère les messages des discussions Pronote.
|
||||||
|
|
||||||
@@ -284,6 +492,8 @@ class PronoteClient:
|
|||||||
:param end: Date de fin de la fenêtre (incluse).
|
:param end: Date de fin de la fenêtre (incluse).
|
||||||
:return: Liste des cours.
|
:return: Liste des cours.
|
||||||
:rtype: list[Lesson]
|
: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 pronotepy.PronoteAPIError: Si l'API Pronote échoue.
|
||||||
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
||||||
:raises requests.RequestException: Si une requête réseau échoue.
|
: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).
|
:param end: Date de fin de la fenêtre (incluse).
|
||||||
:return: Liste des devoirs.
|
:return: Liste des devoirs.
|
||||||
:rtype: list[Homework]
|
: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 pronotepy.PronoteAPIError: Si l'API Pronote échoue.
|
||||||
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
||||||
:raises requests.RequestException: Si une requête réseau échoue.
|
:raises requests.RequestException: Si une requête réseau échoue.
|
||||||
|
|||||||
@@ -23,7 +23,7 @@ from enum import StrEnum
|
|||||||
from typing import Literal, Protocol
|
from typing import Literal, Protocol
|
||||||
|
|
||||||
from pronote_sync.config.settings import Settings
|
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.agenda import Lesson, SchoolEvent
|
||||||
from pronote_sync.models.homework import Homework
|
from pronote_sync.models.homework import Homework
|
||||||
from pronote_sync.models.message import Message
|
from pronote_sync.models.message import Message
|
||||||
@@ -147,13 +147,19 @@ class PronoteFetcher:
|
|||||||
return self._settings.pronote.ical_url is not None
|
return self._settings.pronote.ical_url is not None
|
||||||
|
|
||||||
def _is_pronotepy_configured(self) -> bool:
|
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``
|
:return: ``True`` si pronotepy est configuré pour le mode
|
||||||
sont tous définis, ``False`` sinon.
|
d'authentification actif, ``False`` sinon.
|
||||||
:rtype: bool
|
:rtype: bool
|
||||||
"""
|
"""
|
||||||
pronote = self._settings.pronote
|
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 (
|
return (
|
||||||
pronote.url is not None
|
pronote.url is not None
|
||||||
and pronote.username is not None
|
and pronote.username is not None
|
||||||
@@ -249,10 +255,14 @@ class PronoteFetcher:
|
|||||||
:return: Tuple ``(cours, événements scolaires)``.
|
:return: Tuple ``(cours, événements scolaires)``.
|
||||||
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
||||||
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
|
: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()
|
primary, fallback = self._agenda_sources()
|
||||||
try:
|
try:
|
||||||
return self._fetch_agenda_source(primary)
|
return self._fetch_agenda_source(primary)
|
||||||
|
except PronoteAuthRotationError:
|
||||||
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.error(
|
logger.error(
|
||||||
"Échec de la récupération %s pour l'agenda : %s",
|
"É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)
|
logger.info("Repli sur %s pour l'agenda.", fallback)
|
||||||
try:
|
try:
|
||||||
lessons, school_events = self._fetch_agenda_source(fallback)
|
lessons, school_events = self._fetch_agenda_source(fallback)
|
||||||
|
except PronoteAuthRotationError:
|
||||||
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.error(
|
logger.error(
|
||||||
"Échec de la récupération %s pour l'agenda : %s",
|
"Échec de la récupération %s pour l'agenda : %s",
|
||||||
@@ -370,10 +382,14 @@ class PronoteFetcher:
|
|||||||
:return: Liste des devoirs.
|
:return: Liste des devoirs.
|
||||||
:rtype: list[Homework]
|
:rtype: list[Homework]
|
||||||
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
|
: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()
|
primary, fallback = self._homework_sources()
|
||||||
try:
|
try:
|
||||||
return self._fetch_homework_source(primary, target_date)
|
return self._fetch_homework_source(primary, target_date)
|
||||||
|
except PronoteAuthRotationError:
|
||||||
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.error(
|
logger.error(
|
||||||
"Échec de la récupération %s pour les devoirs : %s",
|
"É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)
|
logger.info("Repli sur %s pour les devoirs.", fallback)
|
||||||
try:
|
try:
|
||||||
homeworks = self._fetch_homework_source(fallback, target_date)
|
homeworks = self._fetch_homework_source(fallback, target_date)
|
||||||
|
except PronoteAuthRotationError:
|
||||||
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.error(
|
logger.error(
|
||||||
"Échec de la récupération %s pour les devoirs : %s",
|
"Échec de la récupération %s pour les devoirs : %s",
|
||||||
|
|||||||
@@ -9,17 +9,19 @@ import pytest
|
|||||||
from pydantic import SecretStr
|
from pydantic import SecretStr
|
||||||
|
|
||||||
from pronote_sync.config.settings import AISettings, AppSettings, PronoteSettings, Settings
|
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.agenda import Lesson, LessonStatus, SchoolEvent
|
||||||
from pronote_sync.models.blog import BlogArticle
|
from pronote_sync.models.blog import BlogArticle
|
||||||
from pronote_sync.models.diff import AgendaDiff
|
from pronote_sync.models.diff import AgendaDiff
|
||||||
from pronote_sync.models.homework import Homework
|
from pronote_sync.models.homework import Homework
|
||||||
from pronote_sync.models.message import Message
|
from pronote_sync.models.message import Message
|
||||||
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
|
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.pipeline.run import PipelineRunner
|
||||||
from pronote_sync.sources.blog.result import BlogRSSFetchResult
|
from pronote_sync.sources.blog.result import BlogRSSFetchResult
|
||||||
from pronote_sync.sources.blog.rss import BlogRSSClient
|
from pronote_sync.sources.blog.rss import BlogRSSClient
|
||||||
from pronote_sync.sources.blog.state import BlogRSSState
|
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.sources.pronote.fallback import PronoteFetcher
|
||||||
from pronote_sync.sync.diff import AgendaComparator
|
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)
|
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(
|
def test_runner_reuses_ical_download_and_parse_within_one_run(
|
||||||
monkeypatch: pytest.MonkeyPatch,
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
pipeline_inputs: tuple[Lesson, Homework],
|
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",
|
||||||
"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
|
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
|
# Ensure trailing newline
|
||||||
|
|||||||
@@ -1315,6 +1315,110 @@ def test_is_pronotepy_configured_without_ent_returns_true() -> None:
|
|||||||
assert fetcher._is_pronotepy_configured() is True
|
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:
|
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.
|
"""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
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
from datetime import date, datetime
|
from datetime import date, datetime
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
import pronotepy
|
import pronotepy
|
||||||
import pytest
|
import pytest
|
||||||
@@ -15,9 +18,11 @@ import pytest_mock
|
|||||||
from pydantic import SecretStr
|
from pydantic import SecretStr
|
||||||
|
|
||||||
from pronote_sync.config.settings import PronoteSettings
|
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.agenda import Lesson, LessonStatus
|
||||||
from pronote_sync.models.homework import Homework
|
from pronote_sync.models.homework import Homework
|
||||||
from pronote_sync.models.message import Message, MessageType
|
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
|
from pronote_sync.sources.pronote.client import PronoteClient, PronoteClientProtocol
|
||||||
|
|
||||||
# --- Protocol tests ---
|
# --- Protocol tests ---
|
||||||
@@ -613,4 +618,585 @@ def test_get_informations_degraded_on_error(
|
|||||||
assert messages == []
|
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
|
# 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