Compare commits

..

17 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
e0cd073f9a test(M4): tests unitaires iCal, client pronotepy et repli
- test_ical.py : 20 tests (fetch file://, parsing VEVENT, détection
  cours annulé, collect_homeworks avec déduplication, normalisation,
  génération d'ID déterministe, redaction des erreurs)
- test_pronote_client.py : 9 tests (Protocol, messages mockés,
  informations, agenda fallback, credentials manquants, sécurité
  mot de passe)
- test_fallback.py : 16 tests (modes ICAL/PRONOTEPY/AUTO, repli,
  PipelineCriticalError si les deux sources échouent, redaction
  des erreurs)
- pre-commit : ajout de responses et pytest-mock au hook mypy
- Total : 42 nouveaux tests M4 (122 tests au total sur le projet)

Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 14:10:16 +02:00
f2be495f55 feat(M4): sources/pronote/fallback.py — repli iCal/pronotepy
- AgendaSource : enum StrEnum (AUTO/ICAL/PRONOTEPY)
- PronoteFetcherProtocol : Protocol consommé par le pipeline
- PronoteFetcher : unifie iCal et pronotepy avec repli automatique
  - AUTO : iCal d'abord, repli pronotepy, PipelineCriticalError si
    les deux échouent
  - ICAL/PRONOTEPY : source explicite, erreur redactée puis
    PipelineCriticalError
  - Messages et informations : toujours via pronotepy
- Avertissement quand le repli pronotepy retourne un résultat vide
  après échec iCal (limite du mode dégradé)

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-06 13:45:08 +02:00
f3177e70f6 feat(M4): sources/pronote/client.py — client pronotepy avec Protocol
- PronoteClientProtocol : Protocol (get_messages, get_informations,
  get_agenda_fallback) pour la logique de repli
- PronoteClient : encapsule pronotepy.Client (connexion paresseuse)
  avec API réelle 2.15.7 (lessons/homework/discussions/information_and_surveys)
- Erreurs PronoteAPIError et ValueError catchées, loggées avec
  redact_exception, retournent des résultats vides
- Aucune fuite de secret (get_secret_value() uniquement à la connexion)
- pre-commit : ajout de pronotepy au hook mypy

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-06 13:32:54 +02:00
cb621c15f4 feat(M4): sources/pronote/ical.py — fetch, parsing et collecte des devoirs
- fetch_ical : HTTP via requests, support file:// (URI décodé),
  validation BEGIN:VCALENDAR, exceptions redactées
- get_calendar_name : extraction X-WR-CALNAME (paramètres + lignes repliées)
- parse_ical : parsing VEVENT → Lesson/SchoolEvent, détection cours
  annulé/déplacé, UID déterministe si absent, homeworks toujours vide
- collect_homeworks : deux passes (due + assigned), déduplication
  par ID, tri par (subject, text), target_date injecté
- generate_homework_id : SHA-1 12 chars (usedforsecurity=False)
- normalize_homework_text : unification whitespace/HTML/lowercase
- pre-commit : ajout de types-requests et icalendar au hook mypy

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-06 13:32:11 +02:00
f82e79360b feat(M4): hiérarchie d'erreurs, fixture iCal et correction TODO
- pronote_sync/errors.py : PronoteSyncError (base) + PipelineCriticalError
- tests/fixtures/pronote-4e.ics : fixture anonymisée (3 VEVENTs : cours
  normal avec devoirs, cours annulé, événement scolaire)
- TODO.md ligne 82 : normalize_uid → normalize_pronote_uid (alignement
  sur le code existant dans utils/uid.py)

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 13:20:24 +02:00
23 changed files with 6328 additions and 583 deletions

View File

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

4
.gitignore vendored
View File

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

View File

@@ -26,7 +26,7 @@ repos:
name: mypy name: mypy
entry: mypy entry: mypy
language: python language: python
additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.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] types: [python]
pass_filenames: true pass_filenames: true

View File

