Compare commits
1 Commits
feat/qr-to
...
docs/guide
| Author | SHA1 | Date | |
|---|---|---|---|
|
bb8a6539bc
|
15
.env.example
15
.env.example
@@ -1,6 +1,6 @@
|
|||||||
# --- Pronote ---
|
# --- Pronote ---
|
||||||
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
|
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
|
||||||
PRONOTE_URL=https://college.ent/pronote/parent.html
|
PRONOTE_URL=https://college.ent/pronote/eleve.html
|
||||||
PRONOTE_ACCOUNT_TYPE=parent
|
PRONOTE_ACCOUNT_TYPE=parent
|
||||||
PRONOTE_USERNAME=parent.dupont
|
PRONOTE_USERNAME=parent.dupont
|
||||||
PRONOTE_PASSWORD=your_secure_password
|
PRONOTE_PASSWORD=your_secure_password
|
||||||
@@ -11,19 +11,6 @@ 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,8 +48,6 @@ 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 ---
|
||||||
|
|||||||
69
AGENTS.md
69
AGENTS.md
@@ -146,31 +146,6 @@ 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).
|
||||||
@@ -351,47 +326,3 @@ Un changement est considéré comme **terminé** lorsque :
|
|||||||
- Le *handoff* distingue clairement :
|
- Le *handoff* distingue clairement :
|
||||||
- Ce qui a été vérifié localement (ex. : tests unitaires, linter).
|
- Ce qui a été vérifié localement (ex. : tests unitaires, linter).
|
||||||
- Ce qui nécessite encore une vérification manuelle (ex. : tests d'intégration avec un serveur CalDAV réel).
|
- Ce qui nécessite encore une vérification manuelle (ex. : tests d'intégration avec un serveur CalDAV réel).
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 13. Versionnage et releases
|
|
||||||
|
|
||||||
### Politique de versionnage
|
|
||||||
|
|
||||||
Le projet suit **Semantic Versioning** (semver.org v2.0.0). Phase actuelle : `0.x` (pré-`1.0.0`).
|
|
||||||
|
|
||||||
| Changement | Incrément |
|
|
||||||
|------------|----------|
|
|
||||||
| Défaut constaté au déploiement | Patch (`0.1.Z`) — correction rétrocompatible |
|
|
||||||
| Ajout ou cassure en phase `0.x` | Minor (`0.Y.0`) |
|
|
||||||
| Déploiement réel validé | `1.0.0` |
|
|
||||||
|
|
||||||
### Règle absolue de validation
|
|
||||||
|
|
||||||
**Aucune montée de version (tag + release) ne peut être effectuée
|
|
||||||
sans validation préalable en environnement réel.** Les tests automatisés et la revue de code
|
|
||||||
ne suffisent pas ; le correctif ou la fonctionnalité doit avoir été testé avec succès
|
|
||||||
sur le serveur de production (ou un environnement équivalent) avant de tagger.
|
|
||||||
|
|
||||||
### Procédure de release
|
|
||||||
|
|
||||||
1. **Valider en environnement réel** : le correctif ou la fonctionnalité est testé
|
|
||||||
sur le serveur de production.
|
|
||||||
2. **Mettre à jour `pyproject.toml`** : incrémenter le champ `version` à la nouvelle version.
|
|
||||||
3. **Mettre à jour `CHANGELOG.md`** : ajouter une entrée sous le format
|
|
||||||
[Keep a Changelog](https://keepachangelog.com/en/1.1.0/) avec la nouvelle version et la date.
|
|
||||||
4. **Committer** : un commit `chore: monter en version x.y.z` regroupe
|
|
||||||
les mises à jour de `pyproject.toml` et `CHANGELOG.md`.
|
|
||||||
5. **Tagger** : créer un tag annoté `vx.y.z` sur le commit de version.
|
|
||||||
6. **Pousser le tag** : `git push origin vx.y.z`.
|
|
||||||
7. **Créer la release** sur Gitea avec le changelog correspondant.
|
|
||||||
|
|
||||||
### Cohérence des versions
|
|
||||||
|
|
||||||
Les trois sources de version doivent toujours être synchronisées au moment d'un tag :
|
|
||||||
- Le tag Git (`vx.y.z`)
|
|
||||||
- `pyproject.toml` (`version = "x.y.z"`)
|
|
||||||
- `CHANGELOG.md` (`## [x.y.z] - YYYY-MM-DD`)
|
|
||||||
|
|
||||||
> **Rappel** : Ne jamais créer un tag sans avoir d'abord mis à jour
|
|
||||||
> `pyproject.toml` et `CHANGELOG.md`.
|
|
||||||
|
|||||||
@@ -37,14 +37,11 @@ class PronoteSettings(BaseSettings):
|
|||||||
username: str | None = None
|
username: str | None = None
|
||||||
password: SecretStr | None = None
|
password: SecretStr | None = None
|
||||||
ent: str | None = None
|
ent: str | None = None
|
||||||
url: str | None = None
|
pronote_url: str | None = None
|
||||||
account_type: Literal["student", "parent"] = "parent"
|
account_type: Literal["student", "parent"] = "parent"
|
||||||
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:
|
||||||
@@ -58,18 +55,6 @@ 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.
|
||||||
@@ -281,9 +266,8 @@ 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 (URL iCal, mots de passe, code PIN QR et
|
dans les sous-configurations (Pronote, CalDAV, XMPP, IA). Les valeurs
|
||||||
clé API IA). Les valeurs vides ou ``None`` sont filtrées ; les
|
vides ou ``None`` sont filtrées ; les doublons sont supprimés.
|
||||||
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, ...]
|
||||||
@@ -291,7 +275,6 @@ 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,23 +21,6 @@ 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,12 +11,7 @@ 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 (
|
from pronote_sync.errors import PipelineCriticalError, PipelineError, PipelineWarning
|
||||||
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
|
||||||
@@ -31,7 +26,6 @@ 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
|
||||||
@@ -139,15 +133,7 @@ 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(
|
pronote_fetcher=PronoteFetcher(settings, PronoteClient(settings.pronote)),
|
||||||
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),
|
||||||
@@ -204,11 +190,6 @@ 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]]
|
||||||
"""
|
"""
|
||||||
@@ -290,28 +271,6 @@ 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, PronoteAuthRotationError
|
from pronote_sync.errors import PipelineCriticalError, PipelineWarning
|
||||||
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,8 +101,6 @@ 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()
|
||||||
@@ -110,8 +108,6 @@ 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"
|
||||||
|
|||||||
@@ -1,195 +0,0 @@
|
|||||||
"""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,24 +10,19 @@ 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.sources.pronote.auth_state import PronoteAuthState
|
from pronote_sync.utils.redaction import redact_exception
|
||||||
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__)
|
||||||
@@ -96,51 +91,6 @@ 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."""
|
||||||
|
|
||||||
@@ -193,210 +143,53 @@ class PronoteClient:
|
|||||||
exceptions se propager pour déclencher le repli iCal.
|
exceptions se propager pour déclencher le repli iCal.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(self, settings: PronoteSettings) -> None:
|
||||||
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).
|
||||||
|
|
||||||
En mode ``password``, utilise l'authentification classique (URL,
|
Le client est créé une seule fois puis réutilisé pour les appels
|
||||||
username, password, ENT). En mode ``qr_token``, utilise le token
|
suivants. Le nom d'ENT est résolu via :func:`_resolve_ent` et le
|
||||||
persisté via :class:`PronoteAuthState`, ou procède à l'enrôlement
|
type de compte (``student`` ou ``parent``) détermine la classe de
|
||||||
initial par QR code si aucun token n'est présent.
|
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
|
||||||
:raises ValueError: Si les credentials requis sont manquants.
|
:raises ValueError: Si ``pronote_url``, ``username``, ``password``
|
||||||
:raises PronoteAuthRotationError: Si le token persisté est invalide
|
ou ``ent`` est manquant, ou si l'ENT est inconnu.
|
||||||
(rotation requise) ou si l'enrôlement QR échoue.
|
|
||||||
:raises pronotepy.PronoteAPIError: Si la connexion échoue.
|
|
||||||
"""
|
|
||||||
if self._client is not None:
|
|
||||||
return self._client
|
|
||||||
|
|
||||||
if self._settings.auth_mode == "qr_token":
|
|
||||||
self._client = self._connect_qr_token()
|
|
||||||
else:
|
|
||||||
self._client = self._connect_password()
|
|
||||||
return self._client
|
|
||||||
|
|
||||||
def _connect_password(self) -> pronotepy.Client:
|
|
||||||
"""Connecte le client ``pronotepy`` en mode ``password``.
|
|
||||||
|
|
||||||
Le nom d'ENT, s'il est configuré, est résolu via :func:`_resolve_ent` ;
|
|
||||||
en l'absence d'ENT, ``ent=None`` est transmis à ``pronotepy`` pour une
|
|
||||||
connexion directe. Le type de compte (``student`` ou ``parent``)
|
|
||||||
détermine la classe de client utilisée. L'erreur de connexion est
|
|
||||||
relancée sans journalisation, la méthode publique appelante étant
|
|
||||||
responsable de la journaliser.
|
|
||||||
|
|
||||||
:return: Le client ``pronotepy`` connecté.
|
|
||||||
:rtype: pronotepy.Client
|
|
||||||
:raises ValueError: Si ``url``, ``username`` ou ``password``
|
|
||||||
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.
|
||||||
"""
|
"""
|
||||||
url = self._settings.url
|
if self._client is None:
|
||||||
username = self._settings.username
|
pronote_url = self._settings.pronote_url
|
||||||
password = self._settings.password
|
username = self._settings.username
|
||||||
ent = self._settings.ent
|
password = self._settings.password
|
||||||
if url is None or username is None or password is None:
|
ent = self._settings.ent
|
||||||
raise ValueError("url, username et password sont requis pour pronotepy")
|
if pronote_url is None or username is None or password is None or ent is None:
|
||||||
resolver = _resolve_ent(ent) if ent is not None else None
|
raise ValueError(
|
||||||
client_class: type[pronotepy.Client] = (
|
"pronote_url, username, password et ent sont requis pour pronotepy"
|
||||||
pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client
|
)
|
||||||
)
|
resolver = _resolve_ent(ent)
|
||||||
self._client = client_class(
|
client_class: type[pronotepy.Client] = (
|
||||||
pronote_url=url,
|
pronotepy.ParentClient
|
||||||
username=username,
|
if self._settings.account_type == "parent"
|
||||||
password=password.get_secret_value(),
|
else pronotepy.Client
|
||||||
ent=resolver,
|
)
|
||||||
)
|
self._client = client_class(
|
||||||
|
pronote_url=pronote_url,
|
||||||
|
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.
|
||||||
|
|
||||||
@@ -492,8 +285,6 @@ 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.
|
||||||
@@ -544,8 +335,6 @@ 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, PronoteAuthRotationError
|
from pronote_sync.errors import PipelineCriticalError
|
||||||
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,23 +147,18 @@ 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 si la source pronotepy est utilisable selon le mode d'authentification.
|
"""Vérifie que la source pronotepy est entièrement configurée.
|
||||||
|
|
||||||
:return: ``True`` si pronotepy est configuré pour le mode
|
:return: ``True`` si ``pronote_url``, ``username``, ``password``
|
||||||
d'authentification actif, ``False`` sinon.
|
et ``ent`` sont tous définis, ``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.pronote_url is not None
|
||||||
and pronote.username is not None
|
and pronote.username is not None
|
||||||
and pronote.password is not None
|
and pronote.password is not None
|
||||||
|
and pronote.ent is not None
|
||||||
)
|
)
|
||||||
|
|
||||||
def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]:
|
def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
@@ -255,14 +250,10 @@ 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",
|
||||||
@@ -276,8 +267,6 @@ 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",
|
||||||
@@ -382,14 +371,10 @@ 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",
|
||||||
@@ -403,8 +388,6 @@ 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,19 +9,17 @@ 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, PronoteAuthRotationError
|
from pronote_sync.errors import PipelineCriticalError, PipelineWarning
|
||||||
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
|
||||||
|
|
||||||
@@ -395,60 +393,6 @@ 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],
|
||||||
@@ -1119,338 +1063,3 @@ 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
|
|
||||||
|
|||||||
@@ -116,95 +116,4 @@ def test_no_singleton_import() -> None:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_url_from_pronote_url_env_var(monkeypatch: MonkeyPatch) -> None:
|
|
||||||
"""Vérifie que PRONOTE_URL mappe au champ url via le préfixe PRONOTE_.
|
|
||||||
|
|
||||||
Ce test couvre la régression où PRONOTE_URL n'était pas mappé vers le
|
|
||||||
champ du modèle à cause du double préfixe PRONOTE_.
|
|
||||||
|
|
||||||
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
test_url = "https://example.index-education.net/pronote/parent.html"
|
|
||||||
monkeypatch.setenv("PRONOTE_URL", test_url)
|
|
||||||
settings = load_settings()
|
|
||||||
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
|
||||||
|
|||||||
@@ -53,7 +53,7 @@ def fixture_mock_settings() -> Settings:
|
|||||||
"""
|
"""
|
||||||
return Settings(
|
return Settings(
|
||||||
pronote=PronoteSettings(
|
pronote=PronoteSettings(
|
||||||
url="https://pronote.example.com",
|
pronote_url="https://pronote.example.com",
|
||||||
ical_url=SecretStr("file:///fake/ical.ics"),
|
ical_url=SecretStr("file:///fake/ical.ics"),
|
||||||
agenda_source="auto",
|
agenda_source="auto",
|
||||||
homework_source="auto",
|
homework_source="auto",
|
||||||
@@ -251,7 +251,7 @@ def test_fetch_agenda_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
|
|||||||
:rtype: None
|
:rtype: None
|
||||||
"""
|
"""
|
||||||
# Disable pronotepy so fallback is None
|
# Disable pronotepy so fallback is None
|
||||||
mock_fetcher._settings.pronote.url = None
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
@@ -280,7 +280,7 @@ def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
|
|||||||
"""
|
"""
|
||||||
# Override settings to use ical mode explicitly and disable fallback
|
# Override settings to use ical mode explicitly and disable fallback
|
||||||
mock_fetcher._settings.pronote.agenda_source = "ical"
|
mock_fetcher._settings.pronote.agenda_source = "ical"
|
||||||
mock_fetcher._settings.pronote.url = None
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
@@ -438,7 +438,7 @@ def test_fetch_homework_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
|
|||||||
target_date = date(2025, 9, 10)
|
target_date = date(2025, 9, 10)
|
||||||
|
|
||||||
# Disable pronotepy so fallback is None
|
# Disable pronotepy so fallback is None
|
||||||
mock_fetcher._settings.pronote.url = None
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
@@ -531,7 +531,7 @@ def test_no_secrets_in_error_messages(
|
|||||||
:rtype: None
|
:rtype: None
|
||||||
"""
|
"""
|
||||||
# Disable pronotepy so fallback is None to trigger PipelineCriticalError
|
# Disable pronotepy so fallback is None to trigger PipelineCriticalError
|
||||||
mock_fetcher._settings.pronote.url = None
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
with (
|
with (
|
||||||
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
@@ -629,7 +629,7 @@ def test_fetch_agenda_no_source_configured_raises(mock_fetcher: PronoteFetcher)
|
|||||||
"""
|
"""
|
||||||
# Disable both sources
|
# Disable both sources
|
||||||
mock_fetcher._settings.pronote.ical_url = None
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
mock_fetcher._settings.pronote.url = None
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
with pytest.raises(PipelineCriticalError) as exc_info:
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
mock_fetcher.fetch_agenda()
|
mock_fetcher.fetch_agenda()
|
||||||
@@ -1020,7 +1020,7 @@ def test_homework_sources_explicit_ical_mode_strict(mock_fetcher: PronoteFetcher
|
|||||||
assert fallback is None
|
assert fallback is None
|
||||||
|
|
||||||
# Without pronotepy configured
|
# Without pronotepy configured
|
||||||
mock_fetcher._settings.pronote.url = None
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
primary, fallback = mock_fetcher._homework_sources()
|
primary, fallback = mock_fetcher._homework_sources()
|
||||||
assert primary == "ical"
|
assert primary == "ical"
|
||||||
assert fallback is None
|
assert fallback is None
|
||||||
@@ -1080,7 +1080,7 @@ def test_homework_sources_auto_no_source_configured_raises(mock_fetcher: Pronote
|
|||||||
"""
|
"""
|
||||||
mock_fetcher._settings.pronote.homework_source = "auto"
|
mock_fetcher._settings.pronote.homework_source = "auto"
|
||||||
mock_fetcher._settings.pronote.ical_url = None
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
mock_fetcher._settings.pronote.url = None
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
with pytest.raises(PipelineCriticalError) as exc_info:
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
mock_fetcher._homework_sources()
|
mock_fetcher._homework_sources()
|
||||||
@@ -1088,102 +1088,6 @@ def test_homework_sources_auto_no_source_configured_raises(mock_fetcher: Pronote
|
|||||||
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
|
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_agenda_auto_ical_configured_fails_fallback_to_pronotepy_without_ent(
|
|
||||||
mock_fetcher: PronoteFetcher,
|
|
||||||
) -> None:
|
|
||||||
"""Test le mode auto : iCal configuré mais échoue, repli sur pronotepy sans ent.
|
|
||||||
|
|
||||||
On mock iCal pour échouer, pronotepy configuré sans ent. On vérifie que pronotepy est appelé
|
|
||||||
et que le résultat provient de pronotepy, pas une erreur.
|
|
||||||
|
|
||||||
:param mock_fetcher: Fetcher de test.
|
|
||||||
:return: None
|
|
||||||
:rtype: None
|
|
||||||
"""
|
|
||||||
start_dt = datetime(2025, 9, 1, 8, 0)
|
|
||||||
end_dt = datetime(2025, 9, 1, 9, 30)
|
|
||||||
lessons = [
|
|
||||||
Lesson(
|
|
||||||
id="l1",
|
|
||||||
start=start_dt,
|
|
||||||
end=end_dt,
|
|
||||||
subject="Maths",
|
|
||||||
teachers=("Dupont",),
|
|
||||||
rooms=("S1",),
|
|
||||||
group="2ndeA",
|
|
||||||
status=LessonStatus.NORMAL,
|
|
||||||
content=None,
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
# Override settings to use auto mode with ical_url configured but ent=None
|
|
||||||
mock_fetcher._settings.pronote.agenda_source = "auto"
|
|
||||||
mock_fetcher._settings.pronote.ent = None # Explicitly None
|
|
||||||
|
|
||||||
with (
|
|
||||||
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
|
||||||
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
|
||||||
):
|
|
||||||
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
|
||||||
m_parse_ical.side_effect = OSError("iCal parse error")
|
|
||||||
client = MagicMock()
|
|
||||||
client.get_lessons.return_value = lessons
|
|
||||||
mock_fetcher._pronote_client = client
|
|
||||||
|
|
||||||
result_lessons, result_events = mock_fetcher.fetch_agenda()
|
|
||||||
|
|
||||||
assert result_lessons == lessons
|
|
||||||
assert result_events == []
|
|
||||||
client.get_lessons.assert_called_once()
|
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_homework_auto_ical_configured_fails_fallback_to_pronotepy_without_ent(
|
|
||||||
mock_fetcher: PronoteFetcher,
|
|
||||||
) -> None:
|
|
||||||
"""Test le mode auto des devoirs : iCal configuré mais échoue, repli sur pronotepy sans ent.
|
|
||||||
|
|
||||||
On mock iCal pour échouer, pronotepy configuré sans ent. On vérifie que pronotepy est appelé
|
|
||||||
et que le résultat provient de pronotepy, pas une erreur.
|
|
||||||
|
|
||||||
:param mock_fetcher: Fetcher de test.
|
|
||||||
:return: None
|
|
||||||
:rtype: None
|
|
||||||
"""
|
|
||||||
target_date = date(2025, 9, 10)
|
|
||||||
homeworks = [
|
|
||||||
Homework(
|
|
||||||
id="hw1",
|
|
||||||
subject="Physique",
|
|
||||||
teachers=(),
|
|
||||||
assigned_on=None,
|
|
||||||
due_on=target_date,
|
|
||||||
text="TP à préparer",
|
|
||||||
html="TP à préparer",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
# Override settings to use auto mode with ical_url configured but ent=None
|
|
||||||
mock_fetcher._settings.pronote.homework_source = "auto"
|
|
||||||
mock_fetcher._settings.pronote.ent = None # Explicitly None
|
|
||||||
|
|
||||||
with (
|
|
||||||
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
|
||||||
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
|
||||||
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
|
|
||||||
):
|
|
||||||
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
|
||||||
m_parse_ical.side_effect = OSError("iCal parse error")
|
|
||||||
client = MagicMock()
|
|
||||||
client.get_homeworks.return_value = homeworks
|
|
||||||
mock_fetcher._pronote_client = client
|
|
||||||
m_collect.return_value = homeworks
|
|
||||||
|
|
||||||
result = mock_fetcher.fetch_homework(target_date)
|
|
||||||
|
|
||||||
assert result == homeworks
|
|
||||||
client.get_homeworks.assert_called_once()
|
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error(
|
def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error(
|
||||||
mock_fetcher: PronoteFetcher,
|
mock_fetcher: PronoteFetcher,
|
||||||
) -> None:
|
) -> None:
|
||||||
@@ -1223,7 +1127,7 @@ def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error(
|
|||||||
def test_fetch_homework_auto_fallback_returns_empty_logs_warning(
|
def test_fetch_homework_auto_fallback_returns_empty_logs_warning(
|
||||||
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
|
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test que fetch_homework retourne [] et journalise un avertissement si le repli est vide.
|
"""Test que fetch_homework retourne [] et journalise un avertissement si le repli retourne vide.
|
||||||
|
|
||||||
On mock ICAL pour échouer, pronotepy configuré et retourne vide. On vérifie le retour et le log.
|
On mock ICAL pour échouer, pronotepy configuré et retourne vide. On vérifie le retour et le log.
|
||||||
Ce test utilise le mode AUTO pour tester le comportement de repli.
|
Ce test utilise le mode AUTO pour tester le comportement de repli.
|
||||||
@@ -1282,203 +1186,4 @@ def test_fetch_informations_logs_and_re_raises_secret(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_is_pronotepy_configured_without_ent_returns_true() -> None:
|
|
||||||
"""Test _is_pronotepy_configured() retourne True quand ent est None.
|
|
||||||
|
|
||||||
Les autres champs de la configuration pronotepy sont présents, donc la
|
|
||||||
fonction renvoie True.
|
|
||||||
|
|
||||||
Ce test valide que PRONOTE_ENT est optionnel pour pronotepy.
|
|
||||||
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
from pronote_sync.config.settings import PronoteSettings, Settings
|
|
||||||
from pronote_sync.sources.pronote.fallback import PronoteFetcher
|
|
||||||
|
|
||||||
# Créer des settings avec pronotepy configuré mais sans ent
|
|
||||||
settings = Settings(
|
|
||||||
pronote=PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password=SecretStr("testpass"),
|
|
||||||
ent=None, # Explicitement None
|
|
||||||
agenda_source="pronotepy",
|
|
||||||
homework_source="pronotepy",
|
|
||||||
),
|
|
||||||
app=Settings().app,
|
|
||||||
)
|
|
||||||
|
|
||||||
client: _MockPronoteClientProtocol = MagicMock()
|
|
||||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
|
||||||
|
|
||||||
# Should return True even without ent
|
|
||||||
assert fetcher._is_pronotepy_configured() is True
|
|
||||||
|
|
||||||
|
|
||||||
def test_is_pronotepy_configured_password_mode_all_set() -> None:
|
|
||||||
"""Test _is_pronotepy_configured() en mode password avec tous les champs définis.
|
|
||||||
|
|
||||||
URL, identifiant et mot de passe sont présents : la fonction retourne True.
|
|
||||||
|
|
||||||
:return: None
|
|
||||||
:rtype: None
|
|
||||||
"""
|
|
||||||
settings = Settings(
|
|
||||||
pronote=PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password=SecretStr("testpass"),
|
|
||||||
agenda_source="pronotepy",
|
|
||||||
homework_source="pronotepy",
|
|
||||||
),
|
|
||||||
app=Settings().app,
|
|
||||||
)
|
|
||||||
|
|
||||||
client: _MockPronoteClientProtocol = MagicMock()
|
|
||||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
|
||||||
|
|
||||||
assert fetcher._is_pronotepy_configured() is True
|
|
||||||
|
|
||||||
|
|
||||||
def test_is_pronotepy_configured_password_mode_missing_password() -> None:
|
|
||||||
"""Test _is_pronotepy_configured() en mode password sans mot de passe.
|
|
||||||
|
|
||||||
Le mot de passe est None : la fonction retourne False.
|
|
||||||
|
|
||||||
:return: None
|
|
||||||
:rtype: None
|
|
||||||
"""
|
|
||||||
settings = Settings(
|
|
||||||
pronote=PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password=None,
|
|
||||||
agenda_source="pronotepy",
|
|
||||||
homework_source="pronotepy",
|
|
||||||
),
|
|
||||||
app=Settings().app,
|
|
||||||
)
|
|
||||||
|
|
||||||
client: _MockPronoteClientProtocol = MagicMock()
|
|
||||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
|
||||||
|
|
||||||
assert fetcher._is_pronotepy_configured() is False
|
|
||||||
|
|
||||||
|
|
||||||
def test_is_pronotepy_configured_qr_token_mode_url_only() -> None:
|
|
||||||
"""Test _is_pronotepy_configured() en mode qr_token avec URL uniquement.
|
|
||||||
|
|
||||||
En mode qr_token, seul l'URL est requis : l'identifiant et le mot de
|
|
||||||
passe peuvent être absents, la fonction retourne True.
|
|
||||||
|
|
||||||
:return: None
|
|
||||||
:rtype: None
|
|
||||||
"""
|
|
||||||
settings = Settings(
|
|
||||||
pronote=PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username=None,
|
|
||||||
password=None,
|
|
||||||
auth_mode="qr_token",
|
|
||||||
agenda_source="pronotepy",
|
|
||||||
homework_source="pronotepy",
|
|
||||||
),
|
|
||||||
app=Settings().app,
|
|
||||||
)
|
|
||||||
|
|
||||||
client: _MockPronoteClientProtocol = MagicMock()
|
|
||||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
|
||||||
|
|
||||||
assert fetcher._is_pronotepy_configured() is True
|
|
||||||
|
|
||||||
|
|
||||||
def test_is_pronotepy_configured_qr_token_mode_no_url() -> None:
|
|
||||||
"""Test _is_pronotepy_configured() en mode qr_token sans URL.
|
|
||||||
|
|
||||||
L'URL est None : la fonction retourne False, même si le mode qr_token
|
|
||||||
ne requiert que PRONOTE_URL.
|
|
||||||
|
|
||||||
:return: None
|
|
||||||
:rtype: None
|
|
||||||
"""
|
|
||||||
settings = Settings(
|
|
||||||
pronote=PronoteSettings(
|
|
||||||
url=None,
|
|
||||||
username=None,
|
|
||||||
password=None,
|
|
||||||
auth_mode="qr_token",
|
|
||||||
agenda_source="pronotepy",
|
|
||||||
homework_source="pronotepy",
|
|
||||||
),
|
|
||||||
app=Settings().app,
|
|
||||||
)
|
|
||||||
|
|
||||||
client: _MockPronoteClientProtocol = MagicMock()
|
|
||||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
|
||||||
|
|
||||||
assert fetcher._is_pronotepy_configured() is False
|
|
||||||
|
|
||||||
|
|
||||||
def test_agenda_sources_auto_without_ical_and_without_ent_returns_pronotepy() -> None:
|
|
||||||
"""Test _agenda_sources() en mode AUTO sans iCal URL et sans ent retourne pronotepy.
|
|
||||||
|
|
||||||
Ce test valide que le mode auto peut utiliser pronotepy même sans ent configuré.
|
|
||||||
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
from pronote_sync.config.settings import PronoteSettings, Settings
|
|
||||||
from pronote_sync.sources.pronote.fallback import PronoteFetcher
|
|
||||||
|
|
||||||
settings = Settings(
|
|
||||||
pronote=PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password=SecretStr("testpass"),
|
|
||||||
ent=None, # Explicitement None
|
|
||||||
ical_url=None, # Pas de iCal URL
|
|
||||||
agenda_source="auto",
|
|
||||||
),
|
|
||||||
app=Settings().app,
|
|
||||||
)
|
|
||||||
|
|
||||||
client: _MockPronoteClientProtocol = MagicMock()
|
|
||||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
|
||||||
|
|
||||||
primary, fallback = fetcher._agenda_sources()
|
|
||||||
|
|
||||||
assert primary == "pronotepy"
|
|
||||||
assert fallback is None
|
|
||||||
|
|
||||||
|
|
||||||
def test_homework_sources_auto_without_ical_and_without_ent_returns_pronotepy() -> None:
|
|
||||||
"""Test _homework_sources() en mode AUTO sans iCal URL et sans ent retourne pronotepy.
|
|
||||||
|
|
||||||
Ce test valide que le mode auto peut utiliser pronotepy pour les devoirs même sans ent configuré.
|
|
||||||
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
from pronote_sync.config.settings import PronoteSettings, Settings
|
|
||||||
from pronote_sync.sources.pronote.fallback import PronoteFetcher
|
|
||||||
|
|
||||||
settings = Settings(
|
|
||||||
pronote=PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password=SecretStr("testpass"),
|
|
||||||
ent=None, # Explicitement None
|
|
||||||
ical_url=None, # Pas de iCal URL
|
|
||||||
homework_source="auto",
|
|
||||||
),
|
|
||||||
app=Settings().app,
|
|
||||||
)
|
|
||||||
|
|
||||||
client: _MockPronoteClientProtocol = MagicMock()
|
|
||||||
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
|
|
||||||
|
|
||||||
primary, fallback = fetcher._homework_sources()
|
|
||||||
|
|
||||||
assert primary == "pronotepy"
|
|
||||||
assert fallback is None
|
|
||||||
|
|
||||||
|
|
||||||
# Ensure trailing newline
|
# Ensure trailing newline
|
||||||
|
|||||||
@@ -1,208 +0,0 @@
|
|||||||
"""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,10 +7,7 @@ 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
|
||||||
@@ -18,11 +15,9 @@ 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 ---
|
||||||
@@ -51,7 +46,7 @@ def pronote_settings() -> PronoteSettings:
|
|||||||
:rtype: PronoteSettings
|
:rtype: PronoteSettings
|
||||||
"""
|
"""
|
||||||
return PronoteSettings(
|
return PronoteSettings(
|
||||||
url="https://pronote.example.com",
|
pronote_url="https://pronote.example.com",
|
||||||
username="testuser",
|
username="testuser",
|
||||||
password=SecretStr("testpass"),
|
password=SecretStr("testpass"),
|
||||||
ent="bordeaux",
|
ent="bordeaux",
|
||||||
@@ -314,7 +309,7 @@ def test_missing_credentials_raises(empty_pronote_settings: PronoteSettings) ->
|
|||||||
"""
|
"""
|
||||||
client = PronoteClient(empty_pronote_settings)
|
client = PronoteClient(empty_pronote_settings)
|
||||||
|
|
||||||
with pytest.raises(ValueError, match="url, username et password sont requis pour pronotepy"):
|
with pytest.raises(ValueError, match="pronote_url, username, password et ent sont requis"):
|
||||||
client._connect()
|
client._connect()
|
||||||
|
|
||||||
|
|
||||||
@@ -370,132 +365,6 @@ def test_connect_parent_account_type(
|
|||||||
pronotepy.Client.assert_not_called() # type: ignore[attr-defined]
|
pronotepy.Client.assert_not_called() # type: ignore[attr-defined]
|
||||||
|
|
||||||
|
|
||||||
def test_connect_without_ent_but_with_required_credentials(
|
|
||||||
mocker: pytest_mock.MockerFixture,
|
|
||||||
) -> None:
|
|
||||||
"""Vérifie que _connect() fonctionne sans ent mais avec les autres identifiants requis.
|
|
||||||
|
|
||||||
Ce test valide que PRONOTE_ENT est optionnel pour une connexion directe Pronote.
|
|
||||||
|
|
||||||
:param mocker: Fixture pytest-mock pour le mocking.
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
from unittest.mock import Mock
|
|
||||||
|
|
||||||
from pronote_sync.config.settings import PronoteSettings
|
|
||||||
from pronote_sync.sources.pronote.client import PronoteClient
|
|
||||||
|
|
||||||
# Settings sans ent mais avec les autres champs requis
|
|
||||||
settings = PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password=SecretStr("testpass"),
|
|
||||||
ent=None, # Explicitement None
|
|
||||||
account_type="parent",
|
|
||||||
)
|
|
||||||
|
|
||||||
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(settings)
|
|
||||||
connected_client = client._connect()
|
|
||||||
|
|
||||||
# Should not raise ValueError about missing ent
|
|
||||||
assert connected_client is mock_client
|
|
||||||
|
|
||||||
# Verify ParentClient was called with ent=None
|
|
||||||
mock_client_class.assert_called_once_with(
|
|
||||||
pronote_url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password="testpass", # pragma: allowlist secret
|
|
||||||
ent=None, # ent should be None, not resolved
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_connect_missing_required_credentials_still_raises(
|
|
||||||
mocker: pytest_mock.MockerFixture,
|
|
||||||
) -> None:
|
|
||||||
"""Vérifie que _connect() lève ValueError si url, username ou password manquent.
|
|
||||||
|
|
||||||
Ce test valide que l'erreur ne mentionne plus ent comme requis.
|
|
||||||
|
|
||||||
:param mocker: Fixture pytest-mock pour le mocking.
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
from pronote_sync.config.settings import PronoteSettings
|
|
||||||
from pronote_sync.sources.pronote.client import PronoteClient
|
|
||||||
|
|
||||||
# Settings avec ent mais sans url
|
|
||||||
settings = PronoteSettings(
|
|
||||||
url=None,
|
|
||||||
username="testuser",
|
|
||||||
password=SecretStr("testpass"),
|
|
||||||
ent=None,
|
|
||||||
account_type="parent",
|
|
||||||
)
|
|
||||||
|
|
||||||
client = PronoteClient(settings)
|
|
||||||
|
|
||||||
with pytest.raises(ValueError) as exc_info:
|
|
||||||
client._connect()
|
|
||||||
|
|
||||||
# Error should NOT mention ent as required
|
|
||||||
assert "url, username et password sont requis" in str(exc_info.value)
|
|
||||||
assert "ent" not in str(exc_info.value)
|
|
||||||
|
|
||||||
|
|
||||||
def test_connect_missing_username_raises(mocker: pytest_mock.MockerFixture) -> None:
|
|
||||||
"""Vérifie que _connect() lève ValueError si username manque.
|
|
||||||
|
|
||||||
:param mocker: Fixture pytest-mock pour le mocking.
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
from pronote_sync.config.settings import PronoteSettings
|
|
||||||
from pronote_sync.sources.pronote.client import PronoteClient
|
|
||||||
|
|
||||||
settings = PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username=None,
|
|
||||||
password=SecretStr("testpass"),
|
|
||||||
ent=None,
|
|
||||||
account_type="parent",
|
|
||||||
)
|
|
||||||
|
|
||||||
client = PronoteClient(settings)
|
|
||||||
|
|
||||||
with pytest.raises(ValueError) as exc_info:
|
|
||||||
client._connect()
|
|
||||||
|
|
||||||
assert "url, username et password sont requis" in str(exc_info.value)
|
|
||||||
|
|
||||||
|
|
||||||
def test_connect_missing_password_raises(mocker: pytest_mock.MockerFixture) -> None:
|
|
||||||
"""Vérifie que _connect() lève ValueError si password manque.
|
|
||||||
|
|
||||||
:param mocker: Fixture pytest-mock pour le mocking.
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
from pronote_sync.config.settings import PronoteSettings
|
|
||||||
from pronote_sync.sources.pronote.client import PronoteClient
|
|
||||||
|
|
||||||
settings = PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password=None,
|
|
||||||
ent=None,
|
|
||||||
account_type="parent",
|
|
||||||
)
|
|
||||||
|
|
||||||
client = PronoteClient(settings)
|
|
||||||
|
|
||||||
with pytest.raises(ValueError) as exc_info:
|
|
||||||
client._connect()
|
|
||||||
|
|
||||||
assert "url, username et password sont requis" in str(exc_info.value)
|
|
||||||
|
|
||||||
|
|
||||||
def test_connect_student_account_type(
|
def test_connect_student_account_type(
|
||||||
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
) -> None:
|
) -> None:
|
||||||
@@ -510,7 +379,7 @@ def test_connect_student_account_type(
|
|||||||
from pronote_sync.sources.pronote.client import PronoteClient
|
from pronote_sync.sources.pronote.client import PronoteClient
|
||||||
|
|
||||||
pronote_settings_student = PronoteSettings(
|
pronote_settings_student = PronoteSettings(
|
||||||
url="https://pronote.example.com",
|
pronote_url="https://pronote.example.com",
|
||||||
username="testuser",
|
username="testuser",
|
||||||
password=SecretStr("testpass"),
|
password=SecretStr("testpass"),
|
||||||
ent="bordeaux",
|
ent="bordeaux",
|
||||||
@@ -529,57 +398,6 @@ def test_connect_student_account_type(
|
|||||||
pronotepy.ParentClient.assert_not_called() # type: ignore[attr-defined]
|
pronotepy.ParentClient.assert_not_called() # type: ignore[attr-defined]
|
||||||
|
|
||||||
|
|
||||||
def test_connect_with_ent_resolution_still_works(
|
|
||||||
mocker: pytest_mock.MockerFixture,
|
|
||||||
) -> None:
|
|
||||||
"""Vérifie que _resolve_ent est appelé et fonctionne quand ent est fourni.
|
|
||||||
|
|
||||||
Ce test valide que lorsque ent est fourni, il est toujours résolu via _resolve_ent.
|
|
||||||
|
|
||||||
:param mocker: Fixture pytest-mock pour le mocking.
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
from unittest.mock import Mock
|
|
||||||
|
|
||||||
from pronote_sync.config.settings import PronoteSettings
|
|
||||||
from pronote_sync.sources.pronote.client import PronoteClient
|
|
||||||
|
|
||||||
settings = PronoteSettings(
|
|
||||||
url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password=SecretStr("testpass"),
|
|
||||||
ent="bordeaux", # ent est fourni
|
|
||||||
account_type="parent",
|
|
||||||
)
|
|
||||||
|
|
||||||
mock_client = mocker.MagicMock()
|
|
||||||
mock_client_class = Mock(return_value=mock_client)
|
|
||||||
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
|
|
||||||
|
|
||||||
# Mock _resolve_ent to return a mock resolver
|
|
||||||
mock_resolver = Mock()
|
|
||||||
mocker.patch(
|
|
||||||
"pronote_sync.sources.pronote.client._resolve_ent",
|
|
||||||
return_value=mock_resolver,
|
|
||||||
)
|
|
||||||
|
|
||||||
client = PronoteClient(settings)
|
|
||||||
_ = client._connect()
|
|
||||||
|
|
||||||
# _resolve_ent should have been called
|
|
||||||
from pronote_sync.sources.pronote.client import _resolve_ent as resolve_ent_func
|
|
||||||
|
|
||||||
resolve_ent_func.assert_called_once_with("bordeaux") # type: ignore[attr-defined]
|
|
||||||
|
|
||||||
# ParentClient should have been called with the resolved ent
|
|
||||||
mock_client_class.assert_called_once_with(
|
|
||||||
pronote_url="https://pronote.example.com",
|
|
||||||
username="testuser",
|
|
||||||
password="testpass", # pragma: allowlist secret
|
|
||||||
ent=mock_resolver,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_messages_degraded_on_error(
|
def test_get_messages_degraded_on_error(
|
||||||
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
) -> None:
|
) -> None:
|
||||||
@@ -618,585 +436,4 @@ 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
|
||||||
|
|||||||
@@ -1,180 +0,0 @@
|
|||||||
"""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