Compare commits

..

9 Commits

Author SHA1 Message Date
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
13 changed files with 1672 additions and 717 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

@@ -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": 4893,
"is_secret": false
} }
] ]
}, },
"generated_at": "2026-09-05T21:51:55Z" "generated_at": "2026-09-06T14:37:44Z"
} }

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).

View File

@@ -47,8 +47,10 @@ Le projet doit implémenter les fonctionnalités suivantes, dans l'ordre logique
- **Idempotence** : Deux exécutions identiques sans changement externe **doivent** produire le même résultat (aucune modification en base ou CalDAV). - **Idempotence** : Deux exécutions identiques sans changement externe **doivent** produire le même résultat (aucune modification en base ou CalDAV).
- **Mode dégradé** : - **Mode dégradé** :
- Si la synthèse IA échoue → envoyer le message **sans synthèse** (mais avec la liste brute des devoirs). - Si la synthèse IA échoue → envoyer le message **sans synthèse** (mais avec la liste brute des devoirs).
- Si `pronotepy` échoue → basculer sur iCal (si disponible) pour l'agenda/devoirs. - En mode source `auto`, si iCal échoue → basculer sur `pronotepy` pour
- Si iCal et `pronotepy` échouent → **échec explicite** avec message clair. l'agenda/devoirs.
- En mode source explicite (`ical` ou `pronotepy`), ne pas basculer silencieusement.
- En mode `auto`, si iCal et `pronotepy` échouent → **échec explicite** avec message clair.
--- ---
@@ -260,15 +262,15 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
| Variable | Description | Exemple (anonymisé) | Type | | Variable | Description | Exemple (anonymisé) | Type |
|------------------------------|-----------------------------------------------------------------------------|---------------------------------------------|---------------| |------------------------------|-----------------------------------------------------------------------------|---------------------------------------------|---------------|
| `PRONOTE_ICAL_URL` | URL du flux iCal Pronote (contient `icalsecurise`). | `https://college.ent/pronote/ical/...` | `str` | | `PRONOTE_URL` | URL de la page Pronote utilisée par `pronotepy` (page parent). | `https://college.ent/pronote/parent.html` | `str` |
| `PRONOTE_ICAL_URL` | URL du flux iCal Pronote (contient `icalsecurise`). | `https://college.ent/pronote/ical/...` | `SecretStr` |
| `PRONOTE_USERNAME` | Identifiant Pronote (si `pronotepy` utilisé). | `parent.dupont` | `str` | | `PRONOTE_USERNAME` | Identifiant Pronote (si `pronotepy` utilisé). | `parent.dupont` | `str` |
| `PRONOTE_PASSWORD` | Mot de passe Pronote (si `pronotepy` utilisé). | `SecretStr` (masqué) | `SecretStr` | | `PRONOTE_PASSWORD` | Mot de passe Pronote (si `pronotepy` utilisé). | `SecretStr` (masqué) | `SecretStr` |
| `PRONOTE_ENT` | ENT Pronote (ex: `monbureaunumerique`, `atrium`). | `monbureaunumerique` | `str` | | `PRONOTE_ENT` | Slug ENT supporté, résolu vers une fonction de `pronotepy.ent`. | `monbureaunumerique` | `str` |
| `CALDAV_URL` | URL du serveur CalDAV. | `https://caldav.example.com/calendars/...` | `str` | | `CALDAV_URL` | URL du serveur CalDAV. | `https://caldav.example.com/calendars/...` | `str` |
| `CALDAV_USERNAME` | Identifiant CalDAV. | `user@example.com` | `str` | | `CALDAV_USERNAME` | Identifiant CalDAV. | `user@example.com` | `str` |
| `CALDAV_PASSWORD` | Mot de passe CalDAV. | `SecretStr` (masqué) | `SecretStr` | | `CALDAV_PASSWORD` | Mot de passe CalDAV. | `SecretStr` (masqué) | `SecretStr` |
| `CALDAV_CALENDAR_PATH` | Chemin du calendrier CalDAV de destination. | `/pronote-sync/` | `str` | | `CALDAV_CALENDAR_PATH` | Chemin du calendrier CalDAV de destination. | `/pronote-sync/` | `str` |
| `CALDAV_CALENDAR_PATH` | Chemin du calendrier CalDAV de destination. | `/pronote-sync/` | `str` |
| `XMPP_JID` | Identifiant XMPP (ex: `user@example.com`). | `user@example.com` | `str` | | `XMPP_JID` | Identifiant XMPP (ex: `user@example.com`). | `user@example.com` | `str` |
| `XMPP_PASSWORD` | Mot de passe XMPP. | `SecretStr` (masqué) | `SecretStr` | | `XMPP_PASSWORD` | Mot de passe XMPP. | `SecretStr` (masqué) | `SecretStr` |
| `XMPP_RECIPIENT` | Destinataire XMPP (ex: `parent@example.com`). | `parent@example.com` | `str` | | `XMPP_RECIPIENT` | Destinataire XMPP (ex: `parent@example.com`). | `parent@example.com` | `str` |
@@ -278,6 +280,19 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
> Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`. > Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`.
> Une section `BLOG_ENABLED` et `BLOG_RSS_URL` a été ajoutée dans `.env.example`. > Une section `BLOG_ENABLED` et `BLOG_RSS_URL` a été ajoutée dans `.env.example`.
Les variables Pronote sont obligatoires selon les sources activées :
- la source iCal exige `PRONOTE_ICAL_URL` ;
- la source `pronotepy` exige `PRONOTE_URL`, `PRONOTE_USERNAME` et
`PRONOTE_PASSWORD` ;
- `PRONOTE_ENT` reste optionnel pour une connexion directe, mais, s'il est fourni, son slug doit
appartenir à une liste fermée et être résolu vers la fonction correspondante de `pronotepy.ent`.
`PRONOTE_URL` et `PRONOTE_ICAL_URL` sont deux contrats distincts : l'un ne doit jamais être déduit
de l'autre. Le cas d'usage actuel est un compte parent ; le client à construire est donc
`pronotepy.ParentClient`. Une généralisation à plusieurs profils ne sera ajoutée qu'en présence
d'un besoin réel et testé.
#### 3.1.2 Variables optionnelles #### 3.1.2 Variables optionnelles
| Variable | Description | Valeur par défaut | Type | | Variable | Description | Valeur par défaut | Type |
@@ -305,6 +320,7 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
```ini ```ini
# --- Pronote --- # --- Pronote ---
PRONOTE_URL=https://college.ent/pronote/parent.html
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_USERNAME=parent.dupont PRONOTE_USERNAME=parent.dupont
PRONOTE_PASSWORD=your_secure_password PRONOTE_PASSWORD=your_secure_password
@@ -358,17 +374,18 @@ LOG_LEVEL=INFO
> `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"` (et non `"pronote-digest"`). > `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"` (et non `"pronote-digest"`).
```python ```python
from typing import Literal, Optional from typing import Literal
from pydantic import SecretStr, Field from pydantic import SecretStr, Field
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
class PronoteSettings(BaseSettings): class PronoteSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="PRONOTE_", env_file=".env", extra="ignore") model_config = SettingsConfigDict(env_prefix="PRONOTE_", env_file=".env", extra="ignore")
ical_url: Optional[str] = None url: str | None = None
username: Optional[str] = None ical_url: SecretStr | None = None
password: Optional[SecretStr] = None username: str | None = None
ent: Optional[str] = None password: SecretStr | None = None
ent: str | None = None
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"
@@ -402,11 +419,11 @@ class AppSettings(BaseSettings):
class Settings(BaseSettings): class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore") model_config = SettingsConfigDict(env_file=".env", extra="ignore")
pronote: PronoteSettings = PronoteSettings() pronote: PronoteSettings = Field(default_factory=PronoteSettings)
caldav: CalDAVSettings = CalDAVSettings() caldav: CalDAVSettings = Field(default_factory=CalDAVSettings)
xmpp: XmppSettings = XmppSettings() xmpp: XmppSettings = Field(default_factory=XmppSettings)
ai: AISettings = AISettings() ai: AISettings = Field(default_factory=AISettings)
app: AppSettings = AppSettings() app: AppSettings = Field(default_factory=AppSettings)
settings = Settings() settings = Settings()
@@ -431,47 +448,20 @@ settings = Settings()
**Note importante** : Tous les messages d'erreur externes (HTTP, Pronote, CalDAV, XMPP, IA) **doivent** être systématiquement expurgés des secrets avant journalisation ou réémission. Utiliser `redact_secrets(str(e))` ou `redact_exception()` pour les logs. **Note importante** : Tous les messages d'erreur externes (HTTP, Pronote, CalDAV, XMPP, IA) **doivent** être systématiquement expurgés des secrets avant journalisation ou réémission. Utiliser `redact_secrets(str(e))` ou `redact_exception()` pour les logs.
**Tests négatifs recommandés** : **Tests négatifs obligatoires** : injecter des sentinelles distinctes dans l'URL, les identifiants,
- Vérifier que les messages d'erreur ne contiennent ni `icalsecurise`, ni `SECRET`, ni mots de passe, ni clés API. le mot de passe et la clé API, provoquer une erreur externe, puis vérifier leur absence dans le
- Exemple de test : message, les logs, `__cause__`, `__context__` et le traceback formaté.
```python
def test_error_messages_do_not_leak_secrets():
"""Vérifie que les messages d'erreur ne fuient pas de secrets."""
from pronote_sync.utils.redaction import redact_secrets
from pronote_sync.sources.pronote.ical import fetch_ical
# Simuler une URL avec token
bad_url = "https://example.com/ical?icalsecurise=SECRET_TOKEN"
try:
fetch_ical(bad_url)
except Exception as e:
error_msg = str(e)
assert "SECRET_TOKEN" not in error_msg
assert "icalsecurise" not in error_msg.lower()
```
**Tests négatifs recommandés** :
- Vérifier que les tokens (`icalsecurise`), clés API et URLs ne apparaissent **jamais** dans les logs ou les messages d'erreur, même en cas d'exception.
- Exemple de test :
```python
def test_error_messages_do_not_leak_secrets():
"""Vérifie que les messages d'erreur ne fuient pas de secrets."""
from pronote_sync.utils.redaction import redact_secrets
from pronote_sync.sources.pronote.ical import fetch_ical
# Simuler une URL avec token
bad_url = "https://example.com/ical?icalsecurise=SECRET_TOKEN"
try:
fetch_ical(bad_url)
except Exception as e:
error_msg = str(e)
assert "SECRET_TOKEN" not in error_msg
assert "icalsecurise" not in error_msg.lower()
```
**Application systématique** : **Application systématique** :
Toutes les exceptions externes (HTTP, Pronote, CalDAV, XMPP, IA) **doivent** être traitées avec `redact_secrets()` ou `redact_exception()` avant toute journalisation ou réémission. Toutes les exceptions externes (HTTP, Pronote, CalDAV, XMPP, IA) **doivent** être traitées avec `redact_secrets()` ou `redact_exception()` avant toute journalisation ou réémission.
Le masquage du message extérieur ne suffit pas si l'exception brute reste chaînée dans
`__cause__` ou `__context__` : un traceback complet pourrait alors révéler l'URL ou les
identifiants d'origine. À la frontière avec une bibliothèque externe, journaliser uniquement la
version expurgée puis lever l'exception applicative avec `raise ... from None`, ou chaîner une
cause elle-même expurgée. Les tests de non-fuite doivent inspecter `str(exc)`, les logs, la cause,
le contexte et le traceback complet.
#### 4.2.1 Masquage des URLs (`redaction.py`) #### 4.2.1 Masquage des URLs (`redaction.py`)
> ⚠️ **Décision d'implémentation** : > ⚠️ **Décision d'implémentation** :
@@ -1349,6 +1339,40 @@ def _format_message(self, message: XmppMessage) -> str:
### 5.1 Flux iCal Pronote ### 5.1 Flux iCal Pronote
#### Contrat normatif des sources Pronote (M4 et jalons suivants)
Les règles ci-dessous priment sur les exemples historiques de cette section :
1. `parse_ical()` retourne les cours et événements scolaires. Sa liste de `Homework` reste vide :
les blocs bruts conservés dans chaque `Lesson` sont transformés ensuite par
`collect_homeworks(lessons, target_date)`.
2. Les blocs de devoirs sont conservés dans une séquence. Une structure `date -> texte` est
interdite, car plusieurs devoirs peuvent partager la même date. La déduplication ne s'effectue
qu'au moment de `collect_homeworks`.
3. Un cours est annulé si `STATUS:CANCELLED` **ou** la catégorie Pronote correspondante est
présente. Le statut déplacé est détecté par sa catégorie.
4. Le client `pronotepy` reçoit l'URL Pronote en premier argument, puis les identifiants, avec une
fonction ENT résolue depuis une liste fermée. Pour le compte parent actuellement visé, utiliser
`pronotepy.ParentClient(pronote_url, username, password, ent=ent_function)`.
5. Le client expose séparément la récupération des cours et celle des devoirs. Les devoirs
`pronotepy` sont filtrés strictement sur `due_on == target_date` avant d'être retournés au
pipeline.
6. Une liste vide est un résultat valide ; une exception signale un échec de source. Les méthodes
critiques d'agenda et de devoirs propagent donc une erreur expurgée au `PronoteFetcher`. Les
messages et informations, non critiques, peuvent se dégrader en listes vides accompagnées d'un
warning.
7. En mode `auto`, iCal est essayé en premier puis `pronotepy` sert de repli. Les modes explicites
`ical` et `pronotepy` sont stricts et ne changent pas silencieusement de source. En mode `auto`,
l'échec des deux sources lève `PipelineCriticalError`.
8. Pendant une exécution du pipeline, un flux iCal déjà téléchargé et parsé est réutilisé pour
l'agenda et les devoirs. Ce partage reste limité à l'exécution courante : aucun cache global ou
persistant n'est nécessaire.
Avant M7, une fixture anonymisée doit confirmer que deux événements équivalents provenant d'iCal
et de `pronotepy` aboutissent au même identifiant canonique. Si ce n'est pas le cas, la
normalisation doit être corrigée à la frontière des sources avant toute synchronisation CalDAV ;
ne pas introduire de moteur de rapprochement complexe sans données qui le justifient.
#### 5.1.1 Observations sur les flux réels #### 5.1.1 Observations sur les flux réels
Les flux iCal générés par Pronote ont des **spécificités importantes** à prendre en compte, basées sur l'analyse du projet TypeScript `pronote-digest` : Les flux iCal générés par Pronote ont des **spécificités importantes** à prendre en compte, basées sur l'analyse du projet TypeScript `pronote-digest` :
@@ -1551,7 +1575,7 @@ def fetch_ical(url: str, timeout: int = 20) -> str:
safe_error = redact_secrets(str(e)) safe_error = redact_secrets(str(e))
raise requests.exceptions.RequestException( raise requests.exceptions.RequestException(
f"Échec de la récupération de {safe_url}: {safe_error}" f"Échec de la récupération de {safe_url}: {safe_error}"
) from e ) from None
# Validation du flux # Validation du flux
if "BEGIN:VCALENDAR" not in content: if "BEGIN:VCALENDAR" not in content:
@@ -2073,7 +2097,8 @@ def parse_ical(raw_ical: str) -> tuple[List[Lesson], List[HomeworkModel], List[S
# Cours annulés ou déplacés # Cours annulés ou déplacés
status = LessonStatus.NORMAL status = LessonStatus.NORMAL
if "Cours - Cours annulé" in categories: ical_status = str(component.get("status", "")).upper()
if ical_status == "CANCELLED" or "Cours - Cours annulé" in categories:
status = LessonStatus.CANCELLED status = LessonStatus.CANCELLED
elif "Cours - Cours déplacé" in categories: elif "Cours - Cours déplacé" in categories:
status = LessonStatus.MOVED status = LessonStatus.MOVED
@@ -2111,274 +2136,72 @@ def parse_ical(raw_ical: str) -> tuple[List[Lesson], List[HomeworkModel], List[S
``` ```
#### 5.1.7 Client `pronotepy` pour messages et informations #### 5.1.7 Client `pronotepy`
`pronotepy` est utilisé **uniquement** pour : `pronotepy` fournit les cours et devoirs de repli, ainsi que les messages, informations et
- Les **messages** des professeurs (`client.get_discussions()`). sondages. La signature réelle de la bibliothèque doit être respectée ; l'URL Pronote est le
- Les **informations et sondages** (`client.get_information_and_surveys()`). premier argument et l'ENT est une fonction, pas une chaîne :
```python ```python
from typing import List, Optional import pronotepy.ent as pronote_ent
from pronotepy import Client, PronoteAPIError from pronotepy import ParentClient
from ..models.message import Message, MessageType
from ..redaction import redact_url
import logging
logger = logging.getLogger(__name__) ENT_RESOLVERS = {
"monbureaunumerique": pronote_ent.monbureaunumerique,
# Ajouter uniquement les ENT effectivement pris en charge et testés.
}
ent_function = ENT_RESOLVERS.get(settings.ent) if settings.ent else None
if settings.ent and ent_function is None:
raise ValueError("PRONOTE_ENT n'est pas pris en charge")
if settings.url is None or settings.username is None or settings.password is None:
raise ValueError("Configuration pronotepy incomplète")
class PronoteClient: client = ParentClient(
"""Client pour interagir avec Pronote via pronotepy.""" settings.url,
settings.username,
settings.password.get_secret_value(),
ent=ent_function,
)
```
def __init__( Le client applicatif expose des méthodes distinctes :
self,
username: Optional[str] = None,
password: Optional[str] = None,
ent: Optional[str] = None,
ical_url: Optional[str] = None,
):
self.username = username
self.password = password
self.ent = ent
self.ical_url = ical_url
self._client: Optional[Client] = None
def _get_client(self) -> Client: - `get_lessons(start, end) -> list[Lesson]` ;
"""Initialise et retourne le client pronotepy.""" - `get_homeworks(start, end) -> list[Homework]` ;
if self._client is None: - `get_messages() -> list[Message]` ;
if not all([self.username, self.password, self.ent]): - `get_informations() -> list[Message]`.
raise ValueError("Username, password et ENT sont requis pour pronotepy")
self._client = Client( Cette séparation évite qu'un appel agenda récupère inutilement les devoirs, et inversement. Les
self.username, méthodes agenda/devoirs ne transforment jamais une erreur en liste vide : elles journalisent une
self.password, version expurgée puis lèvent une erreur expurgée avec `from None`. Les méthodes de messages et
self.ent, d'informations sont non critiques et peuvent retourner une liste vide avec un warning.
)
return self._client
def get_messages(self) -> List[Message]:
"""Récupère les messages des professeurs."""
try:
client = self._get_client()
discussions = client.get_discussions()
messages = []
for discussion in discussions:
for message in discussion.messages:
messages.append(Message(
id=str(message.id),
type=MessageType.DISCUSSION,
title=message.title,
content=message.content,
author=message.author,
date=message.date,
read=message.is_read,
))
return messages
except PronoteAPIError as e:
logger.error(f"Échec de la récupération des messages Pronote: {redact_secrets(str(e))}")
return []
def get_informations(self) -> List[Message]:
"""Récupère les informations et sondages."""
try:
client = self._get_client()
informations = client.get_information_and_surveys()
messages = []
for info in informations:
messages.append(Message(
id=str(info.id),
type=MessageType.INFORMATION,
title=info.title,
content=info.content,
author=info.author,
date=info.date,
read=False, # Par défaut non lu
))
return messages
except PronoteAPIError as e:
logger.error(f"Échec de la récupération des informations Pronote: {redact_secrets(str(e))}")
return []
def get_agenda_fallback(self) -> tuple[List[Lesson], List[HomeworkModel]]:
"""
Récupère l'agenda et les devoirs via pronotepy (repli si iCal échoue).
**À utiliser uniquement si PRONOTE_AGENDA_SOURCE=pronotepy ou PRONOTE_HOMEWORK_SOURCE=pronotepy**.
"""
try:
client = self._get_client()
lessons = []
for lesson in client.get_lessons():
lessons.append(Lesson(
id=str(lesson.id),
start=lesson.start,
end=lesson.end,
subject=lesson.subject,
teachers=[t.name for t in lesson.teachers],
rooms=[r.name for r in lesson.rooms],
status=LessonStatus.NORMAL, # À adapter selon les données
content=lesson.content,
))
homeworks = []
for hw in client.get_homework():
homeworks.append(HomeworkModel(
id=str(hw.id),
subject=hw.subject,
teachers=[t.name for t in hw.teachers],
assigned_on=hw.given_date,
due_on=hw.due_date,
text=hw.description,
html=hw.description, # pronotepy ne fournit pas de HTML
))
return lessons, homeworks
except PronoteAPIError as e:
logger.error(f"Échec de la récupération de l'agenda via pronotepy: {redact_secrets(str(e))}")
return [], []
def close(self) -> None:
"""Fermeture du client."""
if self._client:
self._client.close()
self._client = None
Les objets renvoyés par `client.homework(start, end)` couvrent une fenêtre. Le résultat destiné à
un jour cible est donc filtré explicitement sur `homework.date == target_date`.
#### 5.1.8 Logique de repli (`sources/pronote/fallback.py`) #### 5.1.8 Logique de repli (`sources/pronote/fallback.py`)
```python Le `PronoteFetcher` dépend de `Settings` et d'un protocole de client injecté ; il ne construit pas
from typing import Literal, Optional de singleton et ne contient pas d'identifiants dupliqués.
from enum import Enum
from .ical import fetch_ical, parse_ical
from .client import PronoteClient
from ..models.agenda import Lesson, Homework
| Mode | Comportement agenda/devoirs |
|------|------------------------------|
| `ical` | iCal uniquement ; toute erreur devient critique. |
| `pronotepy` | `pronotepy` uniquement ; toute erreur devient critique. |
| `auto` | iCal d'abord, puis `pronotepy` uniquement si iCal lève une erreur. |
| `auto`, deux échecs | Lever `PipelineCriticalError` avec un message expurgé. |
class AgendaSource(Enum): Une réponse vide est un succès et ne déclenche pas de repli : une journée peut réellement ne
AUTO = "auto" contenir aucun cours ou devoir. Inversement, une exception ne doit jamais être convertie en
ICAL = "ical" `([], [])`, car le `PronoteFetcher` perdrait alors l'information nécessaire pour distinguer un
PRONOTEPY = "pronotepy" échec d'un résultat vide.
class PronoteFetcher:
"""Gère la récupération des données Pronote avec repli."""
def __init__(
self,
ical_url: Optional[str] = None,
username: Optional[str] = None,
password: Optional[str] = None,
ent: Optional[str] = None,
agenda_source: str = "auto",
homework_source: str = "auto",
):
self.ical_url = ical_url
self.username = username
self.password = password
self.ent = ent
self.agenda_source = AgendaSource(agenda_source)
self.homework_source = AgendaSource(homework_source)
self._pronote_client: Optional[PronoteClient] = None
def _get_pronote_client(self) -> PronoteClient:
if self._pronote_client is None:
self._pronote_client = PronoteClient(
username=self.username,
password=self.password,
ent=self.ent,
ical_url=self.ical_url,
)
return self._pronote_client
def fetch_agenda(self) -> tuple[List[Lesson], List[Homework]]:
"""Récupère l'agenda selon la source configurée (`agenda_source`)."""
if self.agenda_source == AgendaSource.ICAL:
return self._fetch_agenda_ical()
elif self.agenda_source == AgendaSource.PRONOTEPY:
return self._fetch_agenda_pronotepy()
else: # AUTO
# Essayer iCal d'abord
try:
lessons, homeworks = self._fetch_agenda_ical()
if lessons or homeworks:
return lessons, homeworks
except Exception as e:
logger.warning(f"Échec de la récupération iCal pour l'agenda: {redact_secrets(str(e))}")
# Repli sur pronotepy
logger.info("Repli sur pronotepy pour l'agenda.")
return self._fetch_agenda_pronotepy()
def fetch_homework(self) -> List[Homework]:
"""Récupère les devoirs selon la source configurée (`homework_source`)."""
if self.homework_source == AgendaSource.ICAL:
# Récupérer uniquement les devoirs depuis iCal
try:
_, homeworks = self._fetch_agenda_ical()
return homeworks
except Exception as e:
logger.warning(f"Échec de la récupération iCal pour les devoirs: {redact_secrets(str(e))}")
return []
elif self.homework_source == AgendaSource.PRONOTEPY:
# Récupérer uniquement les devoirs depuis pronotepy
try:
_, homeworks = self._fetch_agenda_pronotepy()
return homeworks
except Exception as e:
logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}")
return []
else: # AUTO
# Essayer iCal d'abord
try:
_, homeworks = self._fetch_agenda_ical()
if homeworks:
return homeworks
except Exception as e:
logger.warning(f"Échec de la récupération iCal pour les devoirs: {redact_secrets(str(e))}")
# Repli sur pronotepy
logger.info("Repli sur pronotepy pour les devoirs.")
try:
_, homeworks = self._fetch_agenda_pronotepy()
return homeworks
except Exception as e:
logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}")
return []
def _fetch_agenda_ical(self) -> tuple[List[Lesson], List[Homework]]:
"""Récupère l'agenda depuis iCal."""
if not self.ical_url:
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
raw_ical = fetch_ical(self.ical_url)
lessons, homeworks, _ = parse_ical(raw_ical)
return lessons, homeworks
def _fetch_agenda_pronotepy(self) -> tuple[List[Lesson], List[Homework]]:
"""Récupère l'agenda depuis pronotepy."""
client = self._get_pronote_client()
lessons, homeworks = client.get_agenda_fallback()
return lessons, homeworks
def fetch_messages(self) -> List[Message]:
"""Récupère les messages (toujours via pronotepy)."""
client = self._get_pronote_client()
return client.get_messages()
def fetch_informations(self) -> List[Message]:
"""Récupère les informations (toujours via pronotepy)."""
client = self._get_pronote_client()
return client.get_informations()
def close(self) -> None:
"""Fermeture des ressources."""
if self._pronote_client:
self._pronote_client.close()
self._pronote_client = None
```
`fetch_homework(target_date)` applique le même contrat aux deux sources. Pour iCal, il appelle
`collect_homeworks(lessons, target_date)`. Pour `pronotepy`, il filtre les devoirs récupérés sur la
même date cible. En M11, la composition root fournit un contexte d'exécution permettant de
réutiliser le même téléchargement/parsing iCal pour l'agenda et les devoirs lorsque les deux
sélections le permettent.
### 5.2 Résumé des points clés ### 5.2 Résumé des points clés
@@ -4546,8 +4369,10 @@ def get_channel(settings: Settings, channel_name: str = "xmpp") -> Channel:
- **Ne jamais bloquer le pipeline** : Une erreur dans une étape ne doit pas empêcher les autres étapes de s'exécuter (sauf si critique). - **Ne jamais bloquer le pipeline** : Une erreur dans une étape ne doit pas empêcher les autres étapes de s'exécuter (sauf si critique).
- **Modes dégradés** : - **Modes dégradés** :
- **Synthèse IA** : Si elle échoue → envoyer le message **sans synthèse** (mais avec la liste brute des devoirs). - **Synthèse IA** : Si elle échoue → envoyer le message **sans synthèse** (mais avec la liste brute des devoirs).
- **pronotepy** : Si la récupération échoue → basculer sur **iCal** (si disponible). - **Sources Pronote en mode `auto`** : essayer iCal, puis basculer sur `pronotepy`
- **iCal** : Si la récupération échoue → basculer sur **pronotepy** (si configuré). uniquement si iCal lève une erreur.
- **Source Pronote explicite** : `ical` et `pronotepy` sont des modes stricts, sans repli
implicite.
- **CalDAV** : Si la synchronisation échoue → **logger l'erreur** mais continuer le pipeline. - **CalDAV** : Si la synchronisation échoue → **logger l'erreur** mais continuer le pipeline.
- **XMPP** : Si l'envoi échoue → **logger l'erreur** mais continuer le pipeline. - **XMPP** : Si l'envoi échoue → **logger l'erreur** mais continuer le pipeline.
- **Erreurs critiques** : - **Erreurs critiques** :
@@ -4556,6 +4381,9 @@ def get_channel(settings: Settings, channel_name: str = "xmpp") -> Channel:
### 11.2 Hiérarchie des erreurs ### 11.2 Hiérarchie des erreurs
La hiérarchie canonique réside dans `pronote_sync/errors.py`. Les jalons suivants la complètent si
nécessaire mais ne créent pas une seconde hiérarchie dans `pipeline/steps/errors.py`.
```python ```python
from enum import Enum, auto from enum import Enum, auto
from typing import Optional from typing import Optional
@@ -4610,7 +4438,7 @@ from ..models.xmpp import XmppMessage
from ..models.pronote import PronoteData from ..models.pronote import PronoteData
from ..models.sync import CalDAVSyncResult from ..models.sync import CalDAVSyncResult
from ..models.synthesis import SynthesisInput, SynthesisResult from ..models.synthesis import SynthesisInput, SynthesisResult
from ..sources.pronote.fetcher import PronoteFetcher from ..sources.pronote.fallback import PronoteFetcher
from ..sync.caldav import CalDAVClient from ..sync.caldav import CalDAVClient
from ..sync.diff import AgendaComparator from ..sync.diff import AgendaComparator
from ..synthesis.provider import SynthesisProvider from ..synthesis.provider import SynthesisProvider
@@ -4818,78 +4646,26 @@ class PipelineRunner:
Chaque étape du pipeline est **isolée** et peut lever des `PipelineError` ou `PipelineWarning`. Chaque étape du pipeline est **isolée** et peut lever des `PipelineError` ou `PipelineWarning`.
#### 11.4.1 `fetch_step.py` #### 11.4.1 `fetch.py`
```python L'étape de récupération orchestre le contrat de `PronoteFetcher` sans réimplémenter la sélection
from typing import Tuple, List des sources :
from ..models.agenda import Lesson, Homework, SchoolEvent
from ..models.message import Message
from ..sources.pronote.fetcher import PronoteFetcher
from ..utils.redaction import redact_secrets
from .errors import PipelineError, ErrorSeverity, PipelineCriticalError
1. récupérer l'agenda et les événements scolaires ;
2. résoudre le jour cible ;
3. récupérer les devoirs pour cette date cible ;
4. récupérer les messages et informations non critiques ;
5. assembler `PronoteData`.
def fetch_step(fetcher: PronoteFetcher) -> Tuple[List[Lesson], List[Homework], List[SchoolEvent], List[Message]]: Quand l'agenda et les devoirs utilisent iCal pendant la même exécution, le téléchargement et le
""" parsing sont partagés dans un contexte local au run. Une simple valeur mémorisée dans l'instance du
Étape de récupération des données Pronote. fetcher ou dans le contexte d'exécution suffit ; aucun cache global, persistant ou système
d'invalidation n'est requis.
**Logique de precedence** :
- Si `agenda_source=ical` et `homework_source=ical`, un seul fetch iCal suffit (les devoirs sont extraits du même flux).
- Si `agenda_source=ical` et `homework_source=pronotepy`, deux sources distinctes sont utilisées.
- La déduplication globale est effectuée après fusion des résultats.
Args:
fetcher: Fetcher Pronote configuré.
Returns:
Tuple (lessons, homeworks, school_events, messages).
Raises:
PipelineCriticalError: Si aucune source n'est disponible.
PipelineError: Si une source échoue mais qu'une autre est disponible.
"""
try:
# Récupérer l'agenda (cours + événements scolaires)
lessons, agenda_homeworks = fetcher.fetch_agenda()
school_events = [] # À récupérer depuis iCal ou autre source
# Récupérer les devoirs selon la source configurée
# Si la source est iCal et que l'agenda a déjà été récupéré depuis iCal,
# les devoirs sont déjà inclus dans agenda_homeworks (via parsing iCal).
# Sinon, récupérer les devoirs depuis la source dédiée (ex: pronotepy).
if fetcher.homework_source.value == "pronotepy" or (
fetcher.homework_source.value == "auto" and fetcher.agenda_source.value != "ical"
):
# Récupérer les devoirs depuis pronotepy
homework_list = fetcher.fetch_homework()
# Fusionner les devoirs (agenda_homeworks peut être vide si agenda_source != ical)
homeworks = agenda_homeworks + homework_list
else:
# Utiliser les devoirs déjà extraits de l'agenda iCal
homeworks = agenda_homeworks
# Récupérer les messages et informations (toujours via pronotepy)
messages = fetcher.fetch_messages()
informations = fetcher.fetch_informations()
messages.extend(informations)
if not lessons and not homeworks:
raise PipelineCriticalError(
message="Aucun cours ou devoir récupéré depuis Pronote",
step="fetch",
)
return lessons, homeworks, school_events, messages
except Exception as e:
raise PipelineError(
message=f"Échec de la récupération Pronote: {redact_secrets(str(e))}",
severity=ErrorSeverity.ERROR,
step="fetch",
recoverable=False,
) from e
```
Une liste vide est une donnée valide et ne doit pas provoquer d'erreur critique. La criticité dépend
des exceptions remontées par les sources. L'étape importe les erreurs depuis
`pronote_sync.errors`, journalise uniquement des contenus expurgés et ne chaîne jamais une
exception externe brute susceptible de contenir un secret.
#### 11.4.1 bis `fetch_blog_step.py` #### 11.4.1 bis `fetch_blog_step.py`
@@ -4953,9 +4729,9 @@ Les autres étapes (`normalize_step`, `compare_step`, etc.) suivent le même pri
### 11.5 Points clés ### 11.5 Points clés
- **Ne jamais bloquer** : Les erreurs non critiques (ex: synthèse IA) ne bloquent pas le pipeline. - **Ne jamais bloquer** : Les erreurs non critiques (ex: synthèse IA) ne bloquent pas le pipeline.
- **Modes dégradés** : - **Modes dégradés** :
- Si iCal échoue → basculer sur `pronotepy`. - En mode `auto`, si iCal échoue → basculer sur `pronotepy`.
- Si `pronotepy` échoue → basculer sur iCal. - En mode explicite, ne pas changer de source.
- Si les deux échouent → **échec critique**. - En mode `auto`, si les deux sources échouent → **échec critique**.
- **Logs clairs** : Chaque erreur est loggée avec son niveau de gravité. - **Logs clairs** : Chaque erreur est loggée avec son niveau de gravité.
- **Retour d'erreur** : Le pipeline retourne toujours une liste des erreurs/warnings rencontrés. - **Retour d'erreur** : Le pipeline retourne toujours une liste des erreurs/warnings rencontrés.
@@ -5276,10 +5052,11 @@ def sample_settings():
return Settings( return Settings(
pronote=PronoteSettings( pronote=PronoteSettings(
ical_url="https://test.ent/pronote/ical/test.ics", url="https://test.ent/pronote/parent.html",
ical_url=SecretStr("https://test.ent/pronote/ical/test.ics"),
username="test_user", username="test_user",
password=SecretStr("test_password"), password=SecretStr("test_password"),
ent="test_ent", ent="monbureaunumerique",
agenda_source="auto", agenda_source="auto",
homework_source="auto", homework_source="auto",
messages_source="pronotepy", messages_source="pronotepy",
@@ -5331,7 +5108,12 @@ def test_parse_ical_lesson(parsed_lessons):
@pytest.mark.unittest @pytest.mark.unittest
def test_parse_ical_homework(parsed_lessons): def test_parse_ical_homework(parsed_lessons):
"""Test le parsing des devoirs depuis iCal.""" """Test le parsing des devoirs depuis iCal."""
lessons, homeworks, school_events = parsed_lessons from pronote_sync.sources.pronote.ical import collect_homeworks
lessons, parsed_homeworks, school_events = parsed_lessons
assert parsed_homeworks == []
homeworks = collect_homeworks(lessons, date(2026, 9, 10))
assert len(homeworks) == 1 assert len(homeworks) == 1
homework = homeworks[0] homework = homeworks[0]
@@ -5347,20 +5129,15 @@ def test_parse_ical_homework(parsed_lessons):
def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, mock_xmpp_channel, sample_settings): def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, mock_xmpp_channel, sample_settings):
"""Test le pipeline complet avec des mocks.""" """Test le pipeline complet avec des mocks."""
from pronote_sync.pipeline.run import PipelineRunner from pronote_sync.pipeline.run import PipelineRunner
from pronote_sync.sources.pronote.fetcher import PronoteFetcher from pronote_sync.sources.pronote.client import PronoteClient
from pronote_sync.sources.pronote.fallback import PronoteFetcher
from pronote_sync.sync.caldav import CalDAVClient from pronote_sync.sync.caldav import CalDAVClient
from pronote_sync.sync.diff import AgendaComparator from pronote_sync.sync.diff import AgendaComparator
from pronote_sync.sources.theoretical.file import CSVTheoreticalAgendaProvider from pronote_sync.sources.theoretical.file import CSVTheoreticalAgendaProvider
# Configurer le fetcher Pronote # Configurer le fetcher Pronote
fetcher = PronoteFetcher( pronote_client = PronoteClient(sample_settings.pronote)
ical_url=sample_settings.pronote.ical_url, fetcher = PronoteFetcher(sample_settings, pronote_client)
username=sample_settings.pronote.username,
password=sample_settings.pronote.password.get_secret_value(),
ent=sample_settings.pronote.ent,
agenda_source=sample_settings.pronote.agenda_source,
homework_source=sample_settings.pronote.homework_source,
)
# Configurer le client CalDAV # Configurer le client CalDAV
caldav_client = CalDAVClient( caldav_client = CalDAVClient(
@@ -5421,7 +5198,7 @@ def test_collect_homeworks():
"""Test la collecte et déduplication des devoirs depuis des blocs de plusieurs VEVENT.""" """Test la collecte et déduplication des devoirs depuis des blocs de plusieurs VEVENT."""
from datetime import date, datetime from datetime import date, datetime
from pronote_sync.models.agenda import Lesson, LessonStatus from pronote_sync.models.agenda import Lesson, LessonStatus
from pronote_sync.models.homework import HomeworkBlock from pronote_sync.models.agenda import HomeworkBlock
from pronote_sync.sources.pronote.ical import collect_homeworks from pronote_sync.sources.pronote.ical import collect_homeworks
# Créer des cours avec des blocs de devoirs (simulant des VEVENT parsés) # Créer des cours avec des blocs de devoirs (simulant des VEVENT parsés)
@@ -6174,4 +5951,3 @@ Ce guide fournit une **base architecturale et technique solide** pour développe
| `pydantic` | [https://pydantic.dev/](https://pydantic.dev/) | Bibliothèque pour la validation des données. | | `pydantic` | [https://pydantic.dev/](https://pydantic.dev/) | Bibliothèque pour la validation des données. |
| RFC 5545 (iCal) | [https://datatracker.ietf.org/doc/html/rfc5545](https://datatracker.ietf.org/doc/html/rfc5545) | Spécification officielle du format iCal. | | RFC 5545 (iCal) | [https://datatracker.ietf.org/doc/html/rfc5545](https://datatracker.ietf.org/doc/html/rfc5545) | Spécification officielle du format iCal. |
| RFC 4791 (CalDAV) | [https://datatracker.ietf.org/doc/html/rfc4791](https://datatracker.ietf.org/doc/html/rfc4791) | Spécification officielle de CalDAV. | RFC 4791 (CalDAV) | [https://datatracker.ietf.org/doc/html/rfc4791](https://datatracker.ietf.org/doc/html/rfc4791) | Spécification officielle de CalDAV.

37
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_pronote_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.
--- ---
@@ -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"

View File

@@ -3,16 +3,20 @@
Ce module fournit l'encapsulation du client ``pronotepy`` pour la source Ce module fournit l'encapsulation du client ``pronotepy`` pour la source
Pronote : récupération des messages des professeurs, des informations et Pronote : récupération des messages des professeurs, des informations et
sondages, ainsi que des cours et devoirs en mode repli lorsque le flux sondages, ainsi que des cours et devoirs en mode repli lorsque le flux
iCal échoue. Toutes les erreurs sont journalisées avec des secrets masqués. iCal échoue. Les erreurs des méthodes dégradées (messages, informations)
sont journalisées avec des secrets masqués ; les erreurs de récupération
des cours et des devoirs se propagent pour déclencher le repli iCal.
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
from datetime import date from datetime import date
from typing import Protocol from typing import Any, Protocol
import pronotepy import pronotepy
import pronotepy.ent as pronotepy_ent
import requests
from pronote_sync.config.settings import PronoteSettings from pronote_sync.config.settings import PronoteSettings
from pronote_sync.models.agenda import Lesson, LessonStatus from pronote_sync.models.agenda import Lesson, LessonStatus
@@ -23,6 +27,69 @@ from pronote_sync.utils.redaction import redact_exception
logger = logging.getLogger(__name__) 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): class PronoteClientProtocol(Protocol):
"""Interface du client Pronote consommée par la logique de repli.""" """Interface du client Pronote consommée par la logique de repli."""
@@ -42,13 +109,23 @@ class PronoteClientProtocol(Protocol):
""" """
... ...
def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]: def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours et les devoirs via ``pronotepy`` (repli iCal). """Récupère les cours via ``pronotepy`` (repli iCal).
:param start: Date de début de la fenêtre (incluse). :param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse).
:return: Tuple ``(cours, devoirs)``. :return: Liste des cours.
:rtype: tuple[list[Lesson], list[Homework]] :rtype: list[Lesson]
"""
...
def get_homeworks(self, start: date, end: date) -> list[Homework]:
"""Récupère les devoirs via ``pronotepy``.
:param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse).
:return: Liste des devoirs.
:rtype: list[Homework]
""" """
... ...
@@ -56,11 +133,13 @@ class PronoteClientProtocol(Protocol):
class PronoteClient: class PronoteClient:
"""Client d'accès à Pronote via ``pronotepy``. """Client d'accès à Pronote via ``pronotepy``.
Encapsule ``pronotepy.Client`` avec une connexion paresseuse : la Encapsule ``pronotepy.Client`` ou ``pronotepy.ParentClient`` selon le
connexion n'est établie qu'à la première méthode de récupération type de compte, avec une connexion paresseuse : la connexion n'est
appelée. Les erreurs ``pronotepy.PronoteAPIError`` sont journalisées établie qu'à la première méthode de récupération appelée. Les erreurs
avec des secrets masqués et les méthodes de récupération retournent des méthodes dégradées (``get_messages()``, ``get_informations()``)
alors une valeur vide au lieu de propager l'exception. 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: def __init__(self, settings: PronoteSettings) -> None:
@@ -75,24 +154,39 @@ class PronoteClient:
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse). """Crée et connecte le client ``pronotepy`` (connexion paresseuse).
Le client est créé une seule fois puis réutilisé pour les appels Le client est créé une seule fois puis réutilisé pour les appels
suivants. L'erreur de connexion est relancée sans journalisation, suivants. Le nom d'ENT est résolu via :func:`_resolve_ent` et le
la méthode publique appelante étant responsable de la journaliser. type de compte (``student`` ou ``parent``) détermine la classe de
client utilisée. L'erreur de connexion est relancée sans
journalisation, la méthode publique appelante étant responsable
de la journaliser.
:return: Le client ``pronotepy`` connecté. :return: Le client ``pronotepy`` connecté.
:rtype: pronotepy.Client :rtype: pronotepy.Client
:raises ValueError: Si ``username``, ``password`` ou ``ent`` est manquant. :raises ValueError: Si ``pronote_url``, ``username``, ``password``
ou ``ent`` est manquant, ou si l'ENT est inconnu.
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue. :raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
""" """
if self._client is None: if self._client is None:
pronote_url = self._settings.pronote_url
username = self._settings.username username = self._settings.username
password = self._settings.password password = self._settings.password
ent = self._settings.ent ent = self._settings.ent
if username is None or password is None or ent is None: if pronote_url is None or username is None or password is None or ent is None:
raise ValueError("username, password et ent sont requis pour pronotepy") raise ValueError(
try: "pronote_url, username, password et ent sont requis pour pronotepy"
self._client = pronotepy.Client(username, password.get_secret_value(), ent) )
except pronotepy.PronoteAPIError: resolver = _resolve_ent(ent)
raise 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 return self._client
def get_messages(self) -> list[Message]: def get_messages(self) -> list[Message]:
@@ -122,7 +216,13 @@ class PronoteClient:
) )
) )
return messages return messages
except (pronotepy.PronoteAPIError, ValueError) as exc: except (
pronotepy.PronoteAPIError,
ValueError,
requests.RequestException,
ConnectionError,
TimeoutError,
) as exc:
logger.error( logger.error(
"Échec de la récupération des messages Pronote : %s", "Échec de la récupération des messages Pronote : %s",
redact_exception(exc), redact_exception(exc),
@@ -154,62 +254,86 @@ class PronoteClient:
) )
) )
return messages return messages
except (pronotepy.PronoteAPIError, ValueError) as exc: except (
pronotepy.PronoteAPIError,
ValueError,
requests.RequestException,
ConnectionError,
TimeoutError,
) as exc:
logger.error( logger.error(
"Échec de la récupération des informations Pronote : %s", "Échec de la récupération des informations Pronote : %s",
redact_exception(exc), redact_exception(exc),
) )
return [] return []
def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]: def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours et les devoirs via ``pronotepy``. """Récupère les cours via ``pronotepy`` (repli iCal).
À utiliser uniquement si les sources iCal sont indisponibles ou en Les exceptions ne sont pas attrapées : elles se propagent afin que
repli automatique. Les cours annulés sont mappés sur le statut l'appelant puisse détecter l'échec et déclencher le repli (ou une
``CANCELLED`` ; **pronotepy** ne fournissant ni la date de 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 distribution ni les professeurs des devoirs, ces champs restent
vides. vides.
:param start: Date de début de la fenêtre (incluse). :param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse).
:return: Tuple ``(cours, devoirs)`` ; vide en cas d'erreur. :return: Liste des devoirs.
:rtype: tuple[list[Lesson], list[Homework]] :rtype: list[Homework]
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
:raises ValueError: Si la configuration ou l'ENT est invalide.
:raises requests.RequestException: Si une requête réseau échoue.
:raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire.
""" """
try: client = self._connect()
client = self._connect() homeworks: list[Homework] = []
lessons: list[Lesson] = [] for hw in client.homework(start, end):
for lesson in client.lessons(start, end): homeworks.append(
content = lesson.content Homework(
lessons.append( id=hw.id,
Lesson( subject=hw.subject.name,
id=lesson.id, teachers=(),
start=lesson.start, assigned_on=None,
end=lesson.end, due_on=hw.date,
subject=lesson.subject.name if lesson.subject is not None else "", text=hw.description,
teachers=tuple(lesson.teacher_names or ()), html=hw.description,
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,
)
) )
homeworks: list[Homework] = []
for hw in client.homework(start, end):
homeworks.append(
Homework(
id=hw.id,
subject=hw.subject.name,
teachers=(),
assigned_on=None,
due_on=hw.date,
text=hw.description,
html=hw.description,
)
)
return lessons, homeworks
except (pronotepy.PronoteAPIError, ValueError) as exc:
logger.error(
"Échec de la récupération de l'agenda via pronotepy : %s",
redact_exception(exc),
) )
return [], [] return homeworks

View File

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

View File

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

View File

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

View File

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

View File

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