@@ -139,11 +139,11 @@
"type": "Hex High Entropy String", "type": "Hex High Entropy String",
"filename": "GUIDE_DEV_PYTHON.md", "filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_secret": false,
"is_verified": true, "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 ## 2. Stack technique
### Langage et dépendances ### Langage et dépendances
- **Python** : ≥ 3.13 (actuellement 3.14 dans `.venv/`) - **Python** : ≥ 3.13.5 (actuellement 3.14 dans `.venv/`)
- **Dépendances principales** : - **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 ### Outils de développement
- **Linter** : `ruff` (longueur de ligne : 100) - **Linter** : `ruff` (longueur de ligne : 100)
@@ -36,6 +37,7 @@
``` ```
pronote_sync/ pronote_sync/
├── config/ # Configuration et paramètres ├── config/ # Configuration et paramètres
├── errors.py # Hiérarchie canonique des erreurs du pipeline
├── models/ # Modèles de données (Pydantic v2) ├── models/ # Modèles de données (Pydantic v2)
├── sources/ # Connecteurs (Pronote, iCal, etc.) ├── sources/ # Connecteurs (Pronote, iCal, etc.)
├── sync/ # Logique de synchronisation ├── sync/ # Logique de synchronisation
@@ -110,6 +112,8 @@ pronote-sync --dry-run
### Architecture ### Architecture
- **Injection de dépendances** : Utiliser `typing.Protocol` et une **composition root** dans `pipeline/run.py`. **Interdiction** des singletons globaux. - **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 ### Style
- **Longueur de ligne** : 100 caractères maximum (configuré dans `ruff`). - **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**. - **Idempotence** : Deux exécutions identiques sans changement externe doivent produire le **même résultat**.
- **Mode dégradé** : - **Mode dégradé** :
- Si l'IA échoue → retourner un message **sans synthèse**. - Si l'IA échoue → retourner un message **sans synthèse**.
- Si `pronotepy` échoue → fallback vers le parsing **iCal**. - En mode source `auto`, essayer iCal puis utiliser `pronotepy` uniquement si iCal lève une
- Si tout échoue → lever une **erreur explicite**. 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) ### Documentation (docstrings)
- **Obligatoire** : **Toute** fonction, méthode et classe publique doit avoir une docstring. - **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. - **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** : - **Contenu** :
- Une ligne de résumé courte (une phrase). - Une ligne de résumé courte (une phrase).
- Une description étendue optionnelle. - Une description étendue optionnelle.
@@ -157,6 +179,11 @@ def fetch_ical(url: str) -> str:
### Règles absolues ### Règles absolues
- **Aucun secret en clair** : Ni dans le code, ni dans les logs, ni dans les erreurs, ni dans les fixtures. - **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`. - **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 ### Bonnes pratiques
- **Types sécurisés** : Les mots de passe et clés API doivent utiliser `pydantic.SecretStr`. - **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 ## 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. 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. 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 : 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. - Succès de la synchronisation Pronote → CalDAV/XMPP.
- Repli vers iCal en cas d'échec de `pronotepy`. - Repli d'iCal vers `pronotepy` en mode `auto`, sans repli dans les modes explicites.
- Gestion des erreurs 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. 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. 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. 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 : 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. - 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. - 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). - 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. 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`). - [x] 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). - [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)`.
- [ ] Détecter les statuts (`CANCELLED`/`MOVED`) via `CATEGORIES` et `STATUS:CANCELLED`. - [x] 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. - [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.
- [ ] 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] Exposer séparément les cours, devoirs, messages et informations dans le client `pronotepy` ; filtrer les devoirs sur `due_on == target_date`.
- [ ] Implémenter le repli : iCal échoue → pronotepy ; pronotepy échoue → iCal ; les deux échouent → `PipelineCriticalError`. - [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`.
- [ ] Normaliser les UID via `utils/uid.normalize_uid` pour la stabilité des événements. - [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 ### Critères d'acceptation
- `fetch_ical` parse `tests/fixtures/pronote-4e.ics` en leçons/devoirs/événements corrects (cours annulé détecté). - `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.
- Le client pronotepy récupère messages/devoirs (mocké). - 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 repli bascule correctement et lève une erreur critique si aucune source disponible. - Le constructeur `ParentClient` est testé avec l'ordre réel de ses paramètres, l'URL Pronote et une fonction ENT autorisée.
- Aucun secret dans les messages d'erreur de fetch. - 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. 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). - [x] 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`). - [x] 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`). - [x] 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`). - [x] 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] Gérer un flux invalide (`bozo`) et les exceptions sans fuite de secret (retour `[]`/warning).
### Critères d'acceptation ### Critères d'acceptation
- `fetch_and_parse` renvoie les nouveaux articles triés par date décroissante, sans doublons. - `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/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). - [ ] 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. - [ ] 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. - [ ] 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`). - [ ] Garantir l'idempotence (2 exécutions identiques → même `CalDAVSyncResult`).
- [ ] Réutiliser `BlogRSSState` pour l'état blog si pertinent (sinon `sync/blog_state.py`). - [ ] Réutiliser `BlogRSSState` pour l'état blog si pertinent (sinon `sync/blog_state.py`).
### Critères d'acceptation ### Critères d'acceptation
- Le plan de sync est correctement calculé (PronoteData vs état local). - 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. - 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`. - 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. 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 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. - [ ] 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)`. - [ ] 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). - [ ] Implémenter le mode `dry_run` (aucune écriture CalDAV/XMPP).
- [ ] Câbler l'injection des dépendances (Protocol + composition root), sans singleton global. - [ ] 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 ### Critères d'acceptation
- Le pipeline complet s'exécute de bout en bout (mocks) dans le bon ordre. - 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. - 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. - `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 ### Critères d'acceptation
- `pronote-sync --dry-run --log-level DEBUG` s'exécute sans effet de bord. - `pronote-sync --dry-run --log-level DEBUG` s'exécute sans effet de bord.
- Le script console est installable (`[project.scripts]` dans `pyproject.toml`). - 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/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…). - [ ] 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`. - [ ] É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/integration/` : `test_pipeline`, `test_caldav` (mocké), `test_xmpp` (mocké).
- [ ] Écrire `tests/e2e/test_cli.py` : exécution CLI en dry-run. - [ ] Écrire `tests/e2e/test_cli.py` : exécution CLI en dry-run.
- [ ] Tests sans réseau (mocks `responses`/`aioresponses`/`pytest-mock`) ; couverture ≥ 90 %. - [ ] 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 ### Critères d'acceptation
- `pytest` passe et `pytest --cov` atteint ≥ 90 % (`fail_under = 90`). - `pytest` passe et `pytest --cov` atteint ≥ 90 % (`fail_under = 90`).

View File

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

33
pronote_sync/errors.py Normal file
View File

@@ -0,0 +1,33 @@
"""Hiérarchie d'erreurs transversale du projet pronote-sync."""
from __future__ import annotations
class PronoteSyncError(Exception):
"""Erreur de base pour toutes les exceptions du projet pronote-sync.
Toutes les exceptions métier héritent de cette classe pour permettre
une interception unifiée des erreurs applicatives.
"""
def __init__(self, message: str) -> None:
"""Initialise l'erreur avec un message descriptif.
:param message: Message décrivant la cause de l'erreur.
"""
super().__init__(message)
class PipelineCriticalError(PronoteSyncError):
"""Erreur critique du pipeline, levée quand aucune récupération n'est possible.
Par exemple : échec simultané des sources iCal et pronotepy,
rendant impossible toute synchronisation.
"""
def __init__(self, message: str) -> None:
"""Initialise l'erreur critique avec un message descriptif.
:param message: Message décrivant la cause de l'erreur critique.
"""
super().__init__(message)

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

@@ -0,0 +1,339 @@
"""Client d'accès à Pronote via ``pronotepy``.
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. 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 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
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
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."""
def get_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote.
:return: Liste des messages des professeurs.
:rtype: list[Message]
"""
...
def get_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote.
:return: Liste des informations et sondages.
:rtype: list[Message]
"""
...
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: 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]
"""
...
class PronoteClient:
"""Client d'accès à Pronote via ``pronotepy``.
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:
"""Initialise le client Pronote sans se connecter.
:param settings: Paramètres d'accès à Pronote (username, password, ent).
"""
self._settings: PronoteSettings = settings
self._client: pronotepy.Client | None = None
def _connect(self) -> pronotepy.Client:
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse).
Le client est créé une seule fois puis réutilisé pour les appels
suivants. Le nom d'ENT 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 ``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 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]:
"""Récupère les messages des discussions Pronote.
Chaque message d'une discussion est mappé sur un modèle
:class:`Message` de type ``DISCUSSION``, le sujet de la discussion
servant de titre.
:return: Liste des messages des professeurs ; vide en cas d'erreur.
:rtype: list[Message]
"""
try:
client = self._connect()
messages: list[Message] = []
for discussion in client.discussions():
for message in discussion.messages:
messages.append(
Message(
id=message.id,
type=MessageType.DISCUSSION,
title=discussion.subject,
content=message.content,
author=message.author or "",
date=message.created,
read=message.seen,
)
)
return messages
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),
)
return []
def get_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote.
Chaque entrée est mappée sur un modèle :class:`Message` de type
``SURVEY`` si c'est un sondage, ``INFORMATION`` sinon.
:return: Liste des informations et sondages ; vide en cas d'erreur.
:rtype: list[Message]
"""
try:
client = self._connect()
messages: list[Message] = []
for info in client.information_and_surveys():
messages.append(
Message(
id=info.id,
type=MessageType.SURVEY if info.survey else MessageType.INFORMATION,
title=info.title or "",
content=info.content(),
author=info.author,
date=info.creation_date,
read=info.read,
)
)
return messages
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_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours via ``pronotepy`` (repli iCal).
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: 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.
"""
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,
)
)
return homeworks

