Compare commits

...

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

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

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

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

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

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

View File

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

4
.gitignore vendored
View File

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

View File

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

View File

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

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).
- **Mode dégradé** :
- 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.
- Si iCal et `pronotepy` échouent → **échec explicite** avec message clair.
- En mode source `auto`, si iCal échoue → basculer sur `pronotepy` pour
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 |
|------------------------------|-----------------------------------------------------------------------------|---------------------------------------------|---------------|
| `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_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_USERNAME` | Identifiant CalDAV. | `user@example.com` | `str` |
| `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` |
| `XMPP_JID` | Identifiant XMPP (ex: `user@example.com`). | `user@example.com` | `str` |
| `XMPP_PASSWORD` | Mot de passe XMPP. | `SecretStr` (masqué) | `SecretStr` |
| `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`.
> 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
| Variable | Description | Valeur par défaut | Type |
@@ -305,6 +320,7 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
```ini
# --- 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_USERNAME=parent.dupont
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"`).
```python
from typing import Literal, Optional
from typing import Literal
from pydantic import SecretStr, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class PronoteSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="PRONOTE_", env_file=".env", extra="ignore")
ical_url: Optional[str] = None
username: Optional[str] = None
password: Optional[SecretStr] = None
ent: Optional[str] = None
url: str | None = None
ical_url: SecretStr | None = None
username: str | None = None
password: SecretStr | None = None
ent: str | None = None
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
messages_source: Literal["pronotepy"] = "pronotepy"
@@ -402,11 +419,11 @@ class AppSettings(BaseSettings):
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
pronote: PronoteSettings = PronoteSettings()
caldav: CalDAVSettings = CalDAVSettings()
xmpp: XmppSettings = XmppSettings()
ai: AISettings = AISettings()
app: AppSettings = AppSettings()
pronote: PronoteSettings = Field(default_factory=PronoteSettings)
caldav: CalDAVSettings = Field(default_factory=CalDAVSettings)
xmpp: XmppSettings = Field(default_factory=XmppSettings)
ai: AISettings = Field(default_factory=AISettings)
app: AppSettings = Field(default_factory=AppSettings)
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.
**Tests négatifs recommandés** :
- Vérifier que les messages d'erreur ne contiennent ni `icalsecurise`, ni `SECRET`, ni mots de passe, ni clés API.
- 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()
```
**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()
```
**Tests négatifs obligatoires** : injecter des sentinelles distinctes dans l'URL, les identifiants,
le mot de passe et la clé API, provoquer une erreur externe, puis vérifier leur absence dans le
message, les logs, `__cause__`, `__context__` et le traceback formaté.
**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.
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`)
> ⚠️ **Décision d'implémentation** :
@@ -1349,6 +1339,40 @@ def _format_message(self, message: XmppMessage) -> str:
### 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
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))
raise requests.exceptions.RequestException(
f"Échec de la récupération de {safe_url}: {safe_error}"
) from e
) from None
# Validation du flux
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
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
elif "Cours - Cours déplacé" in categories:
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 :
- Les **messages** des professeurs (`client.get_discussions()`).
- Les **informations et sondages** (`client.get_information_and_surveys()`).
`pronotepy` fournit les cours et devoirs de repli, ainsi que les messages, informations et
sondages. La signature réelle de la bibliothèque doit être respectée ; l'URL Pronote est le
premier argument et l'ENT est une fonction, pas une chaîne :
```python
from typing import List, Optional
from pronotepy import Client, PronoteAPIError
from ..models.message import Message, MessageType
from ..redaction import redact_url
import logging
import pronotepy.ent as pronote_ent
from pronotepy import ParentClient
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 pour interagir avec Pronote via pronotepy."""
client = ParentClient(
settings.url,
settings.username,
settings.password.get_secret_value(),
ent=ent_function,
)
```
def __init__(
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
Le client applicatif expose des méthodes distinctes :
def _get_client(self) -> Client:
"""Initialise et retourne le client pronotepy."""
if self._client is None:
if not all([self.username, self.password, self.ent]):
raise ValueError("Username, password et ENT sont requis pour pronotepy")
- `get_lessons(start, end) -> list[Lesson]` ;
- `get_homeworks(start, end) -> list[Homework]` ;
- `get_messages() -> list[Message]` ;
- `get_informations() -> list[Message]`.
self._client = Client(
self.username,
self.password,
self.ent,
)
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
Cette séparation évite qu'un appel agenda récupère inutilement les devoirs, et inversement. Les
méthodes agenda/devoirs ne transforment jamais une erreur en liste vide : elles journalisent une
version expurgée puis lèvent une erreur expurgée avec `from None`. Les méthodes de messages et
d'informations sont non critiques et peuvent retourner une liste vide avec un warning.
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`)
```python
from typing import Literal, Optional
from enum import Enum
from .ical import fetch_ical, parse_ical
from .client import PronoteClient
from ..models.agenda import Lesson, Homework
Le `PronoteFetcher` dépend de `Settings` et d'un protocole de client injecté ; il ne construit pas
de singleton et ne contient pas d'identifiants dupliqués.
| 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):
AUTO = "auto"
ICAL = "ical"
PRONOTEPY = "pronotepy"
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
```
Une réponse vide est un succès et ne déclenche pas de repli : une journée peut réellement ne
contenir aucun cours ou devoir. Inversement, une exception ne doit jamais être convertie en
`([], [])`, car le `PronoteFetcher` perdrait alors l'information nécessaire pour distinguer un
échec d'un résultat vide.
`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
@@ -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).
- **Modes dégradés** :
- **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).
- **iCal** : Si la récupération échoue → basculer sur **pronotepy** (si configuré).
- **Sources Pronote en mode `auto`** : essayer iCal, puis basculer sur `pronotepy`
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.
- **XMPP** : Si l'envoi échoue → **logger l'erreur** mais continuer le pipeline.
- **Erreurs critiques** :
@@ -4556,6 +4381,9 @@ def get_channel(settings: Settings, channel_name: str = "xmpp") -> Channel:
### 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
from enum import Enum, auto
from typing import Optional
@@ -4610,7 +4438,7 @@ from ..models.xmpp import XmppMessage
from ..models.pronote import PronoteData
from ..models.sync import CalDAVSyncResult
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.diff import AgendaComparator
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`.
#### 11.4.1 `fetch_step.py`
#### 11.4.1 `fetch.py`
```python
from typing import Tuple, List
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
L'étape de récupération orchestre le contrat de `PronoteFetcher` sans réimplémenter la sélection
des sources :
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]]:
"""
Étape de récupération des données Pronote.
**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
```
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
fetcher ou dans le contexte d'exécution suffit ; aucun cache global, persistant ou système
d'invalidation n'est requis.
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`
@@ -4953,9 +4729,9 @@ Les autres étapes (`normalize_step`, `compare_step`, etc.) suivent le même pri
### 11.5 Points clés
- **Ne jamais bloquer** : Les erreurs non critiques (ex: synthèse IA) ne bloquent pas le pipeline.
- **Modes dégradés** :
- Si iCal échoue → basculer sur `pronotepy`.
- Si `pronotepy` échoue → basculer sur iCal.
- Si les deux échouent → **échec critique**.
- En mode `auto`, si iCal échoue → basculer sur `pronotepy`.
- En mode explicite, ne pas changer de source.
- En mode `auto`, si les deux sources échouent → **échec critique**.
- **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.
@@ -5276,10 +5052,11 @@ def sample_settings():
return Settings(
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",
password=SecretStr("test_password"),
ent="test_ent",
ent="monbureaunumerique",
agenda_source="auto",
homework_source="auto",
messages_source="pronotepy",
@@ -5331,7 +5108,12 @@ def test_parse_ical_lesson(parsed_lessons):
@pytest.mark.unittest
def test_parse_ical_homework(parsed_lessons):
"""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
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):
"""Test le pipeline complet avec des mocks."""
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.diff import AgendaComparator
from pronote_sync.sources.theoretical.file import CSVTheoreticalAgendaProvider
# Configurer le fetcher Pronote
fetcher = PronoteFetcher(
ical_url=sample_settings.pronote.ical_url,
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,
)
pronote_client = PronoteClient(sample_settings.pronote)
fetcher = PronoteFetcher(sample_settings, pronote_client)
# Configurer le client CalDAV
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."""
from datetime import date, datetime
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
# 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. |
| 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.

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

View File

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

33
pronote_sync/errors.py Normal file
View File

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

View File

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

View File

@@ -0,0 +1,408 @@
"""Logique de repli unifiant les sources iCal et pronotepy.
Ce module fournit l'enum :class:`AgendaSource`, le protocole
:class:`PronoteFetcherProtocol` consommé par le pipeline ainsi que la
classe :class:`PronoteFetcher` qui sélectionne la source selon la
configuration (``PRONOTE_AGENDA_SOURCE`` / ``PRONOTE_HOMEWORK_SOURCE``).
Contrat strict : les modes explicites n'utilisent que la source
configurée, sans aucun repli ; seul le mode ``auto`` applique un repli
unique iCal → pronotepy, en cas d'exception uniquement. Les messages
et informations proviennent toujours de pronotepy. Toutes les erreurs
sont journalisées avec des secrets masqués via
:func:`~pronote_sync.utils.redaction.redact_exception` ; les exceptions
d'origine ne sont jamais chaînées (``from None``).
"""
from __future__ import annotations
import logging
from datetime import date, timedelta
from enum import StrEnum
from typing import Literal, Protocol
from pronote_sync.config.settings import Settings
from pronote_sync.errors import PipelineCriticalError
from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message
from pronote_sync.sources.pronote.client import PronoteClientProtocol
from pronote_sync.sources.pronote.ical import collect_homeworks, fetch_ical, parse_ical
from pronote_sync.utils.redaction import redact_exception
logger = logging.getLogger(__name__)
_SourceName = Literal["ical", "pronotepy"]
class AgendaSource(StrEnum):
"""Source configurée pour la récupération de l'agenda et des devoirs."""
AUTO = "auto"
ICAL = "ical"
PRONOTEPY = "pronotepy"
class PronoteFetcherProtocol(Protocol):
"""Interface de récupération des données Pronote consommée par le pipeline."""
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère les cours et les événements scolaires.
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises PipelineCriticalError: Si toutes les sources configurées échouent.
"""
...
def fetch_homework(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs pour la date cible.
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
:raises PipelineCriticalError: Si toutes les sources configurées échouent.
"""
...
def fetch_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote.
:return: Liste des messages.
:rtype: list[Message]
"""
...
def fetch_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote.
:return: Liste des informations et sondages.
:rtype: list[Message]
"""
...
class PronoteFetcher:
"""Récupère les données Pronote via iCal ou pronotepy, repli réservé au mode ``auto``.
Unifie les sources iCal et pronotepy selon la source configurée
(``agenda_source`` / ``homework_source``) : les modes explicites
n'utilisent que la source configurée, sans aucun repli ; seul le mode
``auto`` essaie une source primaire puis, si elle échoue, une seule
source de repli lorsqu'elle est configurée. Les messages et
informations proviennent toujours de pronotepy.
"""
def __init__(self, settings: Settings, pronote_client: PronoteClientProtocol) -> None:
"""Initialise le fetcher sans récupérer aucune donnée.
:param settings: Configuration racine du pipeline (Pronote, fenêtre de synchronisation).
:param pronote_client: Client pronotepy utilisé pour les sources pronotepy.
"""
self._settings: Settings = settings
self._pronote_client: PronoteClientProtocol = pronote_client
def _fetch_window(self) -> tuple[date, date]:
"""Calcule la fenêtre de synchronisation autour de la date du jour.
:return: Tuple ``(date de début, date de fin)`` de la fenêtre.
:rtype: tuple[date, date]
"""
today = date.today()
start = today - timedelta(days=self._settings.app.sync_past_days)
end = today + timedelta(days=self._settings.app.sync_future_days)
return start, end
def _is_ical_configured(self) -> bool:
"""Vérifie que la source iCal est configurée.
:return: ``True`` si ``ical_url`` est défini, ``False`` sinon.
:rtype: bool
"""
return self._settings.pronote.ical_url is not None
def _is_pronotepy_configured(self) -> bool:
"""Vérifie que la source pronotepy est entièrement configurée.
:return: ``True`` si ``pronote_url``, ``username``, ``password``
et ``ent`` sont tous définis, ``False`` sinon.
:rtype: bool
"""
pronote = self._settings.pronote
return (
pronote.pronote_url is not None
and pronote.username is not None
and pronote.password is not None
and pronote.ent is not None
)
def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère l'agenda depuis le flux iCal.
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide.
:raises OSError: Si le fichier iCal local est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue.
"""
ical_url = self._settings.pronote.ical_url
if ical_url is None:
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
raw_ical = fetch_ical(ical_url.get_secret_value())
lessons, _, school_events = parse_ical(raw_ical)
return lessons, school_events
def _fetch_agenda_pronotepy(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère l'agenda depuis pronotepy.
Les événements scolaires (vacances, jours fériés) ne sont pas
fournis par pronotepy : la liste retournée est vide.
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
:raises ValueError: Si la configuration ou l'ENT est invalide.
:raises requests.RequestException: Si une requête réseau échoue.
:raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire.
"""
start, end = self._fetch_window()
lessons = self._pronote_client.get_lessons(start, end)
return lessons, []
def _agenda_sources(self) -> tuple[_SourceName, _SourceName | None]:
"""Sélectionne la source primaire et le repli unique pour l'agenda.
Les modes explicites ``ICAL`` et ``PRONOTEPY`` désignent la seule
source utilisée, sans aucun repli. En mode ``AUTO``, iCal est
primaire si ``ical_url`` est configuré (repli pronotepy si la
configuration pronotepy est complète), sinon pronotepy sans repli.
:return: Tuple ``(source primaire, source de repli ou ``None``)``.
:rtype: tuple[_SourceName, _SourceName | None]
:raises PipelineCriticalError: Si aucune source n'est configurée en mode ``AUTO``.
"""
source = AgendaSource(self._settings.pronote.agenda_source)
if source is AgendaSource.ICAL:
return "ical", None
if source is AgendaSource.PRONOTEPY:
return "pronotepy", None
if self._is_ical_configured():
return "ical", "pronotepy" if self._is_pronotepy_configured() else None
if self._is_pronotepy_configured():
return "pronotepy", None
raise PipelineCriticalError(
"Impossible de récupérer l'agenda : ni la source iCal ni pronotepy n'est configurée"
) from None
def _fetch_agenda_source(self, name: _SourceName) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère l'agenda depuis la source nommée.
:param name: Nom de la source (``"ical"`` ou ``"pronotepy"``).
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
"""
if name == "ical":
return self._fetch_agenda_ical()
return self._fetch_agenda_pronotepy()
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère les cours et les événements scolaires selon la source configurée.
En mode explicite (``ical`` ou ``pronotepy``), la source désignée
est la seule tentée : si elle échoue, une erreur critique est levée
sans repli. En mode ``auto``, la source primaire est essayée en
premier puis, si elle échoue, la source de repli unique (l'autre
source, si configurée) l'est à son tour ; si la source primaire et
le repli échouent — ou si aucune source n'est configurée — une
erreur critique est levée.
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
"""
primary, fallback = self._agenda_sources()
try:
return self._fetch_agenda_source(primary)
except Exception as exc:
logger.error(
"Échec de la récupération %s pour l'agenda : %s",
primary,
redact_exception(exc),
)
if fallback is None:
raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : la source {primary} a échoué"
) from None
logger.info("Repli sur %s pour l'agenda.", fallback)
try:
lessons, school_events = self._fetch_agenda_source(fallback)
except Exception as exc:
logger.error(
"Échec de la récupération %s pour l'agenda : %s",
fallback,
redact_exception(exc),
)
raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : les sources {primary}"
f" et {fallback} ont échoué"
) from None
if not lessons:
logger.warning(
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
"de %s : impossible de distinguer une absence de cours d'un échec "
"silencieux.",
fallback,
primary,
)
return lessons, school_events
def _fetch_homework_ical(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis le flux iCal pour la date cible.
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
:raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide.
:raises OSError: Si le fichier iCal local est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue.
"""
lessons, _ = self._fetch_agenda_ical()
return collect_homeworks(lessons, target_date)
def _fetch_homework_pronotepy(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis pronotepy pour la date cible.
Les devoirs sont filtrés sur la date d'échéance : seuls ceux dont
``due_on`` correspond à ``target_date`` sont conservés.
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
:raises ValueError: Si la configuration ou l'ENT est invalide.
:raises requests.RequestException: Si une requête réseau échoue.
:raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire.
"""
start, end = self._fetch_window()
homeworks = self._pronote_client.get_homeworks(start, end)
return [hw for hw in homeworks if hw.due_on == target_date]
def _homework_sources(self) -> tuple[_SourceName, _SourceName | None]:
"""Sélectionne la source primaire et le repli unique pour les devoirs.
Les modes explicites ``ICAL`` et ``PRONOTEPY`` désignent la seule
source utilisée, sans aucun repli. En mode ``AUTO``, iCal est
primaire si ``ical_url`` est configuré (repli pronotepy si la
configuration pronotepy est complète), sinon pronotepy sans repli.
:return: Tuple ``(source primaire, source de repli ou ``None``)``.
:rtype: tuple[_SourceName, _SourceName | None]
:raises PipelineCriticalError: Si aucune source n'est configurée en mode ``AUTO``.
"""
source = AgendaSource(self._settings.pronote.homework_source)
if source is AgendaSource.ICAL:
return "ical", None
if source is AgendaSource.PRONOTEPY:
return "pronotepy", None
if self._is_ical_configured():
return "ical", "pronotepy" if self._is_pronotepy_configured() else None
if self._is_pronotepy_configured():
return "pronotepy", None
raise PipelineCriticalError(
"Impossible de récupérer les devoirs : ni la source iCal ni pronotepy n'est configurée"
) from None
def _fetch_homework_source(self, name: _SourceName, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis la source nommée.
:param name: Nom de la source (``"ical"`` ou ``"pronotepy"``).
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
"""
if name == "ical":
return self._fetch_homework_ical(target_date)
return self._fetch_homework_pronotepy(target_date)
def fetch_homework(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs selon la source configurée.
En mode explicite (``ical`` ou ``pronotepy``), la source désignée
est la seule tentée : si elle échoue, une erreur critique est levée
sans repli. En mode ``auto``, la source primaire est essayée en
premier puis, si elle échoue, la source de repli unique (l'autre
source, si configurée) l'est à son tour ; si la source primaire et
le repli échouent — ou si aucune source n'est configurée — une
erreur critique est levée.
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
"""
primary, fallback = self._homework_sources()
try:
return self._fetch_homework_source(primary, target_date)
except Exception as exc:
logger.error(
"Échec de la récupération %s pour les devoirs : %s",
primary,
redact_exception(exc),
)
if fallback is None:
raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : la source {primary} a échoué"
) from None
logger.info("Repli sur %s pour les devoirs.", fallback)
try:
homeworks = self._fetch_homework_source(fallback, target_date)
except Exception as exc:
logger.error(
"Échec de la récupération %s pour les devoirs : %s",
fallback,
redact_exception(exc),
)
raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : les sources {primary}"
f" et {fallback} ont échoué"
) from None
if not homeworks:
logger.warning(
"Le repli %s pour les devoirs a retourné un résultat vide après "
"l'échec de %s : impossible de distinguer une absence de devoirs "
"d'un échec silencieux.",
fallback,
primary,
)
return homeworks
def fetch_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote (toujours via pronotepy).
:return: Liste des messages.
:rtype: list[Message]
"""
try:
return self._pronote_client.get_messages()
except Exception as exc:
logger.error(
"Échec de la récupération des messages : %s",
redact_exception(exc),
)
raise
def fetch_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote (toujours via pronotepy).
:return: Liste des informations et sondages.
:rtype: list[Message]
"""
try:
return self._pronote_client.get_informations()
except Exception as exc:
logger.error(
"Échec de la récupération des informations : %s",
redact_exception(exc),
)
raise

