Compare commits

..

12 Commits

Author SHA1 Message Date
344745d725 merge: corrections d'audit FIXME_M5 dans M5 blog RSS
Intègre les corrections de la revue indépendante (FIXME_M5.md) :
- Transport HTTP séparé du parsing (requests.get + feedparser.parse)
- Rejet des statuts HTTP d'erreur (raise_for_status)
- Préservation des validateurs de cache sur les chemins d'échec
- Sauvegarde atomique de l'état (tmp + Path.replace)
- Déduplication normale silencieuse

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 20:58:13 +02:00
bfae1ca87f fix(M5): corrections d'audit — transport HTTP, statuts d'erreur, cache atomique
Corrige les 5 points de l'audit FIXME_M5 :

1. (Bloquant) Sépare transport HTTP et parsing : utilise requests.get()
   avec timeout explicite et en-têtes conditionnels, puis transmet le
   contenu à feedparser.parse() — supprime le paramètre inexistant
   request_timeout qui faisait échouer toute récupération réelle.
2. Rejette les statuts HTTP 4xx/5xx via raise_for_status() avant le
   parsing.
3. Préserve les validateurs de cache (etag, last_modified) d'entrée sur
   les chemins d'échec (exception, bozo) au lieu de les écraser à None.
4. Sauvegarde atomique de BlogRSSState : écrit dans un .tmp puis
   Path.replace() pour éviter la corruption sur interruption.
5. Déduplication normale silencieuse : les GUID déjà connus sont
   ignorés sans warning ; seuls les doublons intra-flux génèrent un
   avertissement.

Tests : 49 tests (32 client + 17 state) dont 11 nouveaux couvrant
transport HTTP réel, statuts 401/404/500, préservation des validateurs,
en-têtes conditionnels, doublons intra-flux et sauvegarde atomique.
Guide : §5 bis.7.1 aligné avec le nouveau pattern transport/parsing.

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 20:58:04 +02:00
6d1a7a649f feat(M5): source blog RSS — fetch, parsing, déduplication et état persistant
Implémentation complète de la source blog RSS du collège :
- BlogRSSClient (sources/blog/rss.py) : client sans état récupérant et
  parsant le flux via feedparser, avec déduplication par ensemble de
  GUIDs connus, cache HTTP conditionnel (ETag/Last-Modified), conversion
  HTML→texte (BeautifulSoup), tri déterministe (date desc puis id asc),
  et mode dégradé (flux invalide/erreur → warning expurgé + liste vide).
- BlogRSSFetchResult (sources/blog/result.py) : résultat immuable
  contenant articles, en-têtes de cache et indicateur not_modified.
- BlogRSSState (sources/blog/state.py) : persistance JSON tolérante
  (GUIDs triés, version, ETag, Last-Modified) avec redaction des chemins
  dans les logs.
- Fixture tests/fixtures/blog_rss.xml : flux RSS 2.0 anonymisé, 3
  articles, dates fixes, ordre non chronologique.
- 38 tests unitaires (22 client + 16 state) couvrant parsing nominal,
  déduplication intra-flux, 304, bozo, erreurs réseau, non-fuite de
  secrets, tri secondaire, persistance d'état et tolérance aux fichiers
  corrompus.
- Documentation : TODO.md M5 coché, GUIDE_DEV_PYTHON.md §5 bis aligné
  avec l'API livrée (known_guids, BlogRSSFetchResult, BlogRSSState).
- Configuration : feedparser ajouté aux additional_dependencies du hook
  mypy pre-commit pour aligner l'environnement isolé avec le .venv.

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 20:23:49 +02:00
2c935ef648 docs: marquer les étapes M4 comme terminées dans TODO.md
Co-authored-by: coder/coder coder@agents.invalid
2026-09-06 17:43:15 +02:00
16f9b57dc2 fix(M4): modes explicites ical/pronotepy stricts — suppression du repli réciproque
Les modes explicites (ical, pronotepy) ne basculent plus silencieusement vers
l'autre source en cas d'échec. Seul le mode auto conserve le repli iCal →
pronotepy sur exception. Conforme au contrat de TODO.md M4 et AGENTS.md.

- fallback.py : _agenda_sources() et _homework_sources() retournent
  (source, None) en mode explicite ; docstrings mises à jour.
- test_fallback.py : 4 tests remplacés, 4 tests ajoutés (non-repli +
  empty-list-is-success), 4 tests ajustés ; :rtype: None ajouté.

Co-authored-by: coder/coder coder@agents.invalid
Co-authored-by: test-engineer/test-engineer test-engineer@agents.invalid
2026-09-06 17:42:04 +02:00
8b50731bab chore: ignorer les fichiers locaux FIXME_* et TEST_*
Co-Authored-By: Warp <agent@warp.dev>
2026-09-06 17:01:27 +02:00
373aba2ef0 docs: aligner les instructions agents sur le guide
Co-authored-by: codex/gpt-5.6-sol <codex-gpt-5.6-sol@agents.invalid>
2026-09-06 16:49:16 +02:00
8b7ca4b42b fix(M4): correction des 10 FIXME — client pronotepy, repli unique, couverture 95%
Branch fix/m4-fixme fusionnée depuis feature/m4-sources-pronote.

Corrections:
- Client pronotepy : ENT_RESOLVERS (30 callables), factory
  ParentClient/Client, pronote_url/account_type, propagation erreurs
- fallback.py : repli unique (non réciproque), filtre target_date,
  from None sur PipelineCriticalError, fetch_messages/informations
  log + re-raise
- ical.py : STATUS:CANCELLED sans CATEGORIES, blocs multi-dates
- .env.example : PRONOTE_URL, PRONOTE_ACCOUNT_TYPE
- Tests : 150 tests, couverture 93.34% (sources/pronote/ 95%)
- Docs : TODO.md M4 coché, GUIDE synchronisé

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
Co-authored-by: opencode/tech-writer <tech-writer@agents.invalid>

# Conflicts:
#	GUIDE_DEV_PYTHON.md
#	TODO.md
2026-09-06 16:37:51 +02:00
bb1f90bf5f test(M4): couverture fallback.py 81→98% (11 tests ciblés)
- _fetch_agenda_ical sans ical_url → ValueError
- _agenda_sources AUTO pronotepy seul, _homework_sources modes
  explicites + AUTO + aucune source configurée
- fetch_agenda/fetch_homework : repli échoue → PipelineCriticalError,
  repli retourne [] → warning d'ambiguité
- fetch_informations : log redact_exception + re-raise sans secrets
- Couverture pronote_sync/sources/pronote/ : 95% (≥90% requis)

Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 15:39:07 +02:00
9d9a55ed40 test+docs(M4): tests get_lessons/get_homeworks + repli unique + doc sync
Tests:
- test_pronote_client : get_lessons/get_homeworks success + error
  propagation, ENT resolution, factory parent/student, mode dégradé
  messages/informations (ConnectionError, TimeoutError)
- test_fallback : repli unique ICAL→pronotepy et PRONOTEPY→iCal,
  filtre target_date, fetch_messages log + re-raise sans secrets,
  aucune source configurée → PipelineCriticalError
- 59 tests passent (16 client + 17 fallback + 26 iCal)

Documentation:
- TODO.md : 7 items M4 cochés, contrat repli corrigé (unique, non
  réciproque) selon décision D4
- GUIDE_DEV_PYTHON.md : get_agenda_fallback → get_lessons/get_homeworks,
  pronote_url/account_type ajoutés au modèle + env vars + .env.example,
  AgendaSource StrEnum, §5.1.8 fallback contract mis à jour (repli unique)

Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
Co-authored-by: opencode/tech-writer <tech-writer@agents.invalid>
2026-09-06 15:36:42 +02:00
1b550aa818 fix(M4): pronote_url + account_type et parsing iCal (STATUS:CANCELLED + blocs multi-dates)
Configuration (D1+D3) :
- PronoteSettings : ajout pronote_url (str | None) et account_type
  (Literal student/parent, défaut parent)

Parsing iCal (Fix 4+7) :
- parse_ical : détection STATUS:CANCELLED en plus de CATEGORIES
- parse_body : list[tuple[date, str]] au lieu de dict[date, str]
  pour préserver les blocs multiples à la même date
- parse_homework_blocks : adapté aux listes

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-06 15:23:34 +02:00
d0e1f92d7f docs: aligner les jalons sur le contrat M4
Co-authored-by: codex/gpt-5.6-sol <codex-gpt-5.6-sol@agents.invalid>
2026-09-06 14:59:58 +02:00
21 changed files with 4224 additions and 880 deletions

View File

@@ -1,5 +1,7 @@
# --- Pronote ---
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
PRONOTE_URL=https://college.ent/pronote/eleve.html
PRONOTE_ACCOUNT_TYPE=parent
PRONOTE_USERNAME=parent.dupont
PRONOTE_PASSWORD=your_secure_password
PRONOTE_ENT=monbureaunumerique

4
.gitignore vendored
View File

@@ -49,6 +49,10 @@ Thumbs.db
.caldav_sync_state.json
*.state.json
# --- Local scratch / WIP files ---
FIXME_*
TEST_*
# --- Logs ---
*.log

View File

@@ -26,7 +26,7 @@ repos:
name: mypy
entry: mypy
language: python
additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.0", "types-requests>=2.31.0", "icalendar>=5.0.0", "pronotepy>=2.15.0", "responses>=0.25.0", "pytest-mock>=3.10.0"]
additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.0", "types-requests>=2.31.0", "icalendar>=5.0.0", "pronotepy>=2.15.0", "responses>=0.25.0", "pytest-mock>=3.10.0", "feedparser>=6.0.0"]
types: [python]
pass_filenames: true

View File

@@ -139,11 +139,11 @@
"type": "Hex High Entropy String",
"filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_secret": false,
"is_verified": true,
"line_number": 5117
"line_number": 5058,
"is_secret": false
}
]
},
"generated_at": "2026-09-05T21:51:55Z"
"generated_at": "2026-09-06T18:57:58Z"
}

View File