View File

@@ -0,0 +1,408 @@
"""Logique de repli unifiant les sources iCal et pronotepy.
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``).
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` ; les exceptions
d'origine ne sont jamais chaînées (``from None``).
"""
from __future__ import annotations
import logging
from datetime import date, timedelta
from enum import StrEnum
from typing import Literal, Protocol
from pronote_sync.config.settings import Settings
from pronote_sync.errors import PipelineCriticalError
from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message
from pronote_sync.sources.pronote.client import PronoteClientProtocol
from pronote_sync.sources.pronote.ical import collect_homeworks, fetch_ical, parse_ical
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."""
AUTO = "auto"
ICAL = "ical"
PRONOTEPY = "pronotepy"
class PronoteFetcherProtocol(Protocol):
"""Interface de récupération des données Pronote consommée par le pipeline."""
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère les cours et les événements scolaires.
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises PipelineCriticalError: Si toutes les sources configurées échouent.
"""
...
def fetch_homework(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs pour la date cible.
: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.
"""
...
def fetch_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote.
:return: Liste des messages.
:rtype: list[Message]
"""
...
def fetch_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote.
:return: Liste des informations et sondages.
:rtype: list[Message]
"""
...
class PronoteFetcher:
"""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``) : 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.
"""
def __init__(self, settings: Settings, pronote_client: PronoteClientProtocol) -> None:
"""Initialise le fetcher sans récupérer aucune donnée.
:param settings: Configuration racine du pipeline (Pronote, fenêtre de synchronisation).
:param pronote_client: Client pronotepy utilisé pour les sources pronotepy.
"""
self._settings: Settings = settings
self._pronote_client: PronoteClientProtocol = pronote_client
def _fetch_window(self) -> tuple[date, date]:
"""Calcule la fenêtre de synchronisation autour de la date du jour.
:return: Tuple ``(date de début, date de fin)`` de la fenêtre.
:rtype: tuple[date, date]
"""
today = date.today()
start = today - timedelta(days=self._settings.app.sync_past_days)
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.
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide.
:raises OSError: Si le fichier iCal local est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue.
"""
ical_url = self._settings.pronote.ical_url
if ical_url is None:
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
raw_ical = fetch_ical(ical_url.get_secret_value())
lessons, _, school_events = parse_ical(raw_ical)
return lessons, school_events
def _fetch_agenda_pronotepy(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère l'agenda depuis pronotepy.
Les événements scolaires (vacances, jours fériés) ne sont pas
fournis par pronotepy : la liste retournée est vide.
: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_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 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 tentées échouent.
"""
primary, fallback = self._agenda_sources()
try:
return self._fetch_agenda_source(primary)
except Exception as exc:
logger.error(
"Échec de la récupération %s pour l'agenda : %s",
primary,
redact_exception(exc),
)
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.
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
:raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide.
:raises OSError: Si le fichier iCal local est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue.
"""
lessons, _ = self._fetch_agenda_ical()
return collect_homeworks(lessons, target_date)
def _fetch_homework_pronotepy(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis pronotepy pour la date cible.
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 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:
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 les devoirs : ni la source iCal ni pronotepy n'est configurée"
) from None
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)
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:
return self._fetch_homework_source(primary, target_date)
except Exception as exc:
logger.error(
"Échec de la récupération %s pour les devoirs : %s",
primary,
redact_exception(exc),
)
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).
:return: Liste des messages.
:rtype: list[Message]
"""
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).
:return: Liste des informations et sondages.
:rtype: list[Message]
"""
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