View File

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

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

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

1189
tests/unit/test_fallback.py Normal file

File diff suppressed because it is too large Load Diff

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

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

View File

@@ -0,0 +1,439 @@
"""Tests unitaires pour le client Pronote via pronotepy.
Ce module vérifie le comportement du client ``PronoteClient`` et de son
protocole ``PronoteClientProtocol``. Tous les tests sont unitaires et
utilisent des mocks pour éviter tout accès réseau réel à Pronote.
"""
from __future__ import annotations
from datetime import date, datetime
import pronotepy
import pytest
import pytest_mock
from pydantic import SecretStr
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.models.agenda import Lesson, LessonStatus
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
from pronote_sync.sources.pronote.client import PronoteClient, PronoteClientProtocol
# --- Protocol tests ---
def test_protocol_methods(mocker: pytest_mock.MockerFixture) -> None:
"""Vérifie que le protocole PronoteClientProtocol expose les méthodes attendues.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
assert hasattr(PronoteClientProtocol, "get_messages")
assert hasattr(PronoteClientProtocol, "get_informations")
assert hasattr(PronoteClientProtocol, "get_lessons")
assert hasattr(PronoteClientProtocol, "get_homeworks")
# --- Client with mocked pronotepy ---
@pytest.fixture
def pronote_settings() -> PronoteSettings:
"""Fournit des paramètres Pronote valides pour les tests.
:return: Instance de PronoteSettings avec des valeurs par défaut.
:rtype: PronoteSettings
"""
return PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="parent",
)
@pytest.fixture
def empty_pronote_settings() -> PronoteSettings:
"""Fournit des paramètres Pronote vides pour les tests.
:return: Instance de PronoteSettings avec tous les champs à None.
:rtype: PronoteSettings
"""
return PronoteSettings(username=None, password=None, ent=None)
def test_get_messages_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_messages retourne une liste de Message en cas de succès.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
# Mock du client pronotepy
mock_client = mocker.MagicMock()
mock_discussion = mocker.MagicMock()
mock_discussion.subject = "Test Subject"
mock_message = mocker.MagicMock()
mock_message.id = "msg-123"
mock_message.content = "Test message content"
mock_message.author = "Teacher Test"
mock_message.created = datetime(2024, 9, 1, 10, 0, 0)
mock_message.seen = True
mock_discussion.messages = [mock_message]
mock_client.discussions.return_value = [mock_discussion]
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_messages()
assert isinstance(messages, list)
assert len(messages) == 1
message = messages[0]
assert isinstance(message, Message)
assert message.id == "msg-123"
assert message.type == MessageType.DISCUSSION
assert message.title == "Test Subject"
assert message.content == "Test message content"
assert message.author == "Teacher Test"
assert message.date == datetime(2024, 9, 1, 10, 0, 0)
assert message.read is True
def test_get_messages_empty_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_messages retourne une liste vide en cas d'erreur API.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.discussions.side_effect = pronotepy.PronoteAPIError("API error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_messages()
assert messages == []
def test_get_informations_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_informations retourne une liste de Message en cas de succès.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_info = mocker.MagicMock()
mock_info.id = "info-456"
mock_info.title = "Important Info"
mock_info.content.return_value = "Important content"
mock_info.author = "Admin"
mock_info.creation_date = datetime(2024, 9, 2, 14, 30, 0)
mock_info.read = False
mock_info.survey = True
mock_client.information_and_surveys.return_value = [mock_info]
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
assert isinstance(messages, list)
assert len(messages) == 1
message = messages[0]
assert isinstance(message, Message)
assert message.id == "info-456"
assert message.type == MessageType.SURVEY
assert message.title == "Important Info"
assert message.content == "Important content"
assert message.author == "Admin"
assert message.date == datetime(2024, 9, 2, 14, 30, 0)
assert message.read is False
def test_get_informations_empty_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_informations retourne une liste vide en cas d'erreur API.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.information_and_surveys.side_effect = pronotepy.PronoteAPIError("API error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
assert messages == []
def test_get_lessons_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_lessons retourne une liste de Lesson en cas de succès.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
# Mock des cours
mock_lesson = mocker.MagicMock()
mock_lesson.id = "lesson-789"
mock_lesson.start = datetime(2024, 9, 1, 8, 0, 0)
mock_lesson.end = datetime(2024, 9, 1, 9, 30, 0)
mock_lesson.subject = mocker.MagicMock()
mock_lesson.subject.name = "Maths"
mock_lesson.teacher_names = ["Prof A", "Prof B"]
mock_lesson.classrooms = ["Salle 101", "Salle 102"]
mock_lesson.group_name = "Classe 1"
mock_lesson.canceled = False
mock_content = mocker.MagicMock()
mock_content.description = "Lesson content"
mock_lesson.content = mock_content
mock_client.lessons.return_value = [mock_lesson]
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
lessons = client.get_lessons(date(2024, 9, 1), date(2024, 9, 30))
assert isinstance(lessons, list)
assert len(lessons) == 1
lesson = lessons[0]
assert isinstance(lesson, Lesson)
assert lesson.id == "lesson-789"
assert lesson.start == datetime(2024, 9, 1, 8, 0, 0)
assert lesson.end == datetime(2024, 9, 1, 9, 30, 0)
assert lesson.subject == "Maths"
assert lesson.teachers == ("Prof A", "Prof B")
assert lesson.rooms == ("Salle 101", "Salle 102")
assert lesson.group == "Classe 1"
assert lesson.status == LessonStatus.NORMAL
assert lesson.content == "Lesson content"
def test_get_homeworks_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_homeworks retourne une liste de Homework en cas de succès.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
# Mock des devoirs
mock_hw = mocker.MagicMock()
mock_hw.id = "hw-101"
mock_hw.subject = mocker.MagicMock()
mock_hw.subject.name = "Maths"
mock_hw.date = date(2024, 9, 15)
mock_hw.description = "Do your homework"
mock_client.homework.return_value = [mock_hw]
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
homeworks = client.get_homeworks(date(2024, 9, 1), date(2024, 9, 30))
assert isinstance(homeworks, list)
assert len(homeworks) == 1
homework = homeworks[0]
assert isinstance(homework, Homework)
assert homework.id == "hw-101"
assert homework.subject == "Maths"
assert homework.teachers == ()
assert homework.assigned_on is None
assert homework.due_on == date(2024, 9, 15)
assert homework.text == "Do your homework"
assert homework.html == "Do your homework"
def test_get_lessons_propagates_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_lessons propage les exceptions API.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.lessons.side_effect = pronotepy.PronoteAPIError("API error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
with pytest.raises(pronotepy.PronoteAPIError):
client.get_lessons(date(2024, 9, 1), date(2024, 9, 30))
def test_get_homeworks_propagates_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_homeworks propage les exceptions API.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.homework.side_effect = pronotepy.PronoteAPIError("API error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
with pytest.raises(pronotepy.PronoteAPIError):
client.get_homeworks(date(2024, 9, 1), date(2024, 9, 30))
def test_missing_credentials_raises(empty_pronote_settings: PronoteSettings) -> None:
"""Vérifie que les appels échouent avec ValueError si les identifiants sont manquants.
:param empty_pronote_settings: Paramètres Pronote avec tous les champs à None.
:return: None
"""
client = PronoteClient(empty_pronote_settings)
with pytest.raises(ValueError, match="pronote_url, username, password et ent sont requis"):
client._connect()
def test_connect_with_ent_resolution(mocker: pytest_mock.MockerFixture) -> None:
"""Vérifie que _resolve_ent retourne le callable attendu pour un ENT connu.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from pronote_sync.sources.pronote.client import _resolve_ent
resolver = _resolve_ent("bordeaux")
assert resolver is not None
def test_connect_with_unknown_ent_raises(mocker: pytest_mock.MockerFixture) -> None:
"""Vérifie que _resolve_ent lève ValueError pour un ENT inconnu.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from pronote_sync.sources.pronote.client import _resolve_ent
with pytest.raises(ValueError) as exc_info:
_resolve_ent("inconnu")
assert "ENT inconnu : 'inconnu'" in str(exc_info.value)
assert "ENT supportés :" in str(exc_info.value)
def test_connect_parent_account_type(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que account_type='parent' utilise pronotepy.ParentClient.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
from unittest.mock import Mock
from pronote_sync.sources.pronote.client import PronoteClient
mock_client = mocker.MagicMock()
mock_client_class = Mock(return_value=mock_client)
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
mocker.patch("pronotepy.Client")
client = PronoteClient(pronote_settings)
_ = client._connect()
# Verify ParentClient was used
assert mock_client_class.call_count == 1
pronotepy.Client.assert_not_called() # type: ignore[attr-defined]
def test_connect_student_account_type(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que account_type='student' utilise pronotepy.Client.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
from unittest.mock import Mock
from pronote_sync.sources.pronote.client import PronoteClient
pronote_settings_student = PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="student",
)
mock_client = mocker.MagicMock()
mock_client_class = Mock(return_value=mock_client)
mocker.patch("pronotepy.Client", new=mock_client_class)
mocker.patch("pronotepy.ParentClient")
client = PronoteClient(pronote_settings_student)
_ = client._connect()
# Verify Client was used
assert mock_client_class.call_count == 1
pronotepy.ParentClient.assert_not_called() # type: ignore[attr-defined]
def test_get_messages_degraded_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_messages retourne une liste vide en cas d'erreur réseau.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.discussions.side_effect = ConnectionError("Network error")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_messages()
assert messages == []
def test_get_informations_degraded_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Vérifie que get_informations retourne une liste vide en cas d'erreur réseau.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = mocker.MagicMock()
mock_client.information_and_surveys.side_effect = TimeoutError("Timeout")
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
assert messages == []
# Ensure trailing newline