@@ -14,9 +14,10 @@
## 2. Stack technique
### Langage et dépendances
- **Python** : ≥ 3.13 (actuellement 3.14 dans `.venv/`)
- **Python** : ≥ 3.13.5 (actuellement 3.14 dans `.venv/`)
- **Dépendances principales** :
`pydantic>=2.0`, `pydantic-settings`, `icalendar`, `caldav`, `slixmpp`, `pronotepy`, `feedparser`, `requests`, `httpx`
`pydantic>=2.0`, `pydantic-settings`, `icalendar`, `caldav`, `slixmpp`, `pronotepy`,
`feedparser`, `beautifulsoup4`, `requests`, `httpx`, `openai`
### Outils de développement
- **Linter** : `ruff` (longueur de ligne : 100)
@@ -36,6 +37,7 @@
```
pronote_sync/
├── config/ # Configuration et paramètres
├── errors.py # Hiérarchie canonique des erreurs du pipeline
├── models/ # Modèles de données (Pydantic v2)
├── sources/ # Connecteurs (Pronote, iCal, etc.)
├── sync/ # Logique de synchronisation
@@ -110,6 +112,8 @@ pronote-sync --dry-run
### Architecture
- **Injection de dépendances** : Utiliser `typing.Protocol` et une **composition root** dans `pipeline/run.py`. **Interdiction** des singletons globaux.
- **Erreurs** : Conserver une seule hiérarchie dans `pronote_sync/errors.py` ; ne pas créer de
doublon dans `pipeline/steps/errors.py`.
### Style
- **Longueur de ligne** : 100 caractères maximum (configuré dans `ruff`).
@@ -123,12 +127,30 @@ pronote-sync --dry-run
- **Idempotence** : Deux exécutions identiques sans changement externe doivent produire le **même résultat**.
- **Mode dégradé** :
- Si l'IA échoue → retourner un message **sans synthèse**.
- Si `pronotepy` échoue → fallback vers le parsing **iCal**.
- Si tout échoue → lever une **erreur explicite**.
- En mode source `auto`, essayer iCal puis utiliser `pronotepy` uniquement si iCal lève une
exception.
- En mode source explicite (`ical` ou `pronotepy`), ne pas changer silencieusement de source.
- En mode `auto`, si iCal et `pronotepy` échouent → lever une erreur critique explicite.
- Une liste vide est un succès valide ; elle ne doit pas être assimilée à une panne.
### Contrat des sources Pronote
- `PRONOTE_URL` (connexion API) et `PRONOTE_ICAL_URL` (flux iCal sensible) sont deux paramètres
distincts ; ne pas déduire l'un de l'autre.
- Le compte actuellement visé est un compte parent : utiliser
`pronotepy.ParentClient(pronote_url, username, password, ent=ent_function)`.
- Résoudre le slug `PRONOTE_ENT` vers une fonction de `pronotepy.ent` au moyen d'une liste fermée.
- Exposer séparément les cours et les devoirs dans le client ; filtrer les devoirs `pronotepy` sur
la date cible.
- Les récupérations critiques agenda/devoirs propagent une erreur expurgée ; seuls les messages et
informations non critiques peuvent se dégrader en liste vide avec warning.
- Réutiliser un téléchargement/parsing iCal pour l'agenda et les devoirs pendant un même run, sans
cache global ni persistant.
### Documentation (docstrings)
- **Obligatoire** : **Toute** fonction, méthode et classe publique doit avoir une docstring.
- **Format** : Utiliser le format **Sphinx/reST** (pas Google ou NumPy) pour une compatibilité native avec Sphinx.
- **Priorité** : Les blocs historiques de `GUIDE_DEV_PYTHON.md` utilisant `Args:`/`Returns:` sont
illustratifs ; le format Sphinx/reST défini ici prévaut pour le code de production.
- **Contenu** :
- Une ligne de résumé courte (une phrase).
- Une description étendue optionnelle.
@@ -157,6 +179,11 @@ def fetch_ical(url: str) -> str:
### Règles absolues
- **Aucun secret en clair** : Ni dans le code, ni dans les logs, ni dans les erreurs, ni dans les fixtures.
- **Masquage** : Utiliser systématiquement `redact_url()`, `redact_secrets()`, et `redact_exception()` depuis `utils/redaction.py`.
- **Chaînage d'exceptions** : Ne jamais conserver comme `__cause__` ou `__context__` une exception
externe brute susceptible de contenir un secret. Journaliser la version expurgée puis utiliser
`raise ... from None`, ou chaîner une cause elle-même expurgée.
- **Tests de non-fuite** : Vérifier les messages, les logs, `__cause__`, `__context__` et le
traceback complet avec des sentinelles distinctes pour chaque secret.
### Bonnes pratiques
- **Types sécurisés** : Les mots de passe et clés API doivent utiliser `pydantic.SecretStr`.
@@ -225,14 +252,17 @@ Les rôles d'agents disponibles pour ce projet sont les suivants :
## 10. Workflow de modification
1. Lire la demande, [`TODO.md`](./TODO.md), `git status` et les fichiers concernés.
1. Lire la demande, [`TODO.md`](./TODO.md), les sections pertinentes de
[`GUIDE_DEV_PYTHON.md`](./GUIDE_DEV_PYTHON.md), `git status` et les fichiers concernés.
2. Préserver les changements existants de l'utilisateur.
3. Pour une correction, reproduire d'abord le défaut avec un test automatisé lorsque c'est raisonnable.
4. Faire une modification étroite et cohérente, en respectant les conventions du projet (idempotence, mode dégradé, repli iCal/pronotepy).
4. Faire une modification étroite et cohérente, en respectant les conventions du projet
(idempotence, modes explicites stricts et repli iCal → `pronotepy` uniquement en mode `auto`).
5. Faire vérifier le comportement par `@verifier` (ruff, mypy, pytest, bandit) et les cas d'erreur, notamment :
- Succès de la synchronisation Pronote → CalDAV/XMPP.
- Repli vers iCal en cas d'échec de `pronotepy`.
- Gestion des erreurs explicites.
- Repli d'iCal vers `pronotepy` en mode `auto`, sans repli dans les modes explicites.
- Distinction entre résultat vide et échec des sources.
- Gestion des erreurs explicites sans fuite dans les causes ou tracebacks.
6. Mettre à jour la documentation et les exemples dans le même changement si leur comportement public évolue.
7. Cocher dans [`TODO.md`](./TODO.md) uniquement les éléments entièrement réalisés et validés.
8. Terminer avec un *handoff* concis : fichiers modifiés, validations exécutées, limites et prochaine étape.
@@ -277,7 +307,8 @@ Les rôles d'agents disponibles pour ce projet sont les suivants :
Un changement est considéré comme **terminé** lorsque :
- Le cas nominal et les échecs pertinents sont testés (ex. : synchronisation réussie, repli iCal, erreurs explicites).
- Le cas nominal et les échecs pertinents sont testés (ex. : synchronisation réussie, repli
iCal → `pronotepy` en mode `auto`, modes explicites stricts, erreurs explicites).
- Les outils de validation (`ruff`, `mypy`, `pytest`, `bandit`) passent sans erreur.
- Aucun secret ni configuration locale n'apparaît dans le diff ou les fichiers suivis.
- La documentation reste cohérente avec le code (ex. : mise à jour des exemples, des contrats ou des décisions d'architecture).

File diff suppressed because it is too large Load Diff

47
TODO.md
View File

@@ -73,19 +73,23 @@ Définir tous les modèles de domaine, immuables pour les contrats, mutables pou
Récupérer et normaliser l'agenda, les devoirs et les messages Pronote, avec repli entre iCal et pronotepy.
- [ ] Créer `sources/pronote/ical.py` : `fetch_ical(url)` (HTTP via `requests`, erreurs redactées) et parsing iCal → `Lesson`/`Homework`/`SchoolEvent` (`icalendar`).
- [ ] Extraire les blocs de devoirs (`HomeworkBlock`) depuis `DESCRIPTION` et dédupliquer les devoirs (clé normalisée par date).
- [ ] Détecter les statuts (`CANCELLED`/`MOVED`) via `CATEGORIES` et `STATUS:CANCELLED`.
- [ ] Créer `sources/pronote/client.py` : client `pronotepy` (messages, informations, discussions, sondages, et devoirs en repli) avec masquage des erreurs.
- [ ] Créer `sources/pronote/fallback.py` : sélection de source selon `PRONOTE_*_SOURCE` (auto/ical/pronotepy) et `PronoteFetcher` unifiant `fetch_agenda`/`fetch_homework`/`fetch_messages`.
- [ ] Implémenter le repli : iCal échoue → pronotepy ; pronotepy échoue → iCal ; les deux échouent → `PipelineCriticalError`.
- [ ] Normaliser les UID via `utils/uid.normalize_pronote_uid` pour la stabilité des événements.
- [x] Créer `sources/pronote/ical.py` : `fetch_ical(url)` (HTTP via `requests`, erreurs redactées) et parsing iCal → `Lesson`/`Homework`/`SchoolEvent` (`icalendar`).
- [x] Extraire les blocs de devoirs (`HomeworkBlock`) depuis `DESCRIPTION` dans une séquence qui préserve plusieurs blocs à la même date ; dédupliquer ensuite via `collect_homeworks(lessons, target_date)`.
- [x] Détecter les statuts (`CANCELLED`/`MOVED`) via `CATEGORIES` et `STATUS:CANCELLED`.
- [x] Ajouter `PRONOTE_URL` à la configuration et créer `sources/pronote/client.py` autour de `pronotepy.ParentClient(pronote_url, username, password, ent=ent_function)` ; résoudre le slug ENT par liste fermée.
- [x] Exposer séparément les cours, devoirs, messages et informations dans le client `pronotepy` ; filtrer les devoirs sur `due_on == target_date`.
- [x] Créer `sources/pronote/fallback.py` : sélection de source selon `PRONOTE_*_SOURCE` (auto/ical/pronotepy) et `PronoteFetcher` unifiant `fetch_agenda`/`fetch_homework`/`fetch_messages`.
- [x] Implémenter le contrat de source : modes `ical`/`pronotepy` stricts ; mode `auto` = iCal puis repli `pronotepy` uniquement sur exception ; deux échecs en `auto``PipelineCriticalError`.
- [x] Distinguer un succès vide d'un échec : les récupérations critiques agenda/devoirs propagent une erreur expurgée ; seuls les messages/informations non critiques peuvent se dégrader en liste vide avec warning.
- [x] Normaliser les UID via `utils/uid.normalize_pronote_uid` pour la stabilité des événements.
### Critères d'acceptation
- `fetch_ical` parse `tests/fixtures/pronote-4e.ics` en leçons/devoirs/événements corrects (cours annulé détecté).
- Le client pronotepy récupère messages/devoirs (mocké).
- Le repli bascule correctement et lève une erreur critique si aucune source disponible.
- Aucun secret dans les messages d'erreur de fetch.
- `parse_ical` parse `tests/fixtures/pronote-4e.ics` en leçons/événements corrects, conserve les blocs bruts et retourne une liste de `Homework` vide ; `collect_homeworks` retourne ensuite le devoir attendu pour la date cible.
- Les cours annulés sont détectés aussi bien par catégorie que par `STATUS:CANCELLED` ; plusieurs blocs de devoirs partageant une date sont tous conservés avant déduplication.
- Le constructeur `ParentClient` est testé avec l'ordre réel de ses paramètres, l'URL Pronote et une fonction ENT autorisée.
- Le client `pronotepy` récupère messages/cours/devoirs (mocké) et ne retourne que les devoirs de la date cible.
- Le mode `auto` bascule uniquement après une exception et lève une erreur critique si les deux sources échouent ; un résultat vide reste un succès.
- Aucun secret n'apparaît dans le message, les logs, la cause, le contexte ou le traceback complet d'une erreur de source.
---
@@ -93,11 +97,11 @@ Récupérer et normaliser l'agenda, les devoirs et les messages Pronote, avec re
Récupérer le flux RSS du blog du collège, parser et dédupliquer les articles.
- [ ] Créer `sources/blog/rss.py` : `BlogRSSClient.fetch_and_parse(known_guids)` avec `feedparser` (§5 bis.7.1).
- [ ] Parser les dates (RFC 822 / ISO 8601) et convertir le HTML en texte brut (`BeautifulSoup` + `html.unescape`).
- [ ] Créer `sources/blog/state.py` (ou `sync/blog_state.py`) : `BlogRSSState` (JSON : `known_guids`, `etag`, `last_modified`).
- [ ] Implémenter la déduplication par GUID et le cache HTTP (`If-Modified-Since` / `etag`).
- [ ] Gérer un flux invalide (`bozo`) et les exceptions sans fuite de secret (retour `[]`/warning).
- [x] Créer `sources/blog/rss.py` : `BlogRSSClient.fetch_and_parse(known_guids)` avec `feedparser` (§5 bis.7.1).
- [x] Parser les dates (RFC 822 / ISO 8601) et convertir le HTML en texte brut (`BeautifulSoup` + `html.unescape`).
- [x] Créer `sources/blog/state.py` (ou `sync/blog_state.py`) : `BlogRSSState` (JSON : `known_guids`, `etag`, `last_modified`).
- [x] Implémenter la déduplication par GUID et le cache HTTP (`If-Modified-Since` / `etag`).
- [x] Gérer un flux invalide (`bozo`) et les exceptions sans fuite de secret (retour `[]`/warning).
### Critères d'acceptation
- `fetch_and_parse` renvoie les nouveaux articles triés par date décroissante, sans doublons.
@@ -128,12 +132,14 @@ Synchroniser différentiellement les événements Pronote vers le calendrier Cal
- [ ] Créer `sync/caldav.py` : `CalDAVClient` (connexion, liste/ajout/MAJ/suppression, marqueur `X-PRONOTE-SYNC-MANAGED: v1`).
- [ ] Créer `sync/state.py` : état local de sync (SQLite ou JSON) assurant l'idempotence (UID connus).
- [ ] Calculer le `CalDAVSyncPlan` (to_add / to_update / to_remove) par UID stable.
- [ ] Avant de figer le plan de sync, vérifier sur fixture anonymisée que le même cours provenant d'iCal et de `pronotepy` possède le même identifiant canonique ; corriger la normalisation à la frontière des sources si nécessaire.
- [ ] Implémenter la sync différentielle : conserver les cours annulés (`STATUS:CANCELLED`), ne pas supprimer.
- [ ] Garantir l'idempotence (2 exécutions identiques → même `CalDAVSyncResult`).
- [ ] Réutiliser `BlogRSSState` pour l'état blog si pertinent (sinon `sync/blog_state.py`).
### Critères d'acceptation
- Le plan de sync est correctement calculé (PronoteData vs état local).
- Un changement de source iCal ↔ `pronotepy` ne crée ni doublon ni suppression/ajout artificiel pour un cours équivalent.
- Un run dry-run n'écrit rien ; deux runs identiques donnent un résultat identique.
- Les événements annulés restent (`STATUS:CANCELLED`) et sont marqués `MANAGED`.
@@ -194,15 +200,17 @@ Construire et envoyer le message XMPP structuré via un compte bot dédié (mess
Composer et orchestrer toutes les étapes avec gestion d'erreurs dégradée et mode dry-run.
- [ ] Créer `pipeline/steps/errors.py` : `ErrorSeverity`, `PipelineError`, `PipelineWarning`, `PipelineCriticalError`.
- [ ] Compléter si nécessaire la hiérarchie canonique dans `pronote_sync/errors.py` (`ErrorSeverity`, `PipelineError`, `PipelineWarning`, `PipelineCriticalError`) ; ne pas créer de doublon dans `pipeline/steps/errors.py`.
- [ ] Créer les étapes `pipeline/steps/` : `fetch.py`, `normalize.py`, `compare.py`, `caldav_sync.py`, `synthesis.py`, `send.py`, `fetch_blog.py`.
- [ ] Créer `pipeline/run.py` : `PipelineRunner` (composition root) orchestrant fetch → normalize → fetch_blog → compare → caldav_sync → synthesis → send.
- [ ] Gérer les erreurs dégradées (continuer sauf critique) et renvoyer `(PronoteData, erreurs + warns)`.
- [ ] Implémenter le mode `dry_run` (aucune écriture CalDAV/XMPP).
- [ ] Câbler l'injection des dépendances (Protocol + composition root), sans singleton global.
- [ ] Réutiliser, dans une même exécution, un unique téléchargement/parsing iCal pour l'agenda et les devoirs lorsque les sources sélectionnées le permettent ; rester sur un cache local au run, sans cache global ni persistant.
### Critères d'acceptation
- Le pipeline complet s'exécute de bout en bout (mocks) dans le bon ordre.
- Une sélection iCal commune à l'agenda et aux devoirs ne déclenche qu'un téléchargement/parsing du flux par run.
- Une erreur non critique (ex : synthèse IA) n'empêche pas l'envoi XMPP.
- `dry_run=True` n'effectue aucune écriture ; aucune source disponible → erreur critique explicite.
@@ -220,7 +228,7 @@ Exposer le lancement du pipeline via une interface en ligne de commande.
### Critères d'acceptation
- `pronote-sync --dry-run --log-level DEBUG` s'exécute sans effet de bord.
- Le script console est installable (`[project.scripts]` dans `pyproject.toml`).
- Les erreurs affichées ne contiennent aucun secret.
- Les erreurs affichées ne contiennent aucun secret, y compris avec l'affichage d'un traceback complet en mode debug.
---
@@ -231,10 +239,11 @@ Couvrir l'ensemble du code par des tests sans réseau, avec fixtures anonymisée
- [ ] Créer `tests/fixtures/` : `pronote-4e.ics`, `pronote-6e.ics`, `theoretical.ics`, `theoretical.csv`, `blog_rss.xml` (anonymisés, sans `icalsecurise`).
- [ ] Créer `tests/conftest.py` : fixtures partagées (sample_lesson, sample_cancelled_lesson, sample_homework, sample_school_event, sample_message, sample_pronote_data…).
- [ ] Écrire `tests/unit/` : `test_models`, `test_parsing` (iCal), `test_uid`, `test_redaction`, `test_diff`, `test_sync`.
- [ ] Couvrir les régressions M4 : signature réelle de `ParentClient`, ENT autorisé/inconnu, erreur vs résultat vide, `STATUS:CANCELLED` sans catégorie, plusieurs devoirs à la même date, filtrage `pronotepy` sur la date cible et stabilité d'identité entre sources.
- [ ] Écrire `tests/integration/` : `test_pipeline`, `test_caldav` (mocké), `test_xmpp` (mocké).
- [ ] Écrire `tests/e2e/test_cli.py` : exécution CLI en dry-run.
- [ ] Tests sans réseau (mocks `responses`/`aioresponses`/`pytest-mock`) ; couverture ≥ 90 %.
- [ ] Ajouter un test négatif : les messages d'erreur ne fuient pas de secrets (`icalsecurise`, clés API, mots de passe).
- [ ] Ajouter un test négatif : les messages, logs, causes, contextes et tracebacks complets ne fuient pas de secrets (`icalsecurise`, clés API, mots de passe).
### Critères d'acceptation
- `pytest` passe et `pytest --cov` atteint ≥ 90 % (`fail_under = 90`).

View File

@@ -27,6 +27,8 @@ class PronoteSettings(BaseSettings):
username: str | None = None
password: SecretStr | None = None
ent: str | None = None
pronote_url: str | None = None
account_type: Literal["student", "parent"] = "parent"
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
messages_source: Literal["pronotepy"] = "pronotepy"

View File

@@ -0,0 +1,21 @@
"""Source du blog du collège : récupération et suivi du flux RSS.
Ce package expose l'API publique du connecteur du blog du collège :
- :class:`BlogRSSClient` (:mod:`pronote_sync.sources.blog.rss`) : télécharge
et parse le flux RSS, déduplique les entrées par GUID et renvoie les
nouveaux articles dans un :class:`BlogRSSFetchResult`.
- :class:`BlogRSSFetchResult` (:mod:`pronote_sync.sources.blog.result`) :
type de retour figé d'une récupération : nouveaux articles, en-têtes
HTTP de cache (``ETag``/``Last-Modified``) et indicateur ``304 Not
Modified``.
- :class:`BlogRSSState` (:mod:`pronote_sync.sources.blog.state`) : état
local persistant (GUID connus et en-têtes de cache) pour la
déduplication et les requêtes conditionnelles.
"""
from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.sources.blog.state import BlogRSSState
__all__ = ["BlogRSSClient", "BlogRSSFetchResult", "BlogRSSState"]

View File

@@ -0,0 +1,54 @@
"""Résultat de la récupération du flux RSS du blog du collège.
Ce module définit :class:`BlogRSSFetchResult`, le type de retour figé du
client RSS du blog (:mod:`pronote_sync.sources.blog`).
"""
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
from pronote_sync.models.blog import BlogArticle
class BlogRSSFetchResult(BaseModel):
"""Résultat d'une récupération du flux RSS du blog du collège.
Modèle figé (``frozen``) : les instances sont immuables après création.
Il regroupe les nouveaux articles, triés par date de publication
décroissante puis par identifiant croissant, ainsi que les en-têtes
HTTP utiles aux requêtes conditionnelles (``ETag`` et
``Last-Modified``).
:param articles: Nouveaux articles absents de ``known_guids``, triés
par date de publication décroissante puis par identifiant
croissant. Vide par défaut.
:param etag: Valeur de l'en-tête ``ETag`` de la réponse RSS, si elle
est disponible. ``None`` par défaut.
:param last_modified: Valeur de l'en-tête ``Last-Modified`` de la
réponse RSS, si elle est disponible. ``None`` par défaut.
:param not_modified: Vaut ``True`` si le serveur a répondu avec le
statut ``304 Not Modified``, ``False`` sinon.
"""
model_config = ConfigDict(frozen=True)
articles: tuple[BlogArticle, ...] = Field(
default=(),
description=(
"Nouveaux articles absents de known_guids, triés par date de "
"publication décroissante puis par identifiant croissant"
),
)
etag: str | None = Field(
default=None,
description="Valeur de l'en-tête ETag de la réponse RSS, si disponible",
)
last_modified: str | None = Field(
default=None,
description="Valeur de l'en-tête Last-Modified de la réponse RSS, si disponible",
)
not_modified: bool = Field(
default=False,
description="Vaut True si le serveur a répondu 304 Not Modified",
)

View File

@@ -0,0 +1,293 @@
"""Client de récupération et de parsing du flux RSS du blog du collège.
Ce module définit :class:`BlogRSSClient`, un client sans état qui
télécharge le flux RSS du blog via ``requests``, le parse via
``feedparser``, déduplique les entrées par GUID et les convertit en
:class:`~pronote_sync.models.blog.BlogArticle`.
Le résultat d'une récupération est un
:class:`~pronote_sync.sources.blog.result.BlogRSSFetchResult` : les
nouveaux articles (triés par date de publication décroissante, puis par
identifiant croissant) accompagnés des en-têtes HTTP ``ETag`` et
``Last-Modified`` de la réponse. Toute erreur de récupération ou de
parsing est journalisée (URL et exception rédigées) puis dégradée en
résultat vide : une liste vide est un succès valide, pas une panne.
"""
from __future__ import annotations
import logging
import re
from datetime import UTC, datetime
from html import unescape
import feedparser # type: ignore[import-untyped]
import requests
from bs4 import BeautifulSoup
from pronote_sync.models.blog import BlogArticle
from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.utils.redaction import redact_exception, redact_url
logger = logging.getLogger(__name__)
class BlogRSSClient:
"""Client de récupération et de parsing du flux RSS du blog du collège.
Client sans état : aucune E/S n'est effectuée à la construction et
aucune donnée n'est conservée entre deux appels à
:meth:`fetch_and_parse`. Toute erreur de récupération ou de parsing
est journalisée puis dégradée en résultat vide.
:param rss_url: URL du flux RSS du blog du collège.
:param timeout: Timeout HTTP en secondes (défaut : 20).
"""
def __init__(self, rss_url: str, timeout: int = 20) -> None:
"""Initialise le client RSS du blog.
Aucune opération d'E/S n'est réalisée ici : le téléchargement et
le parsing n'ont lieu qu'à l'appel de :meth:`fetch_and_parse`.
:param rss_url: URL du flux RSS du blog du collège.
:param timeout: Timeout HTTP en secondes (défaut : 20).
"""
self.rss_url = rss_url
self.timeout = timeout
def fetch_and_parse(
self,
*,
known_guids: frozenset[str] | None = None,
etag: str | None = None,
last_modified: str | None = None,
) -> BlogRSSFetchResult:
"""Télécharge et parse le flux RSS du blog en nouveaux articles.
Le flux est téléchargé par ``requests`` avec les en-têtes de
requête conditionnelle fournis (``ETag``/``Last-Modified``), puis
parsé par ``feedparser``. Si le serveur répond ``304 Not Modified``,
le résultat est vide avec
``not_modified=True`` et les en-têtes passés en entrée sont
restitués tels quels. Chaque entrée est dédupliquée par GUID,
convertie en :class:`~pronote_sync.models.blog.BlogArticle`, puis
l'ensemble est trié par date de publication décroissante puis par
identifiant croissant. Toute erreur est journalisée (URL et
exception rédigées) et dégradée en résultat vide : aucune
exception n'est propagée.
:param known_guids: Ensemble des GUID d'articles déjà traités ; les
entrées correspondantes sont ignorées. ``None`` pour tout
conserver (défaut).
:param etag: Valeur de l'en-tête ``ETag`` mémorisée pour la requête
conditionnelle, ou ``None`` (défaut).
:param last_modified: Valeur de l'en-tête ``Last-Modified`` mémorisée
pour la requête conditionnelle, ou ``None`` (défaut).
:return: Résultat de la récupération : nouveaux articles (tuple vide
si aucun nouvel article, réponse ``304`` ou erreur), en-têtes de
cache de la réponse et indicateur ``not_modified``.
:rtype: :class:`~pronote_sync.sources.blog.result.BlogRSSFetchResult`
"""
try:
# Téléchargement HTTP explicite via requests : feedparser 6.x
# n'accepte aucun paramètre de transport ; les requêtes
# conditionnelles sont gérées avec les en-têtes HTTP standards.
headers: dict[str, str] = {"user-agent": "pronote-sync"}
if etag is not None:
headers["If-None-Match"] = etag
if last_modified is not None:
headers["If-Modified-Since"] = last_modified
response = requests.get(self.rss_url, headers=headers, timeout=self.timeout)
# Réponse 304 Not Modified : rien n'a changé, on restitue les
# en-têtes mémorisés tels quels pour les conserver.
if response.status_code == 304:
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=True,
)
# Les statuts 4xx/5xx lèvent une exception HTTP, attrapée par le
# gestionnaire général et dégradée en résultat vide.
response.raise_for_status()
response_etag: str | None = response.headers.get("ETag", None)
if response_etag is None:
response_etag = response.headers.get("etag", None)
response_last_modified: str | None = response.headers.get("Last-Modified", None)
if response_last_modified is None:
response_last_modified = response.headers.get("last-modified", None)
# feedparser ne reçoit que le contenu brut de la réponse.
feed = feedparser.parse(response.content)
# Flux invalide (XML malformé, etc.) : avertissement puis résultat
# vide, sans propager l'exception brute. Les validateurs de cache
# d'entrée sont conservés : on ne fait pas confiance aux en-têtes
# d'une réponse au contenu invalide.
if getattr(feed, "bozo", None):
bozo_exception = getattr(feed, "bozo_exception", None)
if bozo_exception is not None:
logger.warning(
"Flux RSS du blog invalide (%s), ignoré : %s",
redact_exception(bozo_exception),
redact_url(self.rss_url),
)
else:
logger.warning(
"Flux RSS du blog invalide, ignoré : %s",
redact_url(self.rss_url),
)
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=False,
)
articles: list[BlogArticle] = []
# Déduplication silencieuse des GUID déjà connus (exécutions
# précédentes) et détection des doublons au sein de la réponse.
known_set = set(known_guids) if known_guids is not None else None
seen_in_feed: set[str] = set()
for entry in getattr(feed, "entries", []):
guid_source = entry.get("id") or entry.get("link")
if not guid_source:
logger.warning(
"Entrée RSS sans GUID ni lien, ignorée : %s",
redact_url(self.rss_url),
)
continue
guid = str(guid_source)
if known_set is not None and guid in known_set:
# Déduplication normale (GUID connu d'une exécution
# précédente) : aucun journal n'est nécessaire.
continue
if guid in seen_in_feed:
logger.warning(
"Entrée RSS en double dans le flux, ignorée : %s",
redact_url(self.rss_url),
)
continue
seen_in_feed.add(guid)
published_at = self._parse_date(
entry.get("published_parsed") or entry.get("pubdate_parsed")
)
if published_at is None:
logger.warning(
"Entrée RSS sans date de publication valide, ignorée : %s",
redact_url(self.rss_url),
)
continue
updated_at = self._parse_date(entry.get("updated_parsed"))
raw_content = entry.get("content")
if raw_content:
content_html = str(raw_content[0].get("value") or "")
else:
content_html = str(entry.get("description") or "")
tags = entry.get("tags")
category_value = tags[0].get("term") if tags else None
if not category_value:
category_value = entry.get("category")
category = str(category_value) if category_value else None
author_value = entry.get("author")
author = str(author_value) if author_value else None
title = str(entry.get("title") or guid)
url = str(entry.get("link") or guid)
articles.append(
BlogArticle(
id=guid,
title=title,
url=url,
published_at=published_at,
updated_at=updated_at,
category=category,
author=author,
content_html=content_html,
content_text=self._html_to_text(content_html),
)
)
# Tri stable : d'abord par identifiant croissant, puis par date de
# publication décroissante ; l'ordre par identifiant est conservé
# entre articles de même date.
articles.sort(key=lambda article: article.id)
articles.sort(key=lambda article: article.published_at, reverse=True)
return BlogRSSFetchResult(
articles=tuple(articles),
etag=response_etag,
last_modified=response_last_modified,
not_modified=False,
)
except Exception as exc:
logger.error(
"Échec de la récupération du flux RSS du blog %s : %s",
redact_url(self.rss_url),
redact_exception(exc),
)
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=False,
)
@staticmethod
def _parse_date(date_tuple: tuple[int, ...] | None) -> datetime | None:
"""Convertit un tuple de date ``struct_time`` en :class:`datetime` UTC.
:param date_tuple: Tuple horodaté au format ``time.struct_time``
(indices 0 à 5 : année, mois, jour, heure, minute, seconde), ou
``None`` si absent.
:return: Date/heure consciente du fuseau UTC, ou ``None`` si le
tuple est absent, vide ou invalide.
:rtype: datetime | None
"""
if not date_tuple:
return None
try:
return datetime(
date_tuple[0],
date_tuple[1],
date_tuple[2],
date_tuple[3],
date_tuple[4],
date_tuple[5],
tzinfo=UTC,
)
except (ValueError, IndexError):
return None
@staticmethod
def _html_to_text(html: str) -> str:
"""Convertit du HTML en texte brut nettoyé.
Le HTML est parsé avec BeautifulSoup, les balises sont remplacées
par des espaces, les entités HTML sont décodées et les suites
d'espaces sont unifiées.
:param html: Contenu HTML à convertir.
:return: Texte brut sans balises, entités décodées et espaces
unifiés ; chaîne vide si ``html`` est vide.
:rtype: str
"""
if not html:
return ""
soup = BeautifulSoup(html, "html.parser")
text = soup.get_text(separator=" ", strip=True)
text = unescape(text)
return re.sub(r"\s+", " ", text).strip()

View File

@@ -0,0 +1,174 @@
"""Gestion de l'état local du flux RSS du blog du collège.
Ce module définit :class:`BlogRSSState`, un gestionnaire d'état persistant
dans un fichier JSON local (``.blog_rss_state.json`` par défaut). Il
mémorise les identifiants (GUID) des articles déjà traités — pour la
déduplication — ainsi que les en-têtes HTTP ``ETag`` et ``Last-Modified``
de la dernière réponse — pour les requêtes conditionnelles.
La lecture et l'écriture sont tolérantes aux erreurs : un fichier absent,
corrompu ou illisible ne fait jamais échouer le pipeline ; l'état vide est
alors utilisé. La sortie JSON est déterministe (``known_guids`` triés
alphabétiquement, champ ``version`` constant).
"""
from __future__ import annotations
import json
import logging
from collections.abc import Iterable
from pathlib import Path
from pronote_sync.utils.redaction import redact_exception, redact_secrets
logger = logging.getLogger(__name__)
_STATE_VERSION = 1
class BlogRSSState:
"""Gère l'état local pour la déduplication des articles et le cache HTTP du flux RSS.
L'état regroupe l'ensemble des GUID d'articles déjà publiés
(``known_guids``) et les en-têtes de cache HTTP (``etag``,
``last_modified``). Il est chargé depuis le fichier JSON à la
construction et sauvegardé à chaque modification. Toute erreur de
lecture ou d'écriture est journalisée sans être propagée.
:param state_file: Chemin du fichier d'état JSON (``str`` ou
:class:`~pathlib.Path`). ``".blog_rss_state.json"`` par défaut.
"""
def __init__(self, state_file: Path | str = ".blog_rss_state.json") -> None:
"""Initialise le gestionnaire d'état depuis le fichier JSON.
:param state_file: Chemin du fichier d'état JSON (``str`` ou
:class:`~pathlib.Path`). ``".blog_rss_state.json"`` par défaut.
"""
self._state_file = Path(state_file)
self._known_guids: set[str] = set()
self._etag: str | None = None
self._last_modified: str | None = None
self._load()
def _load(self) -> None:
"""Charge l'état depuis le fichier JSON.
Si le fichier n'existe pas, l'état reste vide. Si le fichier est
corrompu, illisible ou que la version est absente ou différente
de 1, un avertissement est journalisé et l'état reste vide.
Aucune exception n'est propagée.
"""
if not self._state_file.exists():
return
try:
data = json.loads(self._state_file.read_text(encoding="utf-8"))
if not isinstance(data, dict) or data.get("version") != _STATE_VERSION:
logger.warning(
"Fichier d'état blog RSS %s : version absente ou non supportée, "
"démarrage avec un état vide.",
redact_secrets(str(self._state_file)),
)
return
guids_data = data.get("known_guids", [])
if isinstance(guids_data, list):
self._known_guids = {guid for guid in guids_data if isinstance(guid, str)}
etag_data = data.get("etag")
if isinstance(etag_data, str):
self._etag = etag_data
last_modified_data = data.get("last_modified")
if isinstance(last_modified_data, str):
self._last_modified = last_modified_data
except Exception as exc:
logger.warning(
"Impossible de charger le fichier d'état blog RSS %s : %s, "
"démarrage avec un état vide.",
redact_secrets(str(self._state_file)),
redact_exception(exc),
)
def _save(self) -> None:
"""Sauvegarde l'état dans le fichier JSON de manière atomique.
La sortie est déterministe : ``known_guids`` est trié
alphabétiquement et le champ ``version`` vaut 1. Le JSON est
d'abord écrit dans un fichier temporaire du même répertoire, puis
remplacé atomiquement par :meth:`~pathlib.Path.replace` afin de ne
jamais laisser un fichier partiel en cas d'interruption. En cas
d'erreur d'écriture, une erreur est journalisée sans être
propagée et le fichier temporaire est supprimé.
"""
payload = {
"version": _STATE_VERSION,
"known_guids": sorted(self._known_guids),
"etag": self._etag,
"last_modified": self._last_modified,
}
tmp_file = self._state_file.with_suffix(".tmp")
try:
with open(tmp_file, "w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2)
tmp_file.replace(self._state_file)
except Exception as exc:
logger.error(
"Impossible d'écrire le fichier d'état blog RSS %s : %s.",
redact_secrets(str(self._state_file)),
redact_exception(exc),
)
try:
tmp_file.unlink(missing_ok=True)
except Exception as cleanup_exc:
logger.debug(
"Nettoyage du fichier temporaire échoué : %s",
redact_exception(cleanup_exc),
)
def get_known_guids(self) -> frozenset[str]:
"""Renvoie une copie immuable des GUID d'articles déjà connus.
:return: Copie de type :class:`frozenset` des GUID connus.
:rtype: frozenset[str]
"""
return frozenset(self._known_guids)
def add_guids(self, guids: Iterable[str]) -> None:
"""Ajoute des GUID d'articles à l'état connu et sauvegarde.
Si l'itérable ne contient aucun GUID, l'état n'est pas modifié et
aucune sauvegarde n'est déclenchée.
:param guids: Itérable des GUID d'articles à enregistrer.
"""
new_guids = set(guids)
if not new_guids:
return
self._known_guids.update(new_guids)
self._save()
def get_cache_headers(self) -> tuple[str | None, str | None]:
"""Renvoie les en-têtes de cache HTTP mémorisés.
:return: Tuple ``(etag, last_modified)``, chaque valeur pouvant
être ``None`` si elle n'a jamais été reçue.
:rtype: tuple[str | None, str | None]
"""
return self._etag, self._last_modified
def update_cache_headers(self, etag: str | None, last_modified: str | None) -> None:
"""Met à jour les en-têtes de cache HTTP et sauvegarde.
:param etag: Nouvelle valeur de l'en-tête ``ETag``, ou ``None``
pour l'effacer.
:param last_modified: Nouvelle valeur de l'en-tête
``Last-Modified``, ou ``None`` pour l'effacer.
"""
self._etag = etag
self._last_modified = last_modified
self._save()
def clear(self) -> None:
"""Réinitialise l'état (GUID et en-têtes de cache) et sauvegarde."""
self._known_guids = set()
self._etag = None
self._last_modified = None
self._save()

View File

@@ -3,16 +3,20 @@
Ce module fournit l'encapsulation du client ``pronotepy`` pour la source
Pronote : récupération des messages des professeurs, des informations et
sondages, ainsi que des cours et devoirs en mode repli lorsque le flux
iCal échoue. Toutes les erreurs sont journalisées avec des secrets masqués.
iCal échoue. Les erreurs des méthodes dégradées (messages, informations)
sont journalisées avec des secrets masqués ; les erreurs de récupération
des cours et des devoirs se propagent pour déclencher le repli iCal.
"""
from __future__ import annotations
import logging
from datetime import date
from typing import Protocol
from typing import Any, Protocol
import pronotepy
import pronotepy.ent as pronotepy_ent
import requests
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.models.agenda import Lesson, LessonStatus
@@ -23,6 +27,69 @@ from pronote_sync.utils.redaction import redact_exception
logger = logging.getLogger(__name__)
def _get_ent_callable(name: str) -> Any:
"""Retourne le callable ``pronotepy`` associé à un nom d'ENT.
L'accès par :func:`getattr` évite les erreurs ``attr-defined`` de mypy
sur les attributs non exportés explicitement par ``pronotepy.ent``.
:param name: Nom de l'attribut dans ``pronotepy.ent``.
:return: Callable ``pronotepy`` associé.
:rtype: Any
"""
return getattr(pronotepy_ent, name)
_ENT_NAMES: list[str] = [
"monbureaunumerique",
"ent_elyco",
"bordeaux",
"ent_creuse",
"occitanie_montpellier",
"paris_classe_numerique",
"ile_de_france",
"ent_hdf",
"ac_orleans_tours",
"ac_poitiers",
"ac_rennes",
"laclasse_educonnect",
"ent77",
"ent_ecollege78",
"ent_essonne",
"val_doise",
"val_de_marne",
"ent_var",
"atrium_sud",
"laclasse_lyon",
"eclat_bfc",
"cas_arsene76",
"cas_ent27",
"cas_kosmos",
"ent_creuse_educonnect",
"ent_mayotte",
"ent_somme",
"ent_94",
"extranet_colleges_somme",
"ac_reunion",
]
_ENT_RESOLVERS: dict[str, Any] = {name: _get_ent_callable(name) for name in _ENT_NAMES}
def _resolve_ent(ent_name: str) -> Any:
"""Résout un nom d'ENT en callable ``pronotepy``.
:param ent_name: Nom de l'ENT tel que configuré (ex. ``"bordeaux"``).
:return: Callable ``pronotepy`` associé à l'ENT.
:raises ValueError: Si le nom d'ENT n'est pas reconnu.
"""
resolver = _ENT_RESOLVERS.get(ent_name)
if resolver is None:
supported = ", ".join(sorted(_ENT_RESOLVERS.keys()))
raise ValueError(f"ENT inconnu : {ent_name!r}. ENT supportés : {supported}")
return resolver
class PronoteClientProtocol(Protocol):
"""Interface du client Pronote consommée par la logique de repli."""
@@ -42,13 +109,23 @@ class PronoteClientProtocol(Protocol):
"""
...
def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]:
"""Récupère les cours et les devoirs via ``pronotepy`` (repli iCal).
def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours via ``pronotepy`` (repli iCal).
:param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse).
:return: Tuple ``(cours, devoirs)``.
:rtype: tuple[list[Lesson], list[Homework]]
:return: Liste des cours.
:rtype: list[Lesson]
"""
...
def get_homeworks(self, start: date, end: date) -> list[Homework]:
"""Récupère les devoirs via ``pronotepy``.
:param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse).
:return: Liste des devoirs.
:rtype: list[Homework]
"""
...
@@ -56,11 +133,13 @@ class PronoteClientProtocol(Protocol):
class PronoteClient:
"""Client d'accès à Pronote via ``pronotepy``.
Encapsule ``pronotepy.Client`` avec une connexion paresseuse : la
connexion n'est établie qu'à la première méthode de récupération
appelée. Les erreurs ``pronotepy.PronoteAPIError`` sont journalisées
avec des secrets masqués et les méthodes de récupération retournent
alors une valeur vide au lieu de propager l'exception.
Encapsule ``pronotepy.Client`` ou ``pronotepy.ParentClient`` selon le
type de compte, avec une connexion paresseuse : la connexion n'est
établie qu'à la première méthode de récupération appelée. Les erreurs
des méthodes dégradées (``get_messages()``, ``get_informations()``)
sont journalisées avec des secrets masqués et retournent une valeur
vide ; ``get_lessons()`` et ``get_homeworks()`` laissent les
exceptions se propager pour déclencher le repli iCal.
"""
def __init__(self, settings: PronoteSettings) -> None:
@@ -75,24 +154,39 @@ class PronoteClient:
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse).
Le client est créé une seule fois puis réutilisé pour les appels
suivants. L'erreur de connexion est relancée sans journalisation,
la méthode publique appelante étant responsable de la journaliser.
suivants. Le nom d'ENT est résolu via :func:`_resolve_ent` et 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 ``username``, ``password`` ou ``ent`` est manquant.
:raises ValueError: Si ``pronote_url``, ``username``, ``password``
ou ``ent`` est manquant, ou si l'ENT est inconnu.
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
"""
if self._client is None:
pronote_url = self._settings.pronote_url
username = self._settings.username
password = self._settings.password
ent = self._settings.ent
if username is None or password is None or ent is None:
raise ValueError("username, password et ent sont requis pour pronotepy")
try:
self._client = pronotepy.Client(username, password.get_secret_value(), ent)
except pronotepy.PronoteAPIError:
raise
if pronote_url is None or username is None or password is None or ent is None:
raise ValueError(
"pronote_url, username, password et ent sont requis pour pronotepy"
)
resolver = _resolve_ent(ent)
client_class: type[pronotepy.Client] = (
pronotepy.ParentClient
if self._settings.account_type == "parent"
else pronotepy.Client
)
self._client = client_class(
pronote_url=pronote_url,
username=username,
password=password.get_secret_value(),
ent=resolver,
)
return self._client
def get_messages(self) -> list[Message]:
@@ -122,7 +216,13 @@ class PronoteClient:
)
)
return messages
except (pronotepy.PronoteAPIError, ValueError) as exc:
except (
pronotepy.PronoteAPIError,
ValueError,
requests.RequestException,
ConnectionError,
TimeoutError,
) as exc:
logger.error(
"Échec de la récupération des messages Pronote : %s",
redact_exception(exc),
@@ -154,62 +254,86 @@ class PronoteClient:
)
)
return messages
except (pronotepy.PronoteAPIError, ValueError) as exc:
except (
pronotepy.PronoteAPIError,
ValueError,
requests.RequestException,
ConnectionError,
TimeoutError,
) as exc:
logger.error(
"Échec de la récupération des informations Pronote : %s",
redact_exception(exc),
)
return []
def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]:
"""Récupère les cours et les devoirs via ``pronotepy``.
def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours via ``pronotepy`` (repli iCal).
À utiliser uniquement si les sources iCal sont indisponibles ou en
repli automatique. Les cours annulés sont mappés sur le statut
``CANCELLED`` ; **pronotepy** ne fournissant ni la date de
Les exceptions ne sont pas attrapées : elles se propagent afin que
l'appelant puisse détecter l'échec et déclencher le repli (ou une
erreur explicite).
:param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse).
:return: Liste des cours.
:rtype: list[Lesson]
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
:raises ValueError: Si la configuration ou l'ENT est invalide.
:raises requests.RequestException: Si une requête réseau échoue.
:raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire.
"""
client = self._connect()
lessons: list[Lesson] = []
for lesson in client.lessons(start, end):
content = lesson.content
lessons.append(
Lesson(
id=lesson.id,
start=lesson.start,
end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "",
teachers=tuple(lesson.teacher_names or ()),
rooms=tuple(lesson.classrooms or ()),
group=lesson.group_name,
status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL),
content=content.description if content is not None else None,
)
)
return lessons
def get_homeworks(self, start: date, end: date) -> list[Homework]:
"""Récupère les devoirs via ``pronotepy``.
Les exceptions ne sont pas attrapées : elles se propagent afin que
l'appelant puisse détecter l'échec et déclencher le repli (ou une
erreur explicite). **pronotepy** ne fournissant ni la date de
distribution ni les professeurs des devoirs, ces champs restent
vides.
:param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse).
:return: Tuple ``(cours, devoirs)`` ; vide en cas d'erreur.
:rtype: tuple[list[Lesson], list[Homework]]
:return: Liste des devoirs.
:rtype: list[Homework]
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
:raises ValueError: Si la configuration ou l'ENT est invalide.
:raises requests.RequestException: Si une requête réseau échoue.
:raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire.
"""
try:
client = self._connect()
lessons: list[Lesson] = []
for lesson in client.lessons(start, end):
content = lesson.content
lessons.append(
Lesson(
id=lesson.id,
start=lesson.start,
end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "",
teachers=tuple(lesson.teacher_names or ()),
rooms=tuple(lesson.classrooms or ()),
group=lesson.group_name,
status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL),
content=content.description if content is not None else None,
)
client = self._connect()
homeworks: list[Homework] = []
for hw in client.homework(start, end):
homeworks.append(
Homework(
id=hw.id,
subject=hw.subject.name,
teachers=(),
assigned_on=None,
due_on=hw.date,
text=hw.description,
html=hw.description,
)
homeworks: list[Homework] = []
for hw in client.homework(start, end):
homeworks.append(
Homework(
id=hw.id,
subject=hw.subject.name,
teachers=(),
assigned_on=None,
due_on=hw.date,
text=hw.description,
html=hw.description,
)
)
return lessons, homeworks
except (pronotepy.PronoteAPIError, ValueError) as exc:
logger.error(
"Échec de la récupération de l'agenda via pronotepy : %s",
redact_exception(exc),
)
return [], []
return homeworks

View File

@@ -3,11 +3,14 @@
Ce module fournit l'enum :class:`AgendaSource`, le protocole
:class:`PronoteFetcherProtocol` consommé par le pipeline ainsi que la
classe :class:`PronoteFetcher` qui sélectionne la source selon la
configuration (``PRONOTE_AGENDA_SOURCE`` / ``PRONOTE_HOMEWORK_SOURCE``)
avec repli automatique iCal → pronotepy en mode ``auto``. Les messages
configuration (``PRONOTE_AGENDA_SOURCE`` / ``PRONOTE_HOMEWORK_SOURCE``).
Contrat strict : les modes explicites n'utilisent que la source
configurée, sans aucun repli ; seul le mode ``auto`` applique un repli
unique iCal → pronotepy, en cas d'exception uniquement. Les messages
et informations proviennent toujours de pronotepy. Toutes les erreurs
sont journalisées avec des secrets masqués via
:func:`~pronote_sync.utils.redaction.redact_exception`.
:func:`~pronote_sync.utils.redaction.redact_exception` ; les exceptions
d'origine ne sont jamais chaînées (``from None``).
"""
from __future__ import annotations
@@ -15,7 +18,7 @@ from __future__ import annotations
import logging
from datetime import date, timedelta
from enum import StrEnum
from typing import Protocol
from typing import Literal, Protocol
from pronote_sync.config.settings import Settings
from pronote_sync.errors import PipelineCriticalError
@@ -28,6 +31,8 @@ from pronote_sync.utils.redaction import redact_exception
logger = logging.getLogger(__name__)
_SourceName = Literal["ical", "pronotepy"]
class AgendaSource(StrEnum):
"""Source configurée pour la récupération de l'agenda et des devoirs."""
@@ -77,11 +82,13 @@ class PronoteFetcherProtocol(Protocol):
class PronoteFetcher:
"""Récupère les données Pronote via iCal ou pronotepy avec repli.
"""Récupère les données Pronote via iCal ou pronotepy, repli réservé au mode ``auto``.
Unifie les sources iCal et pronotepy selon la source configurée
(``agenda_source`` / ``homework_source``) : en mode ``AUTO``, le flux
iCal est essayé en premier et pronotepy sert de repli. Les messages et
(``agenda_source`` / ``homework_source``) : les modes explicites
n'utilisent que la source configurée, sans aucun repli ; seul le mode
``auto`` essaie une source primaire puis, si elle échoue, une seule
source de repli lorsqu'elle est configurée. Les messages et
informations proviennent toujours de pronotepy.
"""
@@ -105,6 +112,29 @@ class PronoteFetcher:
end = today + timedelta(days=self._settings.app.sync_future_days)
return start, end
def _is_ical_configured(self) -> bool:
"""Vérifie que la source iCal est configurée.
:return: ``True`` si ``ical_url`` est défini, ``False`` sinon.
:rtype: bool
"""
return self._settings.pronote.ical_url is not None
def _is_pronotepy_configured(self) -> bool:
"""Vérifie que la source pronotepy est entièrement configurée.
:return: ``True`` si ``pronote_url``, ``username``, ``password``
et ``ent`` sont tous définis, ``False`` sinon.
:rtype: bool
"""
pronote = self._settings.pronote
return (
pronote.pronote_url is not None
and pronote.username 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]]:
"""Récupère l'agenda depuis le flux iCal.
@@ -129,70 +159,102 @@ class PronoteFetcher:
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
:raises ValueError: Si la configuration ou l'ENT est invalide.
:raises requests.RequestException: Si une requête réseau échoue.
:raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire.
"""
start, end = self._fetch_window()
lessons, _ = self._pronote_client.get_agenda_fallback(start, end)
lessons = self._pronote_client.get_lessons(start, end)
return lessons, []
def _agenda_sources(self) -> tuple[_SourceName, _SourceName | None]:
"""Sélectionne la source primaire et le repli unique pour l'agenda.
Les modes explicites ``ICAL`` et ``PRONOTEPY`` désignent la seule
source utilisée, sans aucun repli. En mode ``AUTO``, iCal est
primaire si ``ical_url`` est configuré (repli pronotepy si la
configuration pronotepy est complète), sinon pronotepy sans repli.
:return: Tuple ``(source primaire, source de repli ou ``None``)``.
:rtype: tuple[_SourceName, _SourceName | None]
:raises PipelineCriticalError: Si aucune source n'est configurée en mode ``AUTO``.
"""
source = AgendaSource(self._settings.pronote.agenda_source)
if source is AgendaSource.ICAL:
return "ical", None
if source is AgendaSource.PRONOTEPY:
return "pronotepy", None
if self._is_ical_configured():
return "ical", "pronotepy" if self._is_pronotepy_configured() else None
if self._is_pronotepy_configured():
return "pronotepy", None
raise PipelineCriticalError(
"Impossible de récupérer l'agenda : ni la source iCal ni pronotepy n'est configurée"
) from None
def _fetch_agenda_source(self, name: _SourceName) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère l'agenda depuis la source nommée.
:param name: Nom de la source (``"ical"`` ou ``"pronotepy"``).
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
"""
if name == "ical":
return self._fetch_agenda_ical()
return self._fetch_agenda_pronotepy()
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère les cours et les événements scolaires selon la source configurée.
En mode ``AUTO``, iCal est essayé en premier et pronotepy sert de
repli ; si les deux sources échouent, une erreur critique est levée.
En mode explicite (``ical`` ou ``pronotepy``), la source désignée
est la seule tentée : si elle échoue, une erreur critique est levée
sans repli. En mode ``auto``, la source primaire est essayée en
premier puis, si elle échoue, la source de repli unique (l'autre
source, si configurée) l'est à son tour ; si la source primaire et
le repli échouent — ou si aucune source n'est configurée — une
erreur critique est levée.
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises PipelineCriticalError: Si toutes les sources configurées échouent.
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
"""
source = AgendaSource(self._settings.pronote.agenda_source)
if source is AgendaSource.ICAL:
try:
return self._fetch_agenda_ical()
except Exception as exc:
logger.error(
"Échec de la récupération iCal pour l'agenda : %s",
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer l'agenda : la source iCal a échoué"
) from exc
if source is AgendaSource.PRONOTEPY:
try:
return self._fetch_agenda_pronotepy()
except Exception as exc:
logger.error(
"Échec de la récupération pronotepy pour l'agenda : %s",
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer l'agenda : la source pronotepy a échoué"
) from exc
# Mode AUTO : essayer iCal d'abord, puis replier sur pronotepy.
primary, fallback = self._agenda_sources()
try:
return self._fetch_agenda_ical()
except Exception as exc:
logger.warning(
"Échec de la récupération iCal pour l'agenda : %s",
redact_exception(exc),
)
logger.info("Repli sur pronotepy pour l'agenda.")
try:
lessons, school_events = self._fetch_agenda_pronotepy()
return self._fetch_agenda_source(primary)
except Exception as exc:
logger.error(
"Échec de la récupération pronotepy pour l'agenda : %s",
"Échec de la récupération %s pour l'agenda : %s",
primary,
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer l'agenda : les sources iCal et pronotepy ont échoué"
) from exc
if not lessons:
logger.warning(
"Le repli pronotepy pour l'agenda a retourné un résultat vide après l'échec "
"d'iCal : impossible de distinguer une absence de cours d'un échec silencieux."
)
return lessons, school_events
if fallback is None:
raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : la source {primary} a échoué"
) from None
logger.info("Repli sur %s pour l'agenda.", fallback)
try:
lessons, school_events = self._fetch_agenda_source(fallback)
except Exception as exc:
logger.error(
"Échec de la récupération %s pour l'agenda : %s",
fallback,
redact_exception(exc),
)
raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : les sources {primary}"
f" et {fallback} ont échoué"
) from None
if not lessons:
logger.warning(
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
"de %s : impossible de distinguer une absence de cours d'un échec "
"silencieux.",
fallback,
primary,
)
return lessons, school_events
def _fetch_homework_ical(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis le flux iCal pour la date cible.
@@ -207,76 +269,113 @@ class PronoteFetcher:
lessons, _ = self._fetch_agenda_ical()
return collect_homeworks(lessons, target_date)
def _fetch_homework_pronotepy(self) -> list[Homework]:
"""Récupère les devoirs depuis pronotepy.
def _fetch_homework_pronotepy(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis pronotepy pour la date cible.
:return: Liste des devoirs.
:rtype: list[Homework]
"""
start, end = self._fetch_window()
_, homeworks = self._pronote_client.get_agenda_fallback(start, end)
return homeworks
def fetch_homework(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs selon la source configurée.
En mode ``AUTO``, iCal est essayé en premier et pronotepy sert de
repli ; si les deux sources échouent, une erreur critique est levée.
Les devoirs sont filtrés sur la date d'échéance : seuls ceux dont
``due_on`` correspond à ``target_date`` sont conservés.
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
:raises PipelineCriticalError: Si toutes les sources configurées échouent.
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
:raises ValueError: Si la configuration ou l'ENT est invalide.
:raises requests.RequestException: Si une requête réseau échoue.
:raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire.
"""
start, end = self._fetch_window()
homeworks = self._pronote_client.get_homeworks(start, end)
return [hw for hw in homeworks if hw.due_on == target_date]
def _homework_sources(self) -> tuple[_SourceName, _SourceName | None]:
"""Sélectionne la source primaire et le repli unique pour les devoirs.
Les modes explicites ``ICAL`` et ``PRONOTEPY`` désignent la seule
source utilisée, sans aucun repli. En mode ``AUTO``, iCal est
primaire si ``ical_url`` est configuré (repli pronotepy si la
configuration pronotepy est complète), sinon pronotepy sans repli.
:return: Tuple ``(source primaire, source de repli ou ``None``)``.
:rtype: tuple[_SourceName, _SourceName | None]
:raises PipelineCriticalError: Si aucune source n'est configurée en mode ``AUTO``.
"""
source = AgendaSource(self._settings.pronote.homework_source)
if source is AgendaSource.ICAL:
try:
return self._fetch_homework_ical(target_date)
except Exception as exc:
logger.error(
"Échec de la récupération iCal pour les devoirs : %s",
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer les devoirs : la source iCal a échoué"
) from exc
return "ical", None
if source is AgendaSource.PRONOTEPY:
try:
return self._fetch_homework_pronotepy()
except Exception as exc:
logger.error(
"Échec de la récupération pronotepy pour les devoirs : %s",
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer les devoirs : la source pronotepy a échoué"
) from exc
return "pronotepy", None
if self._is_ical_configured():
return "ical", "pronotepy" if self._is_pronotepy_configured() else None
if self._is_pronotepy_configured():
return "pronotepy", None
raise PipelineCriticalError(
"Impossible de récupérer les devoirs : ni la source iCal ni pronotepy n'est configurée"
) from None
# Mode AUTO : essayer iCal d'abord, puis replier sur pronotepy.
try:
def _fetch_homework_source(self, name: _SourceName, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis la source nommée.
:param name: Nom de la source (``"ical"`` ou ``"pronotepy"``).
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
"""
if name == "ical":
return self._fetch_homework_ical(target_date)
except Exception as exc:
logger.warning(
"Échec de la récupération iCal pour les devoirs : %s",
redact_exception(exc),
)
logger.info("Repli sur pronotepy pour les devoirs.")
return self._fetch_homework_pronotepy(target_date)
def fetch_homework(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs selon la source configurée.
En mode explicite (``ical`` ou ``pronotepy``), la source désignée
est la seule tentée : si elle échoue, une erreur critique est levée
sans repli. En mode ``auto``, la source primaire est essayée en
premier puis, si elle échoue, la source de repli unique (l'autre
source, si configurée) l'est à son tour ; si la source primaire et
le repli échouent — ou si aucune source n'est configurée — une
erreur critique est levée.
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
"""
primary, fallback = self._homework_sources()
try:
homeworks = self._fetch_homework_pronotepy()
return self._fetch_homework_source(primary, target_date)
except Exception as exc:
logger.error(
"Échec de la récupération pronotepy pour les devoirs : %s",
"Échec de la récupération %s pour les devoirs : %s",
primary,
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer les devoirs : les sources iCal et pronotepy ont échoué"
) from exc
if not homeworks:
logger.warning(
"Le repli pronotepy pour les devoirs a retourné un résultat vide après l'échec "
"d'iCal : impossible de distinguer une absence de devoirs d'un échec silencieux."
)
return homeworks
if fallback is None:
raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : la source {primary} a échoué"
) from None
logger.info("Repli sur %s pour les devoirs.", fallback)
try:
homeworks = self._fetch_homework_source(fallback, target_date)
except Exception as exc:
logger.error(
"Échec de la récupération %s pour les devoirs : %s",
fallback,
redact_exception(exc),
)
raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : les sources {primary}"
f" et {fallback} ont échoué"
) from None
if not homeworks:
logger.warning(
"Le repli %s pour les devoirs a retourné un résultat vide après "
"l'échec de %s : impossible de distinguer une absence de devoirs "
"d'un échec silencieux.",
fallback,
primary,
)
return homeworks
def fetch_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote (toujours via pronotepy).
@@ -284,7 +383,14 @@ class PronoteFetcher:
:return: Liste des messages.
:rtype: list[Message]
"""
return self._pronote_client.get_messages()
try:
return self._pronote_client.get_messages()
except Exception as exc:
logger.error(
"Échec de la récupération des messages : %s",
redact_exception(exc),
)
raise
def fetch_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote (toujours via pronotepy).
@@ -292,4 +398,11 @@ class PronoteFetcher:
:return: Liste des informations et sondages.
:rtype: list[Message]
"""
return self._pronote_client.get_informations()
try:
return self._pronote_client.get_informations()
except Exception as exc:
logger.error(
"Échec de la récupération des informations : %s",
redact_exception(exc),
)
raise

View File

@@ -218,21 +218,23 @@ def _parse_french_date(value: str) -> date | None:
return None
def parse_body(body: str) -> tuple[str | None, dict[date, str], dict[date, str]]:
def parse_body(body: str) -> tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]]:
"""Parse le corps HTML pour extraire contenu pédagogique et devoirs.
Le contenu est extrait de la section ``<strong>Contenu pédagogique :</strong>``.
Les devoirs à faire sont extraits des sections ``<strong>Pour le JJ/MM/AAAA :</strong>``
(dict date texte) et les devoirs donnés des sections
``<strong>Donné le JJ/MM/AAAA :</strong>`` (dict date → texte).
(liste de tuples ``(date, texte)`` dans l'ordre du flux) et les devoirs donnés
des sections ``<strong>Donné le JJ/MM/AAAA :</strong>`` (liste de tuples
``(date, texte)``). Les listes préservent tous les blocs, même lorsque plusieurs
sections partagent la même date.
:param body: Corps HTML (à partir du premier ``<strong>``).
:return: Tuple ``(contenu pédagogique, devoirs dus, devoirs donnés)``.
:rtype: tuple[str | None, dict[date, str], dict[date, str]]
:rtype: tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]]
"""
content: str | None = None
due_blocks: dict[date, str] = {}
assigned_blocks: dict[date, str] = {}
due_blocks: list[tuple[date, str]] = []
assigned_blocks: list[tuple[date, str]] = []
content_match = _CONTENT_PATTERN.search(body)
if content_match is not None:
@@ -241,34 +243,35 @@ def parse_body(body: str) -> tuple[str | None, dict[date, str], dict[date, str]]
for match in _DUE_PATTERN.finditer(body):
due_date = _parse_french_date(match.group(1))
if due_date is not None:
due_blocks[due_date] = _strip_html(match.group(2))
due_blocks.append((due_date, _strip_html(match.group(2))))
for match in _ASSIGNED_PATTERN.finditer(body):
assigned_date = _parse_french_date(match.group(1))
if assigned_date is not None:
assigned_blocks[assigned_date] = _strip_html(match.group(2))
assigned_blocks.append((assigned_date, _strip_html(match.group(2))))
return content, due_blocks, assigned_blocks
def parse_homework_blocks(
due_blocks: dict[date, str],
assigned_blocks: dict[date, str],
due_blocks: list[tuple[date, str]],
assigned_blocks: list[tuple[date, str]],
) -> tuple[HomeworkBlock, ...]:
"""Construit les :class:`HomeworkBlock` depuis les dicts de devoirs.
"""Construit les :class:`HomeworkBlock` depuis les listes de devoirs.
Les blocs dus (``kind="due"``) précèdent les blocs donnés
(``kind="assigned"``), dans l'ordre d'insertion des dicts.
(``kind="assigned"``), dans l'ordre des listes. Tous les blocs
sont préservés, y compris lorsque plusieurs partagent la même date.
:param due_blocks: Dict date → texte des devoirs à faire.
:param assigned_blocks: Dict date → texte des devoirs donnés.
:param due_blocks: Liste de tuples ``(date, texte)`` des devoirs à faire.
:param assigned_blocks: Liste de tuples ``(date, texte)`` des devoirs donnés.
:return: Tuple de blocs de devoirs pour le cours.
:rtype: tuple[HomeworkBlock, ...]
"""
blocks: list[HomeworkBlock] = []
for due_date, text in due_blocks.items():
for due_date, text in due_blocks:
blocks.append(HomeworkBlock(kind="due", date=due_date, text=text, html=text))
for assigned_date, text in assigned_blocks.items():
for assigned_date, text in assigned_blocks:
blocks.append(HomeworkBlock(kind="assigned", date=assigned_date, text=text, html=text))
return tuple(blocks)
@@ -360,9 +363,10 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School
Les VEVENT de vacances/congés (tout le jour) deviennent des
:class:`SchoolEvent` de type ``holiday``. Les VEVENT horodatés
deviennent des :class:`Lesson` dont le statut dérive de la
catégorie (``Cours - Cours annulé`` → ``CANCELLED``,
``Cours - Cours déplacé`` → ``MOVED``). Les UID sont normalisés ;
deviennent des :class:`Lesson` dont le statut dérive de la propriété
``STATUS`` (``CANCELLED`` → ``CANCELLED``) et de la catégorie
(``Cours - Cours annulé`` → ``CANCELLED``, ``Cours - Cours déplacé`` →
``MOVED``). Les UID sont normalisés ;
un événement sans UID reçoit un UID déterministe généré à partir
de ses champs clés (début, fin, matière, enseignants, salles, groupe).
@@ -416,11 +420,14 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School
if not isinstance(start, datetime) or not isinstance(end, datetime):
continue
status = LessonStatus.NORMAL
if "Cours - Cours annulé" in categories:
status_obj = component.get("status")
status_value = str(status_obj).strip().upper() if status_obj is not None else ""
if status_value == "CANCELLED" or "Cours - Cours annulé" in categories:
status = LessonStatus.CANCELLED
elif "Cours - Cours déplacé" in categories:
status = LessonStatus.MOVED
else:
status = LessonStatus.NORMAL
description = component.get("description")
description_str = str(description) if description is not None else ""

47
tests/fixtures/blog_rss.xml vendored Normal file
View File

@@ -0,0 +1,47 @@
<?xml version="1.0" encoding="utf-8"?>
<rss version="2.0" xmlns:content="http://purl.org/rss/1.0/modules/content/" xmlns:dc="http://purl.org/dc/elements/1.1/">
<channel>
<title>Blog du collège Les Mimosas</title>
<link>https://example.com/blog/</link>
<description>Actualités et informations du collège Les Mimosas</description>
<language>fr-FR</language>
<item>
<title>Information générale</title>
<link>https://example.com/blog/?p=1003</link>
<guid isPermaLink="false">https://example.com/blog/?p=1003</guid>
<pubDate>Wed, 12 Aug 2026 08:00:00 +0000</pubDate>
<description>Information générale à destination des familles.</description>
<content:encoded><![CDATA[
<p>La vie scolaire rappelle aux familles que les billets de cantine sont à commander avant le vendredi soir.</p>
<p>Pour toute question, consultez la page <a href="https://example.com/blog/cantine/">cantines et restauration</a> du site.</p>
]]></content:encoded>
</item>
<item>
<title>Réunion de rentrée</title>
<link>https://example.com/blog/?p=1001</link>
<guid isPermaLink="false">https://example.com/blog/?p=1001</guid>
<pubDate>Mon, 10 Aug 2026 09:00:11 +0000</pubDate>
<category>Administration</category>
<dc:creator>M. Dupont</dc:creator>
<description>Réunion de rentrée des parents d'élèves.</description>
<content:encoded><![CDATA[
<p>La réunion de rentrée des parents d'élèves se tiendra le mardi 15 septembre à 18 h 00 dans la salle polyvalente.</p>
<p>L'équipe pédagogique y présentera le projet d'établissement et le calendrier des conseils de classe. Un temps d'échange est prévu avec les professeurs principaux.</p>
<p>Merci de confirmer votre présence en remplissant le <a href="https://example.com/blog/reunion-rentree-inscription/">formulaire d'inscription</a> avant le 10 septembre.</p>
]]></content:encoded>
</item>
<item>
<title>Sortie pédagogique au musée</title>
<link>https://example.com/blog/?p=1002</link>
<guid isPermaLink="false">https://example.com/blog/?p=1002</guid>
<pubDate>Tue, 11 Aug 2026 14:30:00 +0000</pubDate>
<category>Pédagogie</category>
<description>Sortie pédagogique des élèves de 4e au musée d'art moderne.</description>
<content:encoded><![CDATA[
<p>Les élèves de 4e se rendront au musée d'art moderne le jeudi 8 octobre dans le cadre du cours d'arts plastiques.</p>
<p>La visite guidée portera sur la période impressionniste. Les élèves devront apporter un carnet de croquis et leur pique-nique.</p>
<p>Le détail de l'organisation figure dans la <a href="https://example.com/blog/sortie-musee-autorisation/">note d'autorisation</a> à retourner signée avant le 25 septembre.</p>
]]></content:encoded>
</item>
</channel>
</rss>

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,350 @@
"""Tests unitaires pour le gestionnaire d'état du flux RSS du blog.
Ce module valide le comportement de :class:`BlogRSSState` dans
:mod:`pronote_sync.sources.blog.state`. Les tests couvrent :
- La persistance des GUID connus et des en-têtes de cache HTTP,
- La tolérance aux erreurs (fichier absent, corrompu, version incompatible),
- Le tri alphabétique des GUID lors de la sauvegarde,
- La réinitialisation complète de l'état.
Tous les tests utilisent des fichiers temporaires via la fixture ``tmp_path``.
"""
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch
import pytest
from pronote_sync.sources.blog.state import BlogRSSState
def test_state_file_absent_empty_state(tmp_path: Path) -> None:
"""Vérifie qu'un fichier d'état absent initialise un état vide.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "nonexistent.json"
state = BlogRSSState(state_file)
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
def test_add_guids_persists(tmp_path: Path) -> None:
"""Vérifie que l'ajout de GUID persiste dans le fichier JSON.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["guid-2", "guid-1", "guid-3"])
assert state.get_known_guids() == frozenset({"guid-1", "guid-2", "guid-3"})
# Vérification du contenu du fichier
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["known_guids"] == ["guid-1", "guid-2", "guid-3"]
def test_add_guids_empty_noop(tmp_path: Path) -> None:
"""Vérifie que l'ajout d'une liste vide ne modifie pas le fichier.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
# Ajout initial de GUID
state.add_guids(["guid-1"])
original_content = state_file.read_text(encoding="utf-8")
# Ajout d'une liste vide
state.add_guids([])
# Vérification que le fichier n'a pas été modifié (comparaison par contenu)
assert state_file.read_text(encoding="utf-8") == original_content
def test_state_load_persisted_guids(tmp_path: Path) -> None:
"""Vérifie que les GUID persistés sont rechargés dans une nouvelle instance.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
# Création et sauvegarde de l'état initial
state1 = BlogRSSState(state_file)
state1.add_guids(["guid-1", "guid-2"])
# Création d'une nouvelle instance avec le même fichier
state2 = BlogRSSState(state_file)
assert state2.get_known_guids() == frozenset({"guid-1", "guid-2"})
def test_state_load_cache_headers(tmp_path: Path) -> None:
"""Vérifie que les en-têtes de cache persistés sont rechargés.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
# Création et sauvegarde des en-têtes de cache
state1 = BlogRSSState(state_file)
state1.update_cache_headers("etag-123", "Wed, 01 Sep 2026 GMT")
# Création d'une nouvelle instance avec le même fichier
state2 = BlogRSSState(state_file)
assert state2.get_cache_headers() == ("etag-123", "Wed, 01 Sep 2026 GMT")
def test_corrupt_json_warning(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie qu'un fichier JSON corrompu déclenche un avertissement et initialise un état vide.
: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"):
state = BlogRSSState(state_file)
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
assert "Impossible de charger le fichier d'état blog RSS" in caplog.text
def test_wrong_version_warning(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie qu'une version incompatible déclenche un avertissement et initialise un état vide.
: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": 99, "known_guids": ["x"], "etag": None, "last_modified": None}),
encoding="utf-8",
)
with caplog.at_level("WARNING"):
state = BlogRSSState(state_file)
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
assert "version absente ou non supportée" in caplog.text
def test_missing_version_warning(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie qu'un fichier sans champ version déclenche un avertissement et initialise un état vide.
: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({"known_guids": ["x"], "etag": None, "last_modified": None}),
encoding="utf-8",
)
with caplog.at_level("WARNING"):
state = BlogRSSState(state_file)
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
assert "version absente ou non supportée" in caplog.text
def test_known_guids_sorted_on_save(tmp_path: Path) -> None:
"""Vérifie que les GUID sont triés alphabétiquement lors de la sauvegarde.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["c-guid", "a-guid", "b-guid"])
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["known_guids"] == ["a-guid", "b-guid", "c-guid"]
def test_clear_resets_state(tmp_path: Path) -> None:
"""Vérifie que la méthode clear réinitialise complètement l'état.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
# Ajout de GUID et d'en-têtes de cache
state.add_guids(["guid-1", "guid-2"])
state.update_cache_headers("etag-123", "Wed, 01 Sep 2026 GMT")
# Réinitialisation
state.clear()
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
# Vérification du contenu du fichier
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["known_guids"] == []
assert saved_data["etag"] is None
assert saved_data["last_modified"] is None
def test_clear_persists_to_file(tmp_path: Path) -> None:
"""Vérifie que la réinitialisation est persistée dans le fichier.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
# Création, ajout de données et réinitialisation
state1 = BlogRSSState(state_file)
state1.add_guids(["guid-1"])
state1.update_cache_headers("etag-123", "Wed, 01 Sep 2026 GMT")
state1.clear()
# Création d'une nouvelle instance avec le même fichier
state2 = BlogRSSState(state_file)
assert state2.get_known_guids() == frozenset()
assert state2.get_cache_headers() == (None, None)
def test_str_path_converted_to_path(tmp_path: Path) -> None:
"""Vérifie qu'un chemin de type str est converti en Path.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = str(tmp_path / "state.json")
state = BlogRSSState(state_file)
state.add_guids(["guid-1"])
assert Path(state_file).exists()
def test_update_cache_headers_none_values(tmp_path: Path) -> None:
"""Vérifie que la mise à jour avec des valeurs None fonctionne correctement.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.update_cache_headers(None, None)
assert state.get_cache_headers() == (None, None)
# Vérification du contenu du fichier
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["etag"] is None
assert saved_data["last_modified"] is None
def test_add_guids_multiple_calls(tmp_path: Path) -> None:
"""Vérifie que plusieurs appels à add_guids accumulent les GUID.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["guid-1"])
state.add_guids(["guid-2"])
assert state.get_known_guids() == frozenset({"guid-1", "guid-2"})
def test_version_in_saved_file(tmp_path: Path) -> None:
"""Vérifie que le champ version est présent dans le fichier sauvegardé.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["guid-1"])
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["version"] == 1
def test_get_known_guids_returns_frozenset(tmp_path: Path) -> None:
"""Vérifie que get_known_guids retourne un frozenset.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["guid-1", "guid-2"])
result = state.get_known_guids()
assert type(result) is frozenset
def test_atomic_save_preserves_on_error(tmp_path: Path) -> None:
"""Vérifie que l'état original est préservé en cas d'erreur lors de la sauvegarde atomique.
Si une erreur survient pendant le remplacement atomique du fichier,
le fichier original doit rester intact et le fichier temporaire doit être nettoyé.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
# Créer un état initial avec des GUID
state = BlogRSSState(state_file)
state.add_guids(["original-guid-1", "original-guid-2"])
# Lire le contenu original
original_content = state_file.read_text(encoding="utf-8")
# Mock Path.replace pour simuler une erreur pendant le remplacement atomique
with patch.object(Path, "replace") as mock_replace:
mock_replace.side_effect = OSError("Simulated atomic replace failure")
# Essayer d'ajouter de nouveaux GUID, ce qui déclenchera _save()
state.add_guids(["new-guid"])
# Vérifier que le fichier original est toujours intact
assert state_file.read_text(encoding="utf-8") == original_content
# Vérifier que le fichier temporaire a été nettoyé
tmp_file = state_file.with_suffix(".tmp")
assert not tmp_file.exists()
# Vérifier que l'état en mémoire n'a pas été modifié (car la sauvegarde a échoué)
# Note: En réalité, l'état en mémoire est modifié mais pas persistant
# C'est le fichier qui doit rester intact
assert state.get_known_guids() == frozenset({"original-guid-1", "original-guid-2", "new-guid"})
# Ensure trailing newline

View File

@@ -40,9 +40,8 @@ if TYPE_CHECKING:
class _MockPronoteClientProtocol(Protocol):
def get_messages(self) -> list[Message]: ...
def get_informations(self) -> list[Message]: ...
def get_agenda_fallback(
self, start: date, end: date
) -> tuple[list[Lesson], list[Homework]]: ...
def get_lessons(self, start: date, end: date) -> list[Lesson]: ...
def get_homeworks(self, start: date, end: date) -> list[Homework]: ...
@pytest.fixture(name="mock_settings")
@@ -54,6 +53,7 @@ def fixture_mock_settings() -> Settings:
"""
return Settings(
pronote=PronoteSettings(
pronote_url="https://pronote.example.com",
ical_url=SecretStr("file:///fake/ical.ics"),
agenda_source="auto",
homework_source="auto",
@@ -85,6 +85,7 @@ def test_fetch_agenda_ical_mode(mock_fetcher: PronoteFetcher) -> None:
: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)
@@ -128,11 +129,12 @@ def test_fetch_agenda_ical_mode(mock_fetcher: PronoteFetcher) -> None:
def test_fetch_agenda_pronotepy_mode(mock_fetcher: PronoteFetcher) -> None:
"""Test la récupération de l'agenda en mode source pronotepy.
On mock ``get_agenda_fallback`` du client pour retourner des cours.
On mock ``get_lessons`` du client pour retourner des cours.
On vérifie que le fetcher retourne ces cours (événements scolaires vides).
: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)
@@ -151,14 +153,14 @@ def test_fetch_agenda_pronotepy_mode(mock_fetcher: PronoteFetcher) -> None:
]
client = MagicMock()
client.get_agenda_fallback.return_value = (lessons, [])
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_agenda_fallback.assert_called_once()
client.get_lessons.assert_called_once()
def test_fetch_agenda_auto_ical_success(mock_fetcher: PronoteFetcher) -> None:
@@ -168,6 +170,7 @@ def test_fetch_agenda_auto_ical_success(mock_fetcher: PronoteFetcher) -> None:
: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)
@@ -204,6 +207,7 @@ def test_fetch_agenda_auto_fallback_to_pronotepy(mock_fetcher: PronoteFetcher) -
: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)
@@ -228,13 +232,13 @@ def test_fetch_agenda_auto_fallback_to_pronotepy(mock_fetcher: PronoteFetcher) -
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_agenda_fallback.return_value = (lessons, [])
client.get_lessons.return_value = lessons
mock_fetcher._pronote_client = client
result_lessons, _ = mock_fetcher.fetch_agenda()
assert result_lessons == lessons
client.get_agenda_fallback.assert_called_once()
client.get_lessons.assert_called_once()
def test_fetch_agenda_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
@@ -244,7 +248,11 @@ def test_fetch_agenda_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
# Disable pronotepy so fallback is None
mock_fetcher._settings.pronote.pronote_url = 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,
@@ -257,19 +265,22 @@ def test_fetch_agenda_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "iCal et pronotepy" in str(exc_info.value)
assert "la source ical a échoué" in str(exc_info.value)
def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode ical : échec → PipelineCriticalError masquée.
On mock iCal pour échouer. On vérifie que l'erreur brute est masquée dans la levée.
On mock iCal pour échouer et désactive pronotepy pour que le repli soit None.
On vérifie que l'erreur brute est masquée dans la levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
# Override settings to use ical mode explicitly
# Override settings to use ical mode explicitly and disable fallback
mock_fetcher._settings.pronote.agenda_source = "ical"
mock_fetcher._settings.pronote.pronote_url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
@@ -283,7 +294,7 @@ def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "Impossible de récupérer l'agenda : la source iCal a échoué" in str(exc_info.value)
assert "Impossible de récupérer l'agenda : la source ical a échoué" in str(exc_info.value)
# Vérifie que le message ne contient pas de secret
assert "file:///fake/ical.ics" not in str(exc_info.value)
@@ -291,15 +302,19 @@ def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
def test_fetch_agenda_pronotepy_mode_failure(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode pronotepy : échec → PipelineCriticalError masquée.
On mock pronotepy pour échouer. On vérifie que l'erreur est masquée dans la levée.
On mock pronotepy pour échouer et désactive iCal pour que le repli soit None.
On vérifie que l'erreur est masquée dans la levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
# Override settings to use pronotepy mode explicitly
# Override settings to use pronotepy mode explicitly and disable fallback
mock_fetcher._settings.pronote.agenda_source = "pronotepy"
mock_fetcher._settings.pronote.ical_url = None
client = MagicMock()
client.get_lessons.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
@@ -316,6 +331,7 @@ def test_fetch_homework_ical_mode(mock_fetcher: PronoteFetcher) -> None:
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 5)
@@ -377,6 +393,7 @@ def test_fetch_homework_auto_fallback(mock_fetcher: PronoteFetcher) -> None:
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
homeworks = [
@@ -399,14 +416,14 @@ def test_fetch_homework_auto_fallback(mock_fetcher: PronoteFetcher) -> None:
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_agenda_fallback.return_value = ([], homeworks)
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_agenda_fallback.assert_called_once()
client.get_homeworks.assert_called_once()
def test_fetch_homework_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
@@ -416,9 +433,13 @@ def test_fetch_homework_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
# Disable pronotepy so fallback is None
mock_fetcher._settings.pronote.pronote_url = 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,
@@ -431,7 +452,7 @@ def test_fetch_homework_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_homework(target_date)
assert "iCal et pronotepy" in str(exc_info.value)
assert "la source ical a échoué" in str(exc_info.value)
def test_fetch_messages(mock_fetcher: PronoteFetcher) -> None:
@@ -442,6 +463,7 @@ def test_fetch_messages(mock_fetcher: PronoteFetcher) -> None:
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
messages = [
Message(
@@ -457,7 +479,6 @@ def test_fetch_messages(mock_fetcher: PronoteFetcher) -> None:
client = MagicMock()
client.get_messages.return_value = messages
client.get_informations.return_value = []
client.get_agenda_fallback.return_value = ([], [])
mock_fetcher._pronote_client = client
result = mock_fetcher.fetch_messages()
@@ -473,6 +494,7 @@ def test_fetch_informations(mock_fetcher: PronoteFetcher) -> None:
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
infos = [
Message(
@@ -488,7 +510,6 @@ def test_fetch_informations(mock_fetcher: PronoteFetcher) -> None:
client = MagicMock()
client.get_informations.return_value = infos
client.get_messages.return_value = []
client.get_agenda_fallback.return_value = ([], [])
mock_fetcher._pronote_client = client
result = mock_fetcher.fetch_informations()
@@ -507,7 +528,11 @@ def test_no_secrets_in_error_messages(
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
:rtype: None
"""
# Disable pronotepy so fallback is None to trigger PipelineCriticalError
mock_fetcher._settings.pronote.pronote_url = 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,
@@ -529,4 +554,636 @@ def test_no_secrets_in_error_messages(
assert "icalsecurise=REDACTED" in caplog.text or "icalsecurise" not in caplog.text
def test_fetch_agenda_ical_mode_does_not_fallback_to_pronotepy(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode ICAL : échec iCal, PAS de repli sur pronotepy.
On mock iCal pour échouer. On vérifie que pronotepy n'est PAS appelé
et qu'une PipelineCriticalError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
# Override settings to use ical mode explicitly
mock_fetcher._settings.pronote.agenda_source = "ical"
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()
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "la source ical a échoué" in str(exc_info.value)
# Vérifie que pronotepy n'a PAS été appelé
client.get_lessons.assert_not_called()
def test_fetch_agenda_pronotepy_mode_does_not_fallback_to_ical(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode PRONOTEPY : échec pronotepy, PAS de repli sur iCal.
On mock pronotepy pour échouer. On vérifie que iCal n'est PAS appelé
et qu'une PipelineCriticalError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
# Override settings to use pronotepy mode explicitly
mock_fetcher._settings.pronote.agenda_source = "pronotepy"
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,
):
client = MagicMock()
client.get_lessons.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "la source pronotepy a échoué" in str(exc_info.value)
# Vérifie que iCal n'a PAS été appelé
m_fetch_ical.assert_not_called()
m_parse_ical.assert_not_called()
def test_fetch_agenda_no_source_configured_raises(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode AUTO : aucune source configurée → PipelineCriticalError.
On désactive les deux sources. On vérifie que l'erreur critique est levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
# Disable both sources
mock_fetcher._settings.pronote.ical_url = None
mock_fetcher._settings.pronote.pronote_url = None
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
def test_fetch_agenda_ical_mode_empty_list_is_success(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode ICAL : liste vide = succès valide, sans repli.
On mock iCal pour retourner une liste vide. On vérifie que le résultat
est ([], []) et qu'aucun repli n'est tenté.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.agenda_source = "ical"
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.return_value = "BEGIN:VCALENDAR\n..."
m_parse_ical.return_value = ([], [], [])
client = MagicMock()
mock_fetcher._pronote_client = client
result_lessons, result_events = mock_fetcher.fetch_agenda()
assert result_lessons == []
assert result_events == []
# Vérifie que pronotepy n'a PAS été appelé
client.get_lessons.assert_not_called()
def test_fetch_agenda_pronotepy_mode_empty_list_is_success(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode PRONOTEPY : liste vide = succès valide, sans repli.
On mock pronotepy pour retourner une liste vide. On vérifie que le résultat
est ([], []) et qu'aucun repli n'est tenté.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.agenda_source = "pronotepy"
client = MagicMock()
client.get_lessons.return_value = []
mock_fetcher._pronote_client = client
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,
):
result_lessons, result_events = mock_fetcher.fetch_agenda()
assert result_lessons == []
assert result_events == []
# Vérifie que iCal n'a PAS été appelé
m_fetch_ical.assert_not_called()
m_parse_ical.assert_not_called()
def test_fetch_homework_ical_mode_empty_list_is_success(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode ICAL des devoirs : liste vide = succès valide, sans repli.
On mock iCal pour retourner une liste vide. On vérifie que le résultat
est [] et qu'aucun repli n'est tenté.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
mock_fetcher._settings.pronote.homework_source = "ical"
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.return_value = "BEGIN:VCALENDAR\n..."
m_parse_ical.return_value = ([], [], [])
m_collect.return_value = []
client = MagicMock()
mock_fetcher._pronote_client = client
result = mock_fetcher.fetch_homework(target_date)
assert result == []
# Vérifie que pronotepy n'a PAS été appelé
client.get_homeworks.assert_not_called()
def test_fetch_homework_pronotepy_mode_empty_list_is_success(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode PRONOTEPY des devoirs : liste vide = succès valide, sans repli.
On mock pronotepy pour retourner une liste vide. On vérifie que le résultat
est [] et qu'aucun repli n'est tenté.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
mock_fetcher._settings.pronote.homework_source = "pronotepy"
client = MagicMock()
client.get_homeworks.return_value = []
mock_fetcher._pronote_client = client
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,
):
result = mock_fetcher.fetch_homework(target_date)
assert result == []
# Vérifie que iCal n'a PAS été appelé
m_fetch_ical.assert_not_called()
m_parse_ical.assert_not_called()
def test_fetch_homework_filters_by_target_date(mock_fetcher: PronoteFetcher) -> None:
"""Test que les devoirs sont filtrés par date cible en mode pronotepy.
On mock pronotepy pour retourner des devoirs avec différentes dates d'échéance.
On vérifie que seul le devoir correspondant à la date cible est retourné.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
other_date = date(2025, 9, 11)
homeworks = [
Homework(
id="hw1",
subject="Maths",
teachers=(),
assigned_on=None,
due_on=target_date,
text="Devoir pour aujourd'hui",
html="<p>Devoir pour aujourd'hui</p>",
),
Homework(
id="hw2",
subject="Physique",
teachers=(),
assigned_on=None,
due_on=other_date,
text="Devoir pour demain",
html="<p>Devoir pour demain</p>",
),
]
client = MagicMock()
client.get_homeworks.return_value = homeworks
mock_fetcher._pronote_client = client
result = mock_fetcher.fetch_homework(target_date)
assert len(result) == 1
assert result[0].id == "hw1"
assert result[0].due_on == target_date
client.get_homeworks.assert_called_once()
def test_fetch_homework_ical_mode_does_not_fallback_to_pronotepy(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode ICAL des devoirs : échec iCal, PAS de repli sur pronotepy.
On mock iCal pour échouer. On vérifie que pronotepy n'est PAS appelé
et qu'une PipelineCriticalError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
mock_fetcher._settings.pronote.homework_source = "ical"
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()
mock_fetcher._pronote_client = client
m_collect.side_effect = OSError("collect error")
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_homework(target_date)
assert "la source ical a échoué" in str(exc_info.value)
# Vérifie que pronotepy n'a PAS été appelé
client.get_homeworks.assert_not_called()
def test_fetch_homework_pronotepy_mode_does_not_fallback_to_ical(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode PRONOTEPY des devoirs : échec pronotepy, PAS de repli sur iCal.
On mock pronotepy pour échouer. On vérifie que iCal n'est PAS appelé
et qu'une PipelineCriticalError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
mock_fetcher._settings.pronote.homework_source = "pronotepy"
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,
):
client = MagicMock()
client.get_homeworks.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_homework(target_date)
assert "la source pronotepy a échoué" in str(exc_info.value)
# Vérifie que iCal n'a PAS été appelé
m_fetch_ical.assert_not_called()
m_parse_ical.assert_not_called()
def test_fetch_messages_logs_and_re_raises(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que fetch_messages journalise et relance les exceptions.
On mock get_messages pour lever une exception contenant une URL secrète.
On vérifie que l'exception est relancée et que le log masqué ne contient pas l'URL.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
:rtype: None
"""
client = MagicMock()
error_msg = "Erreur Pronote : impossible de récupérer les messages https://pronote.example.com/messages?token=SECRET_TOKEN_456"
client.get_messages.side_effect = OSError(error_msg)
mock_fetcher._pronote_client = client
with pytest.raises(OSError) as exc_info:
mock_fetcher.fetch_messages()
assert exc_info.value is client.get_messages.side_effect
assert "SECRET_TOKEN_456" not in caplog.text
assert (
"pronote.example.com/messages?token=REDACTED" in caplog.text
or "pronote.example.com/messages" in caplog.text
)
def test_fetch_agenda_ical_url_none_raises_value_error(mock_fetcher: PronoteFetcher) -> None:
"""Test que _fetch_agenda_ical lève ValueError si ical_url est None.
On désactive ical_url et on appelle _fetch_agenda_ical(). On vérifie que
l'erreur ValueError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.ical_url = None
with pytest.raises(ValueError, match="PRONOTE_ICAL_URL est requis pour la source iCal"):
mock_fetcher._fetch_agenda_ical()
def test_agenda_sources_auto_only_pronotepy_configured(mock_fetcher: PronoteFetcher) -> None:
"""Test _agenda_sources en mode AUTO avec seulement pronotepy configuré.
On désactive ical_url mais on garde pronotepy configuré. On vérifie que la
source primaire est pronotepy et qu'il n'y a pas de repli.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.agenda_source = "auto"
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._agenda_sources()
assert primary == "pronotepy"
assert fallback is None
def test_fetch_agenda_fallback_both_fail_raises_pipeline_critical_error(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test que fetch_agenda lève PipelineCriticalError en mode explicite ical.
On mock ICAL pour échouer. On vérifie que l'erreur critique est levée
et que pronotepy n'est PAS appelé (pas de repli en mode explicite).
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.agenda_source = "ical"
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.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
# En mode explicite, seul l'échec de la source primaire est rapporté
assert "la source ical a échoué" in str(exc_info.value)
# Vérifie que pronotepy n'a PAS été appelé (pas de repli en mode explicite)
client.get_lessons.assert_not_called()
def test_fetch_agenda_auto_fallback_returns_empty_logs_warning(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que fetch_agenda 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.
Ce test utilise le mode AUTO pour tester le comportement de repli.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.agenda_source = "auto"
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.PronoteFetcher._fetch_agenda_pronotepy"
) as m_fetch_pronotepy,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
mock_fetcher._pronote_client = client
m_fetch_pronotepy.return_value = ([], []) # Empty result from fallback
lessons, events = mock_fetcher.fetch_agenda()
assert lessons == []
assert events == []
assert "a retourné un résultat vide après l'échec de ical" in caplog.text
def test_homework_sources_explicit_ical_mode_strict(mock_fetcher: PronoteFetcher) -> None:
"""Test _homework_sources en mode ICAL : strict, sans repli.
On vérifie que la source primaire est ical et que le repli est TOUJOURS None,
même si pronotepy est configuré.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.homework_source = "ical"
# With pronotepy configured
primary, fallback = mock_fetcher._homework_sources()
assert primary == "ical"
assert fallback is None
# Without pronotepy configured
mock_fetcher._settings.pronote.pronote_url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "ical"
assert fallback is None
def test_homework_sources_explicit_pronotepy_mode_strict(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test _homework_sources en mode PRONOTEPY : strict, sans repli.
On vérifie que la source primaire est pronotepy et que le repli est TOUJOURS None,
même si iCal est configuré.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.homework_source = "pronotepy"
# With ical configured
primary, fallback = mock_fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback is None
# Without ical configured
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback is None
def test_homework_sources_auto_only_pronotepy_configured(mock_fetcher: PronoteFetcher) -> None:
"""Test _homework_sources en mode AUTO avec seulement pronotepy configuré.
On désactive ical_url mais on garde pronotepy configuré. On vérifie que la
source primaire est pronotepy et qu'il n'y a pas de repli.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.homework_source = "auto"
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback is None
def test_homework_sources_auto_no_source_configured_raises(mock_fetcher: PronoteFetcher) -> None:
"""Test _homework_sources en mode AUTO avec aucune source configurée.
On désactive les deux sources. On vérifie que l'erreur PipelineCriticalError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.homework_source = "auto"
mock_fetcher._settings.pronote.ical_url = None
mock_fetcher._settings.pronote.pronote_url = None
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher._homework_sources()
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test que fetch_homework lève PipelineCriticalError en mode explicite ical.
On mock ICAL pour échouer. On vérifie que l'erreur critique est levée
et que pronotepy n'est PAS appelé (pas de repli en mode explicite).
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
mock_fetcher._settings.pronote.homework_source = "ical"
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.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
m_collect.side_effect = OSError("collect error")
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_homework(target_date)
# En mode explicite, seul l'échec de la source primaire est rapporté
assert "la source ical a échoué" in str(exc_info.value)
# Vérifie que pronotepy n'a PAS été appelé (pas de repli en mode explicite)
client.get_homeworks.assert_not_called()
def test_fetch_homework_auto_fallback_returns_empty_logs_warning(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""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.
Ce test utilise le mode AUTO pour tester le comportement de repli.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
mock_fetcher._settings.pronote.homework_source = "auto"
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()
mock_fetcher._pronote_client = client
m_collect.return_value = [] # Empty result
result = mock_fetcher.fetch_homework(target_date)
assert result == []
assert "a retourné un résultat vide après l'échec de ical" in caplog.text
def test_fetch_informations_logs_and_re_raises_secret(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que fetch_informations journalise et relance les exceptions avec secret masqué.
On mock get_informations pour lever une exception contenant une URL secrète.
On vérifie que l'exception est relancée et que le log ne contient pas le secret.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
:rtype: None
"""
client = MagicMock()
error_msg = "Erreur Pronote : impossible de récupérer les informations https://pronote.example.com/infos?token=SECRET_TOKEN_789"
client.get_informations.side_effect = OSError(error_msg)
mock_fetcher._pronote_client = client
with pytest.raises(OSError) as exc_info:
mock_fetcher.fetch_informations()
assert exc_info.value is client.get_informations.side_effect
assert "SECRET_TOKEN_789" not in caplog.text
assert (
"pronote.example.com/infos?token=REDACTED" in caplog.text
or "pronote.example.com/infos" in caplog.text
)
# Ensure trailing newline

View File

@@ -26,6 +26,7 @@ from pronote_sync.sources.pronote.ical import (
generate_homework_id,
get_calendar_name,
normalize_homework_text,
parse_body,
parse_ical,
)
@@ -463,3 +464,137 @@ def test_generate_homework_id_deterministic() -> None:
id1 = generate_homework_id(due_on, text)
id2 = generate_homework_id(due_on, text)
assert id1 == id2
CANCELLED_STATUS_ICAL = """BEGIN:VCALENDAR
VERSION:2.0
X-WR-CALNAME:Test
BEGIN:VEVENT
UID:Test-123-20260906T120000Z-Index-Education
DTSTART:20260907T080000Z
DTEND:20260907T090000Z
SUMMARY:Test Course
STATUS:CANCELLED
DESCRIPTION:<div></div>
END:VEVENT
END:VCALENDAR"""
NORMAL_STATUS_ICAL = """BEGIN:VCALENDAR
VERSION:2.0
X-WR-CALNAME:Test
BEGIN:VEVENT
UID:Test-123-20260906T120000Z-Index-Education
DTSTART:20260907T080000Z
DTEND:20260907T090000Z
SUMMARY:Test Course
STATUS:CONFIRMED
DESCRIPTION:<div></div>
END:VEVENT
END:VCALENDAR"""
MOVED_BY_CATEGORY_ICAL = """BEGIN:VCALENDAR
VERSION:2.0
X-WR-CALNAME:Test
BEGIN:VEVENT
UID:Test-123-20260906T120000Z-Index-Education
DTSTART:20260907T080000Z
DTEND:20260907T090000Z
SUMMARY:Test Course
CATEGORIES:Cours - Cours déplacé
DESCRIPTION:<div></div>
END:VEVENT
END:VCALENDAR"""
MULTIPLE_BLOCKS_SAME_DATE_ICAL = """BEGIN:VCALENDAR
VERSION:2.0
X-WR-CALNAME:Test
BEGIN:VEVENT
UID:Test-123-20260906T120000Z-Index-Education
DTSTART:20260907T080000Z
DTEND:20260907T090000Z
SUMMARY:Test Course
CATEGORIES:Cours
DESCRIPTION:<div>
Matière : Math
Professeur : M. Dupont
Salle : 204
<strong>Pour le 10/09/2026 :</strong>
Exercice 1 à 5 page 42.
<strong>Pour le 10/09/2026 :</strong>
Exercice 6 à 10 page 43.
</div>
END:VEVENT
END:VCALENDAR"""
def test_parse_ical_status_cancelled_only() -> None:
"""Un cours avec STATUS:CANCELLED mais sans CATEGORIES contenant 'Cours annulé' a status == LessonStatus.CANCELLED.
:return: None
"""
lessons, _, _ = parse_ical(CANCELLED_STATUS_ICAL)
assert len(lessons) == 1
assert lessons[0].status == LessonStatus.CANCELLED
def test_parse_ical_status_normal_without_cancel() -> None:
"""Un cours avec STATUS:CONFIRMED (ou sans STATUS) a status == LessonStatus.NORMAL.
:return: None
"""
lessons, _, _ = parse_ical(NORMAL_STATUS_ICAL)
assert len(lessons) == 1
assert lessons[0].status == LessonStatus.NORMAL
def test_parse_ical_moved_by_category_only() -> None:
"""Un cours avec CATEGORIES:Cours - Cours déplacé et sans STATUS a status == LessonStatus.MOVED.
:return: None
"""
lessons, _, _ = parse_ical(MOVED_BY_CATEGORY_ICAL)
assert len(lessons) == 1
assert lessons[0].status == LessonStatus.MOVED
def test_parse_body_multiple_blocks_same_date() -> None:
"""Un DESCRIPTION avec deux sections 'Pour le' à la même date conserve les deux blocs.
:return: None
"""
body_html = (
"<div>\n"
" <strong>Pour le 10/09/2026:</strong>\n"
" Exercice 1 à 5 page 42.\n"
" <strong>Pour le 10/09/2026:</strong>\n"
" Exercice 6 à 10 page 43.\n"
"</div>"
)
content, due_blocks, assigned_blocks = parse_body(body_html)
assert len(due_blocks) == 2
assert due_blocks[0][0] == date(2026, 9, 10)
assert due_blocks[0][1] == "Exercice 1 à 5 page 42."
assert due_blocks[1][0] == date(2026, 9, 10)
assert due_blocks[1][1] == "Exercice 6 à 10 page 43."
def test_collect_homeworks_from_fixture() -> None:
"""Parse le fixture pronote-4e.ics, appelle collect_homeworks pour le 10/09/2026 et vérifie qu'au moins un devoir est retourné.
:return: None
"""
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
with open(fixture_path, encoding="utf-8") as f:
content = f.read()
lessons, _, _ = parse_ical(content)
homeworks = collect_homeworks(lessons, date(2026, 9, 10))
assert len(homeworks) >= 1
# Vérifie qu'au moins un devoir a le bon sujet et texte
assert any(hw.subject == "Mathématiques" for hw in homeworks)
assert any("Exercices 1 à 5 page 42" in hw.text for hw in homeworks)

View File

@@ -31,7 +31,8 @@ def test_protocol_methods(mocker: pytest_mock.MockerFixture) -> None:
"""
assert hasattr(PronoteClientProtocol, "get_messages")
assert hasattr(PronoteClientProtocol, "get_informations")
assert hasattr(PronoteClientProtocol, "get_agenda_fallback")
assert hasattr(PronoteClientProtocol, "get_lessons")
assert hasattr(PronoteClientProtocol, "get_homeworks")
# --- Client with mocked pronotepy ---
@@ -45,9 +46,11 @@ def pronote_settings() -> PronoteSettings:
:rtype: PronoteSettings
"""
return PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="testent",
ent="bordeaux",
account_type="parent",
)
@@ -82,7 +85,7 @@ def test_get_messages_success(
mock_message.seen = True
mock_discussion.messages = [mock_message]
mock_client.discussions.return_value = [mock_discussion]
mocker.patch("pronotepy.Client", return_value=mock_client)
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_messages()
@@ -111,7 +114,7 @@ def test_get_messages_empty_on_error(
"""
mock_client = mocker.MagicMock()
mock_client.discussions.side_effect = pronotepy.PronoteAPIError("API error")
mocker.patch("pronotepy.Client", return_value=mock_client)
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_messages()
@@ -138,7 +141,7 @@ def test_get_informations_success(
mock_info.read = False
mock_info.survey = True
mock_client.information_and_surveys.return_value = [mock_info]
mocker.patch("pronotepy.Client", return_value=mock_client)
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
@@ -167,7 +170,7 @@ def test_get_informations_empty_on_error(
"""
mock_client = mocker.MagicMock()
mock_client.information_and_surveys.side_effect = pronotepy.PronoteAPIError("API error")
mocker.patch("pronotepy.Client", return_value=mock_client)
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
@@ -175,10 +178,10 @@ def test_get_informations_empty_on_error(
assert messages == []
def test_get_agenda_fallback_success(
def test_get_lessons_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_agenda_fallback retourne un tuple de listes en cas de succès.
"""Vérifie que get_lessons retourne une liste de Lesson en cas de succès.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
@@ -201,20 +204,11 @@ def test_get_agenda_fallback_success(
mock_content.description = "Lesson content"
mock_lesson.content = mock_content
# Mock des devoirs
mock_hw = mocker.MagicMock()
mock_hw.id = "hw-101"
mock_hw.subject = mocker.MagicMock()
mock_hw.subject.name = "Maths"
mock_hw.date = date(2024, 9, 15)
mock_hw.description = "Do your homework"
mock_client.lessons.return_value = [mock_lesson]
mock_client.homework.return_value = [mock_hw]
mocker.patch("pronotepy.Client", return_value=mock_client)
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
lessons, homeworks = client.get_agenda_fallback(date(2024, 9, 1), date(2024, 9, 30))
lessons = client.get_lessons(date(2024, 9, 1), date(2024, 9, 30))
assert isinstance(lessons, list)
assert len(lessons) == 1
@@ -230,6 +224,32 @@ def test_get_agenda_fallback_success(
assert lesson.status == LessonStatus.NORMAL
assert lesson.content == "Lesson content"
def test_get_homeworks_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_homeworks retourne une liste de Homework en cas de succès.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
# Mock des devoirs
mock_hw = mocker.MagicMock()
mock_hw.id = "hw-101"
mock_hw.subject = mocker.MagicMock()
mock_hw.subject.name = "Maths"
mock_hw.date = date(2024, 9, 15)
mock_hw.description = "Do your homework"
mock_client.homework.return_value = [mock_hw]
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
homeworks = client.get_homeworks(date(2024, 9, 1), date(2024, 9, 30))
assert isinstance(homeworks, list)
assert len(homeworks) == 1
homework = homeworks[0]
@@ -243,10 +263,10 @@ def test_get_agenda_fallback_success(
assert homework.html == "Do your homework"
def test_get_agenda_fallback_empty_on_error(
def test_get_lessons_propagates_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_agenda_fallback retourne des listes vides en cas d'erreur API.
"""Vérifie que get_lessons propage les exceptions API.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
@@ -254,91 +274,166 @@ def test_get_agenda_fallback_empty_on_error(
"""
mock_client = mocker.MagicMock()
mock_client.lessons.side_effect = pronotepy.PronoteAPIError("API error")
mocker.patch("pronotepy.Client", return_value=mock_client)
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
lessons, homeworks = client.get_agenda_fallback(date(2024, 9, 1), date(2024, 9, 30))
assert lessons == []
assert homeworks == []
with pytest.raises(pronotepy.PronoteAPIError):
client.get_lessons(date(2024, 9, 1), date(2024, 9, 30))
def test_missing_credentials_returns_empty(empty_pronote_settings: PronoteSettings) -> None:
"""Vérifie que les méthodes retournent une liste vide si les identifiants sont manquants.
def test_get_homeworks_propagates_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_homeworks propage les exceptions API.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.homework.side_effect = pronotepy.PronoteAPIError("API error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
with pytest.raises(pronotepy.PronoteAPIError):
client.get_homeworks(date(2024, 9, 1), date(2024, 9, 30))
def test_missing_credentials_raises(empty_pronote_settings: PronoteSettings) -> None:
"""Vérifie que les appels échouent avec ValueError si les identifiants sont manquants.
:param empty_pronote_settings: Paramètres Pronote avec tous les champs à None.
:return: None
"""
client = PronoteClient(empty_pronote_settings)
messages = client.get_messages()
assert messages == []
informations = client.get_informations()
assert informations == []
lessons, homeworks = client.get_agenda_fallback(date(2024, 9, 1), date(2024, 9, 30))
assert lessons == []
assert homeworks == []
with pytest.raises(ValueError, match="pronote_url, username, password et ent sont requis"):
client._connect()
def test_password_used_in_connection(
def test_connect_with_ent_resolution(mocker: pytest_mock.MockerFixture) -> None:
"""Vérifie que _resolve_ent retourne le callable attendu pour un ENT connu.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from pronote_sync.sources.pronote.client import _resolve_ent
resolver = _resolve_ent("bordeaux")
assert resolver is not None
def test_connect_with_unknown_ent_raises(mocker: pytest_mock.MockerFixture) -> None:
"""Vérifie que _resolve_ent lève ValueError pour un ENT inconnu.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from pronote_sync.sources.pronote.client import _resolve_ent
with pytest.raises(ValueError) as exc_info:
_resolve_ent("inconnu")
assert "ENT inconnu : 'inconnu'" in str(exc_info.value)
assert "ENT supportés :" in str(exc_info.value)
def test_connect_parent_account_type(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que le mot de passe est bien utilisé pour la connexion.
"""Vérifie que account_type='parent' utilise pronotepy.ParentClient.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
# Patch pronotepy.Client to return our mock
from pronote_sync.sources.pronote import client as client_module
from unittest.mock import Mock
from pronote_sync.sources.pronote.client import PronoteClient
mock_client = mocker.MagicMock()
mock_discussion = mocker.MagicMock()
mock_message = mocker.MagicMock()
mock_message.id = "msg-123"
mock_message.content = "Test"
mock_message.author = "Teacher"
mock_message.created = datetime(2024, 9, 1, 10, 0, 0)
mock_message.seen = False
mock_discussion.messages = [mock_message]
mock_discussion.subject = "Test Subject"
mock_client.discussions.return_value = [mock_discussion]
# Patch pronotepy.Client to return our mock
mocker.patch.object(client_module, "pronotepy")
client_module.pronotepy.Client = lambda u, p, e: mock_client # type: ignore[attr-defined] # noqa: ARG005
# Setup mock client
mock_client = mocker.MagicMock()
mock_discussion = mocker.MagicMock()
mock_message = mocker.MagicMock()
mock_message.id = "msg-123"
mock_message.content = "Test"
mock_message.author = "Teacher"
mock_message.created = datetime(2024, 9, 1, 10, 0, 0)
mock_message.seen = False
mock_discussion.messages = [mock_message]
mock_discussion.subject = "Test Subject"
mock_client.discussions.return_value = [mock_discussion]
# Patch pronotepy.Client to return our mock
mocker.patch("pronote_sync.sources.pronote.client.pronotepy.Client", return_value=mock_client)
mock_client_class = Mock(return_value=mock_client)
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
mocker.patch("pronotepy.Client")
client = PronoteClient(pronote_settings)
_ = client.get_messages()
_ = client._connect()
# Vérifie que le client a été créé avec le mot de passe
# Le mock de Client doit avoir été appelé avec username, password, ent
client_class_mock = client_module.pronotepy.Client # type: ignore[attr-defined]
client_class_mock.assert_called_once()
call_args = client_class_mock.call_args
assert call_args is not None
assert len(call_args.args) >= 3
assert call_args.args[0] == "testuser"
assert call_args.args[1] == "testpass"
assert call_args.args[2] == "testent"
# Verify ParentClient was used
assert mock_client_class.call_count == 1
pronotepy.Client.assert_not_called() # type: ignore[attr-defined]
def test_connect_student_account_type(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que account_type='student' utilise pronotepy.Client.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
from unittest.mock import Mock
from pronote_sync.sources.pronote.client import PronoteClient
pronote_settings_student = PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="student",
)
mock_client = mocker.MagicMock()
mock_client_class = Mock(return_value=mock_client)
mocker.patch("pronotepy.Client", new=mock_client_class)
mocker.patch("pronotepy.ParentClient")
client = PronoteClient(pronote_settings_student)
_ = client._connect()
# Verify Client was used
assert mock_client_class.call_count == 1
pronotepy.ParentClient.assert_not_called() # type: ignore[attr-defined]
def test_get_messages_degraded_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_messages retourne une liste vide en cas d'erreur réseau.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.discussions.side_effect = ConnectionError("Network error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_messages()
assert messages == []
def test_get_informations_degraded_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_informations retourne une liste vide en cas d'erreur réseau.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.information_and_surveys.side_effect = TimeoutError("Timeout")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
assert messages == []
# Ensure trailing newline