@@ -0,0 +1,467 @@
"""Récupération et parsing du flux iCal Pronote.
Ce module fournit le téléchargement du flux iCal Pronote (via HTTP ou
``file://`` pour les tests) ainsi que son parsing en modèles : cours
(:class:`~pronote_sync.models.agenda.Lesson`), événements scolaires
(:class:`~pronote_sync.models.agenda.SchoolEvent`) et devoirs
(:class:`~pronote_sync.models.homework.Homework`).
La collecte finale des devoirs est réalisée à part, une fois la date
cible connue, via :func:`collect_homeworks`.
"""
from __future__ import annotations
import hashlib
import re
import urllib.parse
from datetime import date, datetime
from html import unescape
from pathlib import Path
from typing import TypedDict
import requests
from icalendar import Calendar
from ...models.agenda import (
HomeworkBlock,
Lesson,
LessonStatus,
SchoolEvent,
SchoolEventKind,
)
from ...models.homework import Homework
from ...utils.redaction import redact_exception, redact_url
from ...utils.uid import generate_deterministic_uid, normalize_pronote_uid
_HEADER_LABEL_PATTERN = re.compile(r"\b(Matière|Professeurs?|Salles?|Groupe)\s*:\s*")
_CALNAME_PATTERN = re.compile(r"^X-WR-CALNAME(?:;[^:]*)?:([^\r\n]*)", re.MULTILINE)
_TAG_PATTERN = re.compile(r"<[^>]+>")
_CONTENT_PATTERN = re.compile(
r"<strong>Contenu pédagogique\s*:\s*</strong>(.*?)(?=<strong>|</div>\s*$|\Z)",
re.DOTALL,
)
_DUE_PATTERN = re.compile(
r"<strong>Pour le (\d{2}/\d{2}/\d{4})\s*:\s*</strong>(.*?)(?=<strong>|</div>\s*$|\Z)",
re.DOTALL,
)
_ASSIGNED_PATTERN = re.compile(
r"<strong>Donné le (\d{2}/\d{2}/\d{4})\s*:\s*</strong>(.*?)(?=<strong>|</div>\s*$|\Z)",
re.DOTALL,
)
_HEADERS = {
"accept": "text/calendar",
"user-agent": "pronote-sync",
}
class HeaderInfo(TypedDict):
"""Métadonnées du cours extraites de l'en-tête de la DESCRIPTION."""
subject: str
teachers: list[str]
rooms: list[str]
group: str | None
def fetch_ical(url: str, timeout: int = 20) -> str:
"""Récupère le contenu brut d'un flux iCal Pronote.
Gère les URLs ``file://`` pour les tests locaux (le chemin est
décodé de l'échappement URI, ex. ``%20`` → espace) et valide que
le flux commence bien par ``BEGIN:VCALENDAR``. Toutes les erreurs
sont relancées avec un message dont les secrets (token
``icalsecurise``) sont masqués.
:param url: URL du flux iCal (avec token ``icalsecurise``) ou chemin ``file://``.
:param timeout: Timeout HTTP en secondes (défaut : 20).
:return: Contenu brut du flux iCal.
:rtype: str
:raises OSError: Si le fichier local ``file://`` est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue.
:raises ValueError: Si le flux ne commence pas par ``BEGIN:VCALENDAR``.
"""
if url.startswith("file://"):
parsed_url = urllib.parse.urlparse(url)
path = Path(urllib.parse.unquote(parsed_url.path))
try:
content = path.read_text(encoding="utf-8")
except OSError as exc:
raise OSError(
f"Impossible de lire le fichier iCal {redact_url(url)} : {redact_exception(exc)}"
) from exc
if not content.lstrip().startswith("BEGIN:VCALENDAR"):
raise ValueError(f"Fichier iCal invalide (pas de BEGIN:VCALENDAR) : {redact_url(url)}")
return content
try:
response = requests.get(url, headers=_HEADERS, timeout=timeout)
response.raise_for_status()
content = response.text
except Exception as exc:
raise requests.RequestException(
f"Échec de la récupération du flux iCal {redact_url(url)} : {redact_exception(exc)}"
) from exc
if not content.lstrip().startswith("BEGIN:VCALENDAR"):
raise ValueError(f"Flux iCal invalide (pas de BEGIN:VCALENDAR) : {redact_url(url)}")
return str(content)
def _unfold_ical(raw_ical: str) -> str:
"""Déplie les lignes de continuation iCalendar.
Une ligne commençant par un espace ou une tabulation prolonge la
ligne précédente : le caractère d'espacement initial est retiré et
la suite est jointe à la ligne précédente.
:param raw_ical: Contenu brut du flux iCal.
:return: Contenu avec les lignes de continuation dépliées.
:rtype: str
"""
unfolded: list[str] = []
for line in raw_ical.splitlines():
if line.startswith((" ", "\t")) and unfolded:
unfolded[-1] += line[1:]
else:
unfolded.append(line)
return "\n".join(unfolded)
def get_calendar_name(raw_ical: str) -> str | None:
"""Extrait le nom du calendrier depuis la propriété ``X-WR-CALNAME``.
La propriété peut comporter des paramètres (par exemple
``X-WR-CALNAME;LANGUAGE=fr:Nom``) et les lignes de continuation
iCalendar sont dépliées avant la recherche.
:param raw_ical: Contenu brut du flux iCal.
:return: Nom du calendrier ou ``None`` si la propriété est absente.
:rtype: str | None
"""
match = _CALNAME_PATTERN.search(_unfold_ical(raw_ical))
if match is None:
return None
return match.group(1).strip()
def split_header_and_body(description: str) -> tuple[str, str]:
"""Sépare l'en-tête texte du corps HTML dans la DESCRIPTION.
L'en-tête est la partie avant le premier ``<strong>``, le corps
HTML commence à partir du premier ``<strong>``.
:param description: Contenu brut de la DESCRIPTION d'un VEVENT.
:return: Tuple ``(en-tête, corps)`` ; le corps est vide si aucun ``<strong>``.
:rtype: tuple[str, str]
"""
strong_start = description.find("<strong>")
if strong_start == -1:
return description.strip(), ""
header = description[:strong_start].strip()
body = description[strong_start:]
return header, body
def parse_header(header: str) -> HeaderInfo:
"""Parse l'en-tête texte pour extraire les métadonnées du cours.
Les labels reconnus sont : ``Matière :``, ``Professeur(s) :``,
``Salle(s) :`` et ``Groupe :``. La recherche se fait par position
des labels, ce qui supporte aussi bien un en-tête multi-lignes
qu'un en-tête dont les lignes sont jointes sur une seule ligne.
:param header: En-tête texte (avant le premier ``<strong>``).
:return: Dictionnaire typé avec les champs subject, teachers, rooms, group.
:rtype: HeaderInfo
"""
info: HeaderInfo = {"subject": "", "teachers": [], "rooms": [], "group": None}
matches = list(_HEADER_LABEL_PATTERN.finditer(header))
for index, match in enumerate(matches):
value_start = match.end()
value_end = matches[index + 1].start() if index + 1 < len(matches) else len(header)
value = unescape(header[value_start:value_end].strip())
label = match.group(1).lower()
if label == "matière":
info["subject"] = value
elif label in ("professeur", "professeurs"):
info["teachers"] = [part.strip() for part in value.split(",") if part.strip()]
elif label in ("salle", "salles"):
info["rooms"] = [part.strip() for part in value.split(",") if part.strip()]
elif label == "groupe":
info["group"] = value
return info
def _strip_html(text: str) -> str:
"""Retire les balises HTML d'un texte et nettoie les espaces.
:param text: Texte pouvant contenir des balises HTML.
:return: Texte brut sans balises, entités HTML décodées.
:rtype: str
"""
cleaned = _TAG_PATTERN.sub("", text)
return unescape(cleaned).strip()
def _parse_french_date(value: str) -> date | None:
"""Convertit une date au format ``JJ/MM/AAAA`` en :class:`date`.
:param value: Date au format ``JJ/MM/AAAA``.
:return: Date parsée, ou ``None`` si le format est invalide.
:rtype: date | None
"""
try:
return datetime.strptime(value, "%d/%m/%Y").date()
except ValueError:
return None
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>``
(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, list[tuple[date, str]], list[tuple[date, str]]]
"""
content: str | None = None
due_blocks: list[tuple[date, str]] = []
assigned_blocks: list[tuple[date, str]] = []
content_match = _CONTENT_PATTERN.search(body)
if content_match is not None:
content = _strip_html(content_match.group(1))
for match in _DUE_PATTERN.finditer(body):
due_date = _parse_french_date(match.group(1))
if due_date is not None:
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.append((assigned_date, _strip_html(match.group(2))))
return content, due_blocks, assigned_blocks
def parse_homework_blocks(
due_blocks: list[tuple[date, str]],
assigned_blocks: list[tuple[date, str]],
) -> tuple[HomeworkBlock, ...]:
"""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 des listes. Tous les blocs
sont préservés, y compris lorsque plusieurs partagent la même date.
: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:
blocks.append(HomeworkBlock(kind="due", date=due_date, text=text, html=text))
for assigned_date, text in assigned_blocks:
blocks.append(HomeworkBlock(kind="assigned", date=assigned_date, text=text, html=text))
return tuple(blocks)
def normalize_homework_text(text: str) -> str:
"""Normalise le texte d'un devoir pour la déduplication.
Unifie les espaces multiples, supprime les balises HTML, puis
applique trim et minuscules.
:param text: Texte brut du devoir.
:return: Texte normalisé.
:rtype: str
"""
normalized = re.sub(r"\s+", " ", text)
normalized = _TAG_PATTERN.sub("", normalized)
return normalized.strip().lower()
def generate_homework_id(due_on: date, normalized_text: str) -> str:
"""Génère un ID stable pour un devoir.
L'ID est la clé ``AAAA-MM-JJ|texte_normalisé`` hachée en SHA-1 dont
on garde les 12 premiers caractères hexadécimaux. Le hachage n'est
pas utilisé à des fins de sécurité (``usedforsecurity=False``).
:param due_on: Date d'échéance du devoir.
:param normalized_text: Texte normalisé du devoir.
:return: ID stable (12 caractères hexadécimaux).
:rtype: str
"""
payload = f"{due_on.isoformat()}|{normalized_text}".encode()
return hashlib.sha1(payload, usedforsecurity=False).hexdigest()[:12]
def collect_homeworks(lessons: list[Lesson], target_date: date) -> list[Homework]:
"""Collecte et déduplique les devoirs en deux passes globales.
Passe 1 : les blocs ``due`` (devoirs à faire pour ``target_date``)
de tous les cours. Passe 2 : les blocs ``assigned`` (devoirs donnés
le jour cible) des cours du jour ``target_date``. La déduplication
se fait par texte normalisé (premier venu, premier servi) et le
résultat est trié par matière puis texte.
:param lessons: Liste de tous les cours (VEVENT) parsés.
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste unique de devoirs, triée par matière puis texte.
:rtype: list[Homework]
"""
by_text: dict[str, Homework] = {}
for lesson in lessons:
for block in lesson.homework_blocks:
if block.kind == "due" and block.date == target_date:
key = normalize_homework_text(block.text)
if key not in by_text:
by_text[key] = Homework(
id=generate_homework_id(target_date, key),
subject=lesson.subject,
teachers=lesson.teachers,
assigned_on=lesson.start.date(),
due_on=block.date,
text=block.text,
html=block.html,
)
for lesson in lessons:
if lesson.start.date() != target_date:
continue
for block in lesson.homework_blocks:
if block.kind == "assigned":
key = normalize_homework_text(block.text)
if key not in by_text:
by_text[key] = Homework(
id=generate_homework_id(target_date, key),
subject=lesson.subject,
teachers=lesson.teachers,
assigned_on=block.date,
due_on=target_date,
text=block.text,
html=block.html,
)
return sorted(by_text.values(), key=lambda hw: (hw.subject.lower(), hw.text.lower()))
def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[SchoolEvent]]:
"""Parse un flux iCal Pronote en événements typés.
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 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).
:param raw_ical: Contenu brut du flux iCal.
:return: Tuple ``(lessons, homeworks, school_events)`` où ``homeworks``
est **toujours vide** : la collecte/déduplication se fait plus tard
via :func:`collect_homeworks` une fois la date cible connue.
:rtype: tuple[list[Lesson], list[Homework], list[SchoolEvent]]
"""
calendar = Calendar.from_ical(raw_ical)
lessons: list[Lesson] = []
homeworks: list[Homework] = []
school_events: list[SchoolEvent] = []
for component in calendar.walk():
if component.name != "VEVENT":
continue
dtstart = component.get("dtstart")
dtend = component.get("dtend")
if dtstart is None or dtend is None:
continue
start = dtstart.dt
end = dtend.dt
categories_obj = component.get("categories")
if categories_obj is None:
categories: list[str] = []
else:
categories = [str(category) for category in categories_obj.cats]
# Événements de type vacances/congés (tout le jour).
if any(cat in ("Congés", "Vacances") for cat in categories):
from_date = start.date() if isinstance(start, datetime) else start
to_date = end.date() if isinstance(end, datetime) else end
summary = component.get("summary")
label = str(summary) if summary is not None else ""
school_events.append(
SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label=label,
from_date=from_date,
to_date=to_date,
)
)
continue
# Les cours sont des événements horodatés ; les événements tout
# le jour non scolaires sont ignorés.
if not isinstance(start, datetime) or not isinstance(end, datetime):
continue
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 ""
header, body = split_header_and_body(description_str)
lesson_data = parse_header(header)
content, due_blocks, assigned_blocks = parse_body(body)
homework_blocks = parse_homework_blocks(due_blocks, assigned_blocks)
uid_value = component.get("uid")
if uid_value is None or str(uid_value).strip() == "":
uid = generate_deterministic_uid(
start=start,
end=end,
subject=lesson_data["subject"],
teachers=lesson_data["teachers"],
rooms=lesson_data["rooms"],
group=lesson_data["group"],
)
else:
uid = normalize_pronote_uid(str(uid_value))
lessons.append(
Lesson(
id=uid,
start=start,
end=end,
subject=lesson_data["subject"],
teachers=tuple(lesson_data["teachers"]),
rooms=tuple(lesson_data["rooms"]),
group=lesson_data["group"],
status=status,
content=content,
homework_blocks=homework_blocks,
)
)
return lessons, homeworks, school_events

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>

57
tests/fixtures/pronote-4e.ics vendored Normal file
View File

@@ -0,0 +1,57 @@
BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Index Education//Pronote//FR
X-WR-CALNAME:Classe de 4e
BEGIN:VEVENT
UID:Edt_12345@index-education.net-20260905T120000Z-Index-Education
DTSTAMP:20260905T120000Z
DTSTART:20260905T080000Z
DTEND:20260905T090000Z
SUMMARY:Mathématiques
CATEGORIES:Cours
DESCRIPTION:<div>
Matière : Mathématiques
Professeur : M. Dupont
Salle : 204
Groupe : Classe entière
<strong>Contenu pédagogique :
</strong>
Résoudre des équations du second degré.
<strong>Pour le 10/09/2026 :
</strong>
Exercices 1 à 5 page 42.
<strong>Donné le 05/09/2026 :
</strong>
Exercices 1 à 5 page 42.
</div>
END:VEVENT
BEGIN:VEVENT
UID:Edt_67890@index-education.net-20260905T120000Z-Index-Education
DTSTAMP:20260905T120000Z
DTSTART:20260905T090000Z
DTEND:20260905T100000Z
SUMMARY:Français
CATEGORIES:Cours - Cours annulé
STATUS:CANCELLED
DESCRIPTION:<div>
Matière : Français
Professeur : Mme Martin
Salle : 205
Groupe : Classe entière
<strong>Contenu pédagogique :
</strong>
Étude d'un texte littéraire.
</div>
END:VEVENT
BEGIN:VEVENT
UID:Edt_11111@index-education.net-20260905T120000Z-Index-Education
DTSTAMP:20260905T120000Z
DTSTART;VALUE=DATE:20261020
DTEND;VALUE=DATE:20261021
SUMMARY:Vacances de la Toussaint
CATEGORIES:Congés
DESCRIPTION:Période de vacances scolaires.
END:VEVENT
END:VCALENDAR

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

1189
tests/unit/test_fallback.py Normal file

File diff suppressed because it is too large Load Diff

600
tests/unit/test_ical.py Normal file
View File

@@ -0,0 +1,600 @@
"""Tests unitaires pour le module iCal : téléchargement et parsing.
Ce module teste :
- La récupération du flux iCal (file://, HTTP)
- Le parsing des événements en modèles Lesson, SchoolEvent
- L'extraction et normalisation des devoirs
- La collecte et déduplication des devoirs par date cible
Les tests utilisent des mocks pour éviter tout appel réseau réel.
"""
from __future__ import annotations
import tempfile
import urllib.parse
from datetime import date, datetime
from pathlib import Path
import pytest
import responses
from pronote_sync.models.agenda import HomeworkBlock, Lesson, LessonStatus
from pronote_sync.sources.pronote.ical import (
collect_homeworks,
fetch_ical,
generate_homework_id,
get_calendar_name,
normalize_homework_text,
parse_body,
parse_ical,
)
@pytest.fixture
def valid_ical_content() -> str:
"""Contenu iCal valide pour tests de parsing."""
return """BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Test//Test//FR
X-WR-CALNAME:Test Calendar
BEGIN:VEVENT
UID:test-1@test.net
DTSTAMP:20260905T120000Z
DTSTART:20260905T080000Z
DTEND:20260905T090000Z
SUMMARY:Math
CATEGORIES:Cours
DESCRIPTION:<div>Matière : Math\nProfesseur : M. Dupont\nSalle : 204\n<strong>Contenu pédagogique :</strong>Résoudre des équations.</div>
END:VEVENT
END:VCALENDAR
"""
@pytest.fixture
def invalid_ical_content() -> str:
"""Contenu iCal invalide (sans BEGIN:VCALENDAR)."""
return "INVALID:CONTENT\nThis is not a valid iCal file."
def test_fetch_ical_file_protocol() -> None:
"""fetch_ical("file://tests/fixtures/pronote-4e.ics") retourne un contenu commençant par BEGIN:VCALENDAR.
:return: None
"""
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
url = f"file://{fixture_path}"
content = fetch_ical(url)
assert content.lstrip().startswith("BEGIN:VCALENDAR")
def test_fetch_ical_file_uri_decoding() -> None:
"""fetch_ical("file://path%20with%20spaces") décode correctement le chemin.
:return: None
"""
# Créer un fichier temporaire avec un espace dans le nom
with tempfile.TemporaryDirectory() as tmpdir:
temp_path = Path(tmpdir) / "fichier avec espaces.ics"
temp_path.write_text(
"BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//Test//Test//FR\nEND:VCALENDAR",
encoding="utf-8",
)
# URL encodée avec espace
encoded_name = urllib.parse.quote("fichier avec espaces.ics")
url = f"file://{tmpdir}/{encoded_name}"
# Cela devrait fonctionner car Path.read_text décode l'URL
content = fetch_ical(url)
assert content.lstrip().startswith("BEGIN:VCALENDAR")
@responses.activate
def test_fetch_ical_invalid_content() -> None:
"""Si le contenu ne commence pas par BEGIN:VCALENDAR, une exception est levée.
:return: None
"""
responses.add(
responses.GET,
"https://example.com/ical.ics",
body="INVALID:CONTENT",
status=200,
)
with pytest.raises(ValueError, match="Flux iCal invalide"):
fetch_ical("https://example.com/ical.ics")
@responses.activate
def test_fetch_ical_http() -> None:
"""Mock de requests.get pour retourner un contenu iCal valide.
:return: None
"""
valid_content = "BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//Test//Test//FR\nEND:VCALENDAR"
responses.add(
responses.GET,
"https://pronote.example.com/ical.ics",
body=valid_content,
status=200,
)
content = fetch_ical("https://pronote.example.com/ical.ics")
assert content == valid_content
@responses.activate
def test_fetch_ical_redacts_errors() -> None:
"""Les messages d'erreur ne contiennent pas l'URL complète (doit être masquée).
:return: None
"""
responses.add(
responses.GET,
"https://pronote.example.com/ical.ics",
body=Exception("Erreur réseau"),
status=500,
)
with pytest.raises(Exception) as exc_info:
fetch_ical("https://pronote.example.com/ical.ics?token=secret123")
error_msg = str(exc_info.value)
# Vérifie que le token secret n'est pas dans le message
assert "secret123" not in error_msg
# Vérifie que l'URL est masquée (utilise redact_url)
assert "https://pronote.example.com/ical.ics" in error_msg
# Le message doit contenir la partie masquée
assert "...ics" in error_msg or "pronote.example.com/ical" in error_msg
def test_get_calendar_name() -> None:
"""get_calendar_name(raw_ical) retourne le nom du calendrier.
:return: None
"""
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
with open(fixture_path, encoding="utf-8") as f:
content = f.read()
name = get_calendar_name(content)
assert name == "Classe de 4e"
def test_get_calendar_name_with_params() -> None:
"""Test avec X-WR-CALNAME;LANGUAGE=fr:TestName.
:return: None
"""
raw_ical = """BEGIN:VCALENDAR
X-WR-CALNAME;LANGUAGE=fr:TestName
END:VCALENDAR
"""
name = get_calendar_name(raw_ical)
assert name == "TestName"
def test_get_calendar_name_none() -> None:
"""Retourne None si X-WR-CALNAME est absent.
:return: None
"""
raw_ical = """BEGIN:VCALENDAR
VERSION:2.0
END:VCALENDAR
"""
name = get_calendar_name(raw_ical)
assert name is None
def test_parse_ical_returns_lessons() -> None:
"""parse_ical(fixture_content) retourne au moins 2 cours.
: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, homeworks, school_events = parse_ical(content)
assert len(lessons) >= 2
def test_parse_ical_detects_cancelled_course() -> None:
"""Un cours a status == LessonStatus.CANCELLED.
: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)
cancelled_lessons = [lesson for lesson in lessons if lesson.status == LessonStatus.CANCELLED]
assert len(cancelled_lessons) >= 1
def test_parse_ical_returns_school_events() -> None:
"""Au moins 1 événement scolaire 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()
_, _, school_events = parse_ical(content)
assert len(school_events) >= 1
def test_parse_ical_homeworks_empty() -> None:
"""La liste des devoirs est toujours vide depuis parse_ical.
:return: None
"""
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
with open(fixture_path, encoding="utf-8") as f:
content = f.read()
_, homeworks, _ = parse_ical(content)
assert homeworks == []
def test_parse_ical_lesson_fields() -> None:
"""Vérifie qu'un cours a les bons champs (matière, profs, salles).
: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)
assert len(lessons) > 0
# Vérifions le premier cours (Mathématiques)
lesson = lessons[0]
assert lesson.subject == "Mathématiques"
assert lesson.teachers == ("M. Dupont",)
assert lesson.rooms == ("204",)
assert lesson.status == LessonStatus.NORMAL
def test_collect_homeworks_dedup() -> None:
"""Étant donné des cours avec des blocs de devoirs, collect_homeworks retourne des devoirs dédupliqués.
:return: None
"""
# Créer des cours avec des blocs de devoirs en double
lesson1 = Lesson(
id="lesson1",
start=datetime(2026, 9, 10, 8, 0),
end=datetime(2026, 9, 10, 9, 0),
subject="Math",
teachers=("M. Dupont",),
rooms=("204",),
group=None,
status=LessonStatus.NORMAL,
content=None,
homework_blocks=(
HomeworkBlock(
kind="due",
date=date(2026, 9, 10),
text="Exercice 1 à 5 page 42",
html="<p>Exercice 1 à 5 page 42</p>",
),
),
)
lesson2 = Lesson(
id="lesson2",
start=datetime(2026, 9, 10, 10, 0),
end=datetime(2026, 9, 10, 11, 0),
subject="Physique",
teachers=("M. Martin",),
rooms=("205",),
group=None,
status=LessonStatus.NORMAL,
content=None,
homework_blocks=(
HomeworkBlock(
kind="due",
date=date(2026, 9, 10),
text="Exercice 1 à 5 page 42",
html="<p>Exercice 1 à 5 page 42</p>",
),
),
)
homeworks = collect_homeworks([lesson1, lesson2], target_date=date(2026, 9, 10))
assert len(homeworks) == 1 # Un seul devoir dédupliqué
def test_collect_homeworks_id_stability() -> None:
"""Deux devoirs avec le même texte et date d'échéance produisent le même ID.
:return: None
"""
lesson1 = Lesson(
id="lesson1",
start=datetime(2026, 9, 10, 8, 0),
end=datetime(2026, 9, 10, 9, 0),
subject="Math",
teachers=("M. Dupont",),
rooms=("204",),
group=None,
status=LessonStatus.NORMAL,
content=None,
homework_blocks=(
HomeworkBlock(
kind="due",
date=date(2026, 9, 10),
text="Devoir commun",
html="<p>Devoir commun</p>",
),
),
)
lesson2 = Lesson(
id="lesson2",
start=datetime(2026, 9, 10, 10, 0),
end=datetime(2026, 9, 10, 11, 0),
subject="Physique",
teachers=("M. Martin",),
rooms=("205",),
group=None,
status=LessonStatus.NORMAL,
content=None,
homework_blocks=(
HomeworkBlock(
kind="due",
date=date(2026, 9, 10),
text="Devoir commun",
html="<p>Devoir commun</p>",
),
),
)
homeworks = collect_homeworks([lesson1, lesson2], target_date=date(2026, 9, 10))
assert len(homeworks) == 1
assert homeworks[0].id == generate_homework_id(date(2026, 9, 10), "devoir commun")
def test_collect_homeworks_sorted() -> None:
"""Les résultats sont triés par (subject.lower(), text.lower()).
:return: None
"""
lesson1 = Lesson(
id="lesson1",
start=datetime(2026, 9, 10, 8, 0),
end=datetime(2026, 9, 10, 9, 0),
subject="Zoologie",
teachers=("M. A",),
rooms=("204",),
group=None,
status=LessonStatus.NORMAL,
content=None,
homework_blocks=(
HomeworkBlock(
kind="due", date=date(2026, 9, 10), text="Devoir B", html="<p>Devoir B</p>"
),
),
)
lesson2 = Lesson(
id="lesson2",
start=datetime(2026, 9, 10, 10, 0),
end=datetime(2026, 9, 10, 11, 0),
subject="Mathématiques",
teachers=("M. B",),
rooms=("205",),
group=None,
status=LessonStatus.NORMAL,
content=None,
homework_blocks=(
HomeworkBlock(
kind="due", date=date(2026, 9, 10), text="Devoir A", html="<p>Devoir A</p>"
),
),
)
homeworks = collect_homeworks([lesson1, lesson2], target_date=date(2026, 9, 10))
assert len(homeworks) == 2
assert homeworks[0].subject == "Mathématiques"
assert homeworks[0].text == "Devoir A"
assert homeworks[1].subject == "Zoologie"
assert homeworks[1].text == "Devoir B"
def test_collect_homeworks_empty() -> None:
"""Aucun bloc de devoir → liste vide.
:return: None
"""
lesson = Lesson(
id="lesson1",
start=datetime(2026, 9, 10, 8, 0),
end=datetime(2026, 9, 10, 9, 0),
subject="Math",
teachers=("M. Dupont",),
rooms=("204",),
group=None,
status=LessonStatus.NORMAL,
content=None,
)
homeworks = collect_homeworks([lesson], target_date=date(2026, 9, 10))
assert homeworks == []
def test_normalize_homework_text() -> None:
"""Vérifie la normalisation des espaces, suppression HTML et minuscules.
:return: None
"""
text = " <p>Exercice 1 à 5</p> \n\n page 42 "
normalized = normalize_homework_text(text)
assert normalized == "exercice 1 à 5 page 42"
def test_generate_homework_id_format() -> None:
"""Retourne un ID de 12 caractères hexadécimaux.
:return: None
"""
due_on = date(2026, 9, 10)
text = "devoir test"
homework_id = generate_homework_id(due_on, text)
assert len(homework_id) == 12
assert all(c in "0123456789abcdef" for c in homework_id)
def test_generate_homework_id_deterministic() -> None:
"""Mêmes entrées → même sortie.
:return: None
"""
due_on = date(2026, 9, 10)
text = "devoir commun"
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

@@ -0,0 +1,439 @@
"""Tests unitaires pour le client Pronote via pronotepy.
Ce module vérifie le comportement du client ``PronoteClient`` et de son
protocole ``PronoteClientProtocol``. Tous les tests sont unitaires et
utilisent des mocks pour éviter tout accès réseau réel à Pronote.
"""
from __future__ import annotations
from datetime import date, datetime
import pronotepy
import pytest
import pytest_mock
from pydantic import SecretStr
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.models.agenda import Lesson, LessonStatus
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
from pronote_sync.sources.pronote.client import PronoteClient, PronoteClientProtocol
# --- Protocol tests ---
def test_protocol_methods(mocker: pytest_mock.MockerFixture) -> None:
"""Vérifie que le protocole PronoteClientProtocol expose les méthodes attendues.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
assert hasattr(PronoteClientProtocol, "get_messages")
assert hasattr(PronoteClientProtocol, "get_informations")
assert hasattr(PronoteClientProtocol, "get_lessons")
assert hasattr(PronoteClientProtocol, "get_homeworks")
# --- Client with mocked pronotepy ---
@pytest.fixture
def pronote_settings() -> PronoteSettings:
"""Fournit des paramètres Pronote valides pour les tests.
:return: Instance de PronoteSettings avec des valeurs par défaut.
:rtype: PronoteSettings
"""
return PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="parent",
)
@pytest.fixture
def empty_pronote_settings() -> PronoteSettings:
"""Fournit des paramètres Pronote vides pour les tests.
:return: Instance de PronoteSettings avec tous les champs à None.
:rtype: PronoteSettings
"""
return PronoteSettings(username=None, password=None, ent=None)
def test_get_messages_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_messages retourne une liste de Message en cas de succès.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
# Mock du client pronotepy
mock_client = mocker.MagicMock()
mock_discussion = mocker.MagicMock()
mock_discussion.subject = "Test Subject"
mock_message = mocker.MagicMock()
mock_message.id = "msg-123"
mock_message.content = "Test message content"
mock_message.author = "Teacher Test"
mock_message.created = datetime(2024, 9, 1, 10, 0, 0)
mock_message.seen = True
mock_discussion.messages = [mock_message]
mock_client.discussions.return_value = [mock_discussion]
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_messages()
assert isinstance(messages, list)
assert len(messages) == 1
message = messages[0]
assert isinstance(message, Message)
assert message.id == "msg-123"
assert message.type == MessageType.DISCUSSION
assert message.title == "Test Subject"
assert message.content == "Test message content"
assert message.author == "Teacher Test"
assert message.date == datetime(2024, 9, 1, 10, 0, 0)
assert message.read is True
def test_get_messages_empty_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_messages retourne une liste vide en cas d'erreur API.
: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 = pronotepy.PronoteAPIError("API error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_messages()
assert messages == []
def test_get_informations_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_informations retourne une liste de Message 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_info = mocker.MagicMock()
mock_info.id = "info-456"
mock_info.title = "Important Info"
mock_info.content.return_value = "Important content"
mock_info.author = "Admin"
mock_info.creation_date = datetime(2024, 9, 2, 14, 30, 0)
mock_info.read = False
mock_info.survey = True
mock_client.information_and_surveys.return_value = [mock_info]
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
assert isinstance(messages, list)
assert len(messages) == 1
message = messages[0]
assert isinstance(message, Message)
assert message.id == "info-456"
assert message.type == MessageType.SURVEY
assert message.title == "Important Info"
assert message.content == "Important content"
assert message.author == "Admin"
assert message.date == datetime(2024, 9, 2, 14, 30, 0)
assert message.read is False
def test_get_informations_empty_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_informations retourne une liste vide en cas d'erreur API.
: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 = pronotepy.PronoteAPIError("API error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
assert messages == []
def test_get_lessons_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""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.
:return: None
"""
mock_client = mocker.MagicMock()
# Mock des cours
mock_lesson = mocker.MagicMock()
mock_lesson.id = "lesson-789"
mock_lesson.start = datetime(2024, 9, 1, 8, 0, 0)
mock_lesson.end = datetime(2024, 9, 1, 9, 30, 0)
mock_lesson.subject = mocker.MagicMock()
mock_lesson.subject.name = "Maths"
mock_lesson.teacher_names = ["Prof A", "Prof B"]
mock_lesson.classrooms = ["Salle 101", "Salle 102"]
mock_lesson.group_name = "Classe 1"
mock_lesson.canceled = False
mock_content = mocker.MagicMock()
mock_content.description = "Lesson content"
mock_lesson.content = mock_content
mock_client.lessons.return_value = [mock_lesson]
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
lessons = client.get_lessons(date(2024, 9, 1), date(2024, 9, 30))
assert isinstance(lessons, list)
assert len(lessons) == 1
lesson = lessons[0]
assert isinstance(lesson, Lesson)
assert lesson.id == "lesson-789"
assert lesson.start == datetime(2024, 9, 1, 8, 0, 0)
assert lesson.end == datetime(2024, 9, 1, 9, 30, 0)
assert lesson.subject == "Maths"
assert lesson.teachers == ("Prof A", "Prof B")
assert lesson.rooms == ("Salle 101", "Salle 102")
assert lesson.group == "Classe 1"
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]
assert isinstance(homework, Homework)
assert homework.id == "hw-101"
assert homework.subject == "Maths"
assert homework.teachers == ()
assert homework.assigned_on is None
assert homework.due_on == date(2024, 9, 15)
assert homework.text == "Do your homework"
assert homework.html == "Do your homework"
def test_get_lessons_propagates_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_lessons 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.lessons.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_lessons(date(2024, 9, 1), date(2024, 9, 30))
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)
with pytest.raises(ValueError, match="pronote_url, username, password et ent sont requis"):
client._connect()
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 account_type='parent' utilise pronotepy.ParentClient.
: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
mock_client = mocker.MagicMock()
mock_client_class = Mock(return_value=mock_client)
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
mocker.patch("pronotepy.Client")
client = PronoteClient(pronote_settings)
_ = client._connect()
# 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