Compare commits

..

13 Commits

Author SHA1 Message Date
d2cf59c713 docs: marquer le jalon M8 (comparaison agenda théorique) comme terminé
M8 livré : AgendaComparator dans sync/diff.py avec matching déterministe,
tolérance ±15 min, normalisation NFKC des matières, REMOVED par existence.
Le critère d'acceptation 3 (absence de THEORETICAL_AGENDA_PATH) est couvert
par design et reporté à M11 (composition root).

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-07 13:58:25 +02:00
093253a41c test(M8): tests unitaires pour AgendaComparator (17 cas)
Couvre : agendas vides, ADDED/REMOVED/MODIFIED, tolérance ±15 min
(bord inclusif), normalisation NFKC des matières, matching multi-candidats
par plus petit id, comparaison ordre-insensible des enseignants/salles,
statut != NORMAL, REMOVED par existence (pas par sélection), ordre
déterministe et idempotence. Couverture de sync/diff.py : 92%.

Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-07 13:50:56 +02:00
10e5f22501 feat(M8): comparateur d'agenda (AgendaComparator) dans sync/diff.py
Comparaison déterministe entre l'agenda réel (Lesson) et l'agenda
théorique (TheoreticalLesson) produisant un AgendaDiff (ADDED/REMOVED/
MODIFIED). Matching par jour + tolérance ±15 min symétrique + matière
normalisée ; tri des candidats par id stable. REMOVED par existence
(non-appariement), pas par sélection. Comparaison ordre-insensible des
enseignants et salles via set(). Détection MODIFIED incluant les horaires,
la matière, les enseignants, les salles et le statut.

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-07 13:39:22 +02:00
557555c65b refactor: déplacer normalize_subject vers utils/text.py avec ré-export
La normalisation des matières (NFKC + espaces + ponctuation + minuscules)
est désormais dans pronote_sync/utils/text.py pour permettre son partage
entre sources/theoretical/file.py et sync/diff.py (M8) sans couplage de
couche. L'import depuis file.py est préservé par ré-export explicite.

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-07 13:30:15 +02:00
c309bcbb64 docs: marquer le jalon M7 (synchronisation CalDAV) comme terminé dans TODO.md 2026-09-07 13:18:04 +02:00
88a75cd162 Ignore zvec-grep 2026-09-07 12:29:17 +02:00
91f0b6d9a0 merge: jalon M7 — synchronisation différentielle CalDAV 2026-09-07 12:26:06 +02:00
a1bae41be8 fix(M7): corrections d'audit FIXME_M7 — sécurité, fenêtre, UID, timezone
Corrige les 5 constats de l'audit FIXME_M7 :

#1 (Bloquant) — Protection des événements non marqués :
- upsert_event() vérifie le marqueur X-PRONOTE-SYNC-MANAGED avant
  modification ; lève PronoteSyncError en cas de collision avec un
  événement non géré (aucune écriture)
- delete_event() vérifie le marqueur ; no-op avec warning si non géré
- Méthode privée _is_managed_event() factorisant le contrôle

#2 (Bloquant) — Fenêtre de synchronisation :
- Calcul en journées entières (minuit à minuit exclusif)
- Filtrage des données locales (lessons, homeworks, school_events) avant
  passage au planner
- Paramètre now injectable pour les tests

#3 (Bloquant) — UID canonique vs brut :
- list_managed_events() retourne (raw_uid, canonical_uid, vevent)
- compute_plan() matche par UID canonique, route les raw UID vers
  *_to_remove, retourne le mapping remote_raw_by_canonical
- executor.execute() utilise le raw UID pour les mises à jour (pas de
  doublon)
- Pas de migration destructive des UID distants existants

#4 (Correction) — Normalisation temporelle UTC :
- normalize_datetime_to_utc() dans utils/uid.py : naïve → Europe/Paris →
  UTC ; consciente → UTC
- Utilisée par generate_deterministic_uid() et component_to_signature()
- Deux représentations du même instant → même UID et même signature

#5 (Compatibilité) — date_search déprécié :
- Remplacement par calendar.search(start, end, event=True, expand=True)

Documentation :
- GUIDE_DEV_PYTHON.md : suppression des références obsolètes à
  sync/state.py et état SQLite/JSON ; mise à jour de l'API CalDAV
  (search au lieu de date_search, upsert par UID)
- TODO.md : M7 décoché (corrections en cours de validation)

Tests : 390 passés, couverture 95.61%

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>
2026-09-07 12:24:22 +02:00
b4b0247919 feat(M7): synchronisation différentielle CalDAV
Implémente la synchronisation des événements Pronote vers un calendrier
CalDAV (Nextcloud) de façon idempotente et sécurisée.

Production :
- sync/serialization.py : sérialisation Lesson/Homework/SchoolEvent vers
  VEVENT, signature sémantique (exclut DTSTAMP/CREATED/LAST-MODIFIED),
  enveloppe VCALENDAR complète avec VERSION:2.0 et PRODID
- sync/caldav.py : passerelle CalDAV isolant caldav>=1.3.0, résolution du
  calendrier via principal().calendars() avec boundary matching, upsert par
  UID (fetch-then-save), exceptions expurgées et __context__ propre, mot de
  passe non stocké en clair, context manager
- sync/planner.py : calcul explicite du CalDAVSyncPlan (add/update/remove
  par comparaison de signatures sémantiques, routage par préfixe d'UID)
- sync/executor.py : exécution du plan avec dry-run (aucune écriture),
  isolation des erreurs par événement, statut FAILED/SKIPPED/SUCCESS
- sync/synchronizer.py : orchestration en trois phases (scan, plan,
  exécution), SKIPPED si CalDAV non configuré
- sync/__init__.py : export synchronize()
- sources/pronote/client.py : normalisation UID via normalize_pronote_uid/
  generate_deterministic_uid (parité avec ical.py)
- config/settings.py : CalDAVSettings durci (url SecretStr, validation
  HTTPS, allow_insecure_http pour localhost, serializer redact_url)

Tests (381 passés, couverture 95.58%) :
- tests/unit/test_sync_serialization.py (21 tests)
- tests/unit/test_caldav_planner.py (16 tests)
- tests/unit/test_caldav_executor.py (18 tests)
- tests/unit/test_caldav_gateway.py (24 tests)
- tests/unit/test_caldav_security.py (18 tests)
- tests/unit/test_uid_equivalence.py (8 tests)
- tests/integration/test_caldav_sync.py (11 tests, faux serveur en mémoire)
- tests/conftest.py : fixtures partagées

Documentation :
- GUIDE_DEV_PYTHON.md §7 : API réelle caldav>=1.3.0, principal().calendars(),
  VCALENDAR complet, upsert par UID, pas d'état local, événements non gérés
  protégés, CalDAVSettings durci (SecretStr, HTTPS, allow_insecure_http)
- TODO.md : M7 coché
- .env.example : CALDAV_ALLOW_INSECURE_HTTP=false

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>
2026-09-07 09:24:18 +02:00
ebbe39f1f0 docs(M7): aligner guide, TODO et pre-commit pour la sync CalDAV
Met à jour la documentation et la configuration pour le jalon M7 selon
les décisions d'architecture :

- GUIDE_DEV_PYTHON.md §7 : API réelle caldav>=1.3.0 (pas le pseudo-code),
  calendar_path (pas calendar_name), plan CalDAVSyncPlan explicite avant
  exécution, pas d'état local (scan distant), événements non marqués
  jamais modifiés
- TODO.md M7 : suppression de sync/state.py et BlogRSSState, ajout de
  l'exécution du plan et de la protection des événements non gérés
- .pre-commit-config.yaml : caldav>=1.3.0 ajouté aux additional_dependencies
  du hook mypy

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/tech-writer <tech-writer@agents.invalid>
2026-09-07 00:00:54 +02:00
958bb3ec5d merge: corrections d'audit FIXME_M6 dans l'agenda théorique
Intègre les corrections de la relecture indépendante :
- Normalisation des matières (NFKC + espaces + ponctuation + casse)
- Expurgation du secret dans l'erreur de collision d'IDs
- Tests renforcés (parité isolée, normalisation, non-fuite)
- Documentation alignée (TODO coché, guide §8.4 corrigé)

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 23:31:25 +02:00
1d26d49e74 fix(M6): corrections d'audit FIXME_M6 — normalisation, secret, tests, doc
Corrige les 5 points de l'audit FIXME_M6 :

1. Normalisation des matières : fonction normalize_subject (NFKC +
   unification des espaces + suppression ponctuation + minuscule)
   partagée par la génération d'ID et le futur comparateur M8.
2. Expurgation du secret dans l'erreur de collision d'IDs :
   redact_secrets enveloppe l'identifiant dans le message.
3. Test even/odd avec même matière pour isoler la parité comme seul
   différenciateur d'ID ; tests de normalisation (casse, espaces,
   Unicode) ; test de non-fuite de secret.
4. TODO.md M6 : 8 items cochés après validation.
5. GUIDE_DEV §8.4 : bloc de code corrigé (clôture, types Lesson/
   TheoreticalLesson, comparaison des horaires en minutes, début ET
   fin, référence à normalize_subject).

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 23:30:47 +02:00
29270427ef merge: jalon M6 — agenda théorique JSON avec parité et vacances scolaires
Intègre le jalon M6 complet :
- Source d'agenda théorique au format JSON (avec parité paire/impaire)
- Service de parité des semaines (WeekParityService) basé sur une date
  de référence configurée
- Calendrier de vacances scolaires (SchoolHolidayCalendar, zone A)
- Provider JSON avec filtrage par parité et vacances
- Factory de câblage de configuration
- Fixtures theoretical.json et school_holidays.json
- 57 tests unitaires
- Documentation alignée (TODO.md, GUIDE_DEV_PYTHON.md, .env.example)

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 23:06:47 +02:00
29 changed files with 6810 additions and 481 deletions

View File

@@ -16,6 +16,8 @@ CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
CALDAV_USERNAME=user@example.com CALDAV_USERNAME=user@example.com
CALDAV_PASSWORD=your_caldav_password CALDAV_PASSWORD=your_caldav_password
CALDAV_CALENDAR_PATH=/pronote-sync/ CALDAV_CALENDAR_PATH=/pronote-sync/
# Autoriser HTTP (non-HTTPS) pour un serveur CalDAV local (localhost uniquement)
CALDAV_ALLOW_INSECURE_HTTP=false
# Fenêtre de synchronisation (jours) # Fenêtre de synchronisation (jours)
SYNC_PAST_DAYS=7 SYNC_PAST_DAYS=7

1
.gitignore vendored
View File

@@ -39,6 +39,7 @@ coverage.xml
*.swp *.swp
*.swo *.swo
*~ *~
.zvec-grep/
# --- OS files --- # --- OS files ---
.DS_Store .DS_Store

View File

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

View File

@@ -140,10 +140,42 @@
"filename": "GUIDE_DEV_PYTHON.md", "filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": true, "is_verified": true,
"line_number": 5046, "line_number": 4940,
"is_secret": false "is_secret": false
} }
],
"tests/unit/test_caldav_gateway.py": [
{
"type": "Secret Keyword",
"filename": "tests/unit/test_caldav_gateway.py",
"hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f",
"is_verified": false,
"line_number": 152
},
{
"type": "Basic Auth Credentials",
"filename": "tests/unit/test_caldav_gateway.py",
"hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f",
"is_verified": false,
"line_number": 763
}
],
"tests/unit/test_caldav_security.py": [
{
"type": "Basic Auth Credentials",
"filename": "tests/unit/test_caldav_security.py",
"hashed_secret": "8e1f07a2939b6324c70f48a3e7f64b463a4a3f8b",
"is_verified": false,
"line_number": 27
},
{
"type": "Secret Keyword",
"filename": "tests/unit/test_caldav_security.py",
"hashed_secret": "6b554cd7b7e0115065fb4907307a74f1902154d4",
"is_verified": false,
"line_number": 28
}
] ]
}, },
"generated_at": "2026-09-06T21:06:14Z" "generated_at": "2026-09-07T10:24:08Z"
} }

File diff suppressed because it is too large Load Diff

58
TODO.md
View File

@@ -114,14 +114,14 @@ Récupérer le flux RSS du blog du collège, parser et dédupliquer les articles
Lire l'agenda théorique (JSON) via une interface de provider extensible, avec gestion de la parité des semaines (paire/impaire) et des vacances scolaires. Lire l'agenda théorique (JSON) via une interface de provider extensible, avec gestion de la parité des semaines (paire/impaire) et des vacances scolaires.
- [ ] Créer `sources/theoretical/provider.py` : protocole `TheoreticalAgendaProvider` (§8.2). - [x] Créer `sources/theoretical/provider.py` : protocole `TheoreticalAgendaProvider` (§8.2).
- [ ] Créer `sources/theoretical/file.py` : parser JSON → liste de `TheoreticalLesson` avec filtrage par parité de semaine (paire/impaire/toutes). - [x] Créer `sources/theoretical/file.py` : parser JSON → liste de `TheoreticalLesson` avec filtrage par parité de semaine (paire/impaire/toutes).
- [ ] Créer `sources/theoretical/parity.py` : service `WeekParityService` déterminant la parité d'une date à partir d'une date de référence configurée. - [x] Créer `sources/theoretical/parity.py` : service `WeekParityService` déterminant la parité d'une date à partir d'une date de référence configurée.
- [ ] Créer `sources/theoretical/holidays.py` : service `SchoolHolidayCalendar` lisant un fichier JSON de vacances scolaires (zone A) et exposant `is_holiday(date)`. - [x] Créer `sources/theoretical/holidays.py` : service `SchoolHolidayCalendar` lisant un fichier JSON de vacances scolaires (zone A) et exposant `is_holiday(date)`.
- [ ] Implémenter le provider JSON : filtrage par parité + vacances, génération d'identifiants déterministes incluant le type de semaine. - [x] Implémenter le provider JSON : filtrage par parité + vacances, génération d'identifiants déterministes incluant le type de semaine.
- [ ] Ajouter la configuration : `SCHOOL_HOLIDAYS_PATH`, `THEORETICAL_WEEK_ANCHOR_DATE`, `THEORETICAL_WEEK_ANCHOR_TYPE` dans `AppSettings`. - [x] Ajouter la configuration : `SCHOOL_HOLIDAYS_PATH`, `THEORETICAL_WEEK_ANCHOR_DATE`, `THEORETICAL_WEEK_ANCHOR_TYPE` dans `AppSettings`.
- [ ] Normaliser les matières et créneaux pour le matching déterministe. - [x] Normaliser les matières et créneaux pour le matching déterministe.
- [ ] Créer les fixtures : `tests/fixtures/theoretical.json` et `tests/fixtures/school_holidays.json`. - [x] Créer les fixtures : `tests/fixtures/theoretical.json` et `tests/fixtures/school_holidays.json`.
### Critères d'acceptation ### Critères d'acceptation
- `file.py` lit `tests/fixtures/theoretical.json` en `TheoreticalLesson` avec filtrage par parité. - `file.py` lit `tests/fixtures/theoretical.json` en `TheoreticalLesson` avec filtrage par parité.
@@ -131,24 +131,25 @@ Lire l'agenda théorique (JSON) via une interface de provider extensible, avec g
- Une configuration incomplète (ancre de parité manquante alors que des leçons `even`/`odd` existent) produit une erreur explicite. - Une configuration incomplète (ancre de parité manquante alors que des leçons `even`/`odd` existent) produit une erreur explicite.
--- ---
## M7. Synchronisation CalDAV — Priorité : Haute ## M7. Synchronisation CalDAV — Priorité : Haute
Synchroniser différentiellement les événements Pronote vers le calendrier CalDAV, de façon idempotente. Synchroniser différentiellement les événements Pronote vers le calendrier CalDAV, de façon idempotente.
- [ ] Créer `sync/caldav.py` : `CalDAVClient` (connexion, liste/ajout/MAJ/suppression, marqueur `X-PRONOTE-SYNC-MANAGED: v1`). - [x] Créer `sync/caldav.py` : passerelle CalDAV isolant la bibliothèque `caldav>=1.3.0` (connexion via `DAVClient`, résolution du calendrier via `calendar_path`, récupération/ajout/MAJ/suppression des événements, marqueur `X-PRONOTE-SYNC-MANAGED: v1`).
- [ ] Créer `sync/state.py` : état local de sync (SQLite ou JSON) assurant l'idempotence (UID connus). - [x] Calculer le `CalDAVSyncPlan` (to_add / to_update / to_remove) par UID stable, explicitement avant l'exécution de la sync.
- [ ] Calculer le `CalDAVSyncPlan` (to_add / to_update / to_remove) par UID stable. - [x] Implémenter l'exécution du plan : ajout, mise à jour (si modifié), suppression (si absent). En mode `dry_run`, loguer le plan sans écrire.
- [ ] 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. - [x] 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 des UID dans `sources/pronote/client.py` à la frontière des sources si nécessaire.
- [ ] Implémenter la sync différentielle : conserver les cours annulés (`STATUS:CANCELLED`), ne pas supprimer. - [x] 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`). - [x] Garantir l'idempotence (2 exécutions identiques → même `CalDAVSyncResult`), sans état local persistant (scan du calendrier distant).
- [ ] Réutiliser `BlogRSSState` pour l'état blog si pertinent (sinon `sync/blog_state.py`). - [x] Ne jamais modifier ou supprimer les événements non marqués `X-PRONOTE-SYNC-MANAGED`.
### Critères d'acceptation ### Critères d'acceptation
- Le plan de sync est correctement calculé (PronoteData vs état local).
- Le plan de sync est correctement calculé (données Pronote vs événements distants gérés).
- Un changement de source iCal ↔ `pronotepy` ne crée ni doublon ni suppression/ajout artificiel pour un cours équivalent. - Un changement de source iCal ↔ `pronotepy` ne crée ni doublon ni suppression/ajout artificiel pour un cours équivalent.
- Un run dry-run n'écrit rien ; deux runs identiques donnent un résultat identique. - Un run dry-run n'écrit rien ; deux runs identiques donnent un résultat identique.
- Les événements annulés restent (`STATUS:CANCELLED`) et sont marqués `MANAGED`. - Les événements annulés restent (`STATUS:CANCELLED`) et sont marqués `MANAGED`.
- Les événements non marqués ne sont jamais modifiés ni supprimés.
--- ---
@@ -156,15 +157,15 @@ Synchroniser différentiellement les événements Pronote vers le calendrier Cal
Comparer l'agenda réel et l'agenda théorique pour générer les ajouts/suppressions/modifications. Comparer l'agenda réel et l'agenda théorique pour générer les ajouts/suppressions/modifications.
- [ ] Créer `sync/diff.py` : `AgendaComparator` avec matching déterministe (jour + créneau avec tolérance + matière normalisée). - [x] Créer `sync/diff.py` : `AgendaComparator` avec matching déterministe (jour + créneau avec tolérance + matière normalisée).
- [ ] Générer `AgendaDiff` / `AgendaChange` (added / removed / modified). - [x] Générer `AgendaDiff` / `AgendaChange` (added / removed / modified).
- [ ] Appliquer la politique de départage : tri par UID stable puis comparaison exacte ; première correspondance en cas de multi-match (§8.4). - [x] Appliquer la politique de départage : tri par UID stable puis comparaison exacte ; première correspondance en cas de multi-match (§8.4).
- [ ] Gérer l'absence de fichier théorique (diff vide, non bloquant). - [x] Gérer l'absence de fichier théorique (diff vide, non bloquant).
### Critères d'acceptation ### Critères d'acceptation
- La comparaison produit les bons `added`/`removed`/`modified`. - La comparaison produit les bons `added`/`removed`/`modified`.
- Le matching est déterministe (même entrée → même résultat). - Le matching est déterministe (même entrée → même résultat).
- Sans `THEORETICAL_AGENDA_PATH`, retourne un diff vide sans erreur. - Sans `THEORETICAL_AGENDA_PATH`, retourne un diff vide sans erreur. *(Couvert par design : `AgendaComparator` exige un provider non optionnel ; la composition root produit un diff vide si absent. Validation runtime reportée à M11.)*
--- ---
@@ -220,6 +221,7 @@ Composer et orchestrer toutes les étapes avec gestion d'erreurs dégradée et m
- Une sélection iCal commune à l'agenda et aux devoirs ne déclenche qu'un téléchargement/parsing du flux par run. - Une sélection iCal commune à l'agenda et aux devoirs ne déclenche qu'un téléchargement/parsing du flux par run.
- Une erreur non critique (ex : synthèse IA) n'empêche pas l'envoi XMPP. - Une erreur non critique (ex : synthèse IA) n'empêche pas l'envoi XMPP.
- `dry_run=True` n'effectue aucune écriture ; aucune source disponible → erreur critique explicite. - `dry_run=True` n'effectue aucune écriture ; aucune source disponible → erreur critique explicite.
- Si `THEORETICAL_AGENDA_PATH` est absent, le pipeline produit un diff vide sans erreur et n'instancie pas `AgendaComparator` ; si présent, il instancie le comparateur et effectue la comparaison.
--- ---
@@ -245,12 +247,12 @@ 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.json`, `school_holidays.json`, `blog_rss.xml` (anonymisés, sans `icalsecurise`). - [ ] Créer `tests/fixtures/` : `pronote-4e.ics`, `pronote-6e.ics`, `theoretical.json`, `school_holidays.json`, `blog_rss.xml` (anonymisés, sans `icalsecurise`).
- [ ] Créer `tests/conftest.py` : fixtures partagées (sample_lesson, sample_cancelled_lesson, sample_homework, sample_school_event, sample_message, sample_pronote_data…). - [ ] Créer `tests/conftest.py` : fixtures partagées (sample_lesson, sample_cancelled_lesson, sample_homework, sample_school_event, sample_message, sample_pronote_data…).
- [ ] Écrire `tests/unit/` : `test_models`, `test_parsing` (iCal), `test_uid`, `test_redaction`, `test_diff`, `test_sync`. - [x] É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. - [x] 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é). - [x] Écrire `tests/integration/` : `test_pipeline`, `test_caldav` (mocké), `test_xmpp` (mocké).
- [ ] Écrire `tests/e2e/test_cli.py` : exécution CLI en dry-run. - [x] Écrire `tests/e2e/test_cli.py` : exécution CLI en dry-run.
- [ ] Tests sans réseau (mocks `responses`/`aioresponses`/`pytest-mock`) ; couverture ≥ 90 %. - [x] Tests sans réseau (mocks `responses`/`aioresponses`/`pytest-mock`) ; couverture ≥ 90 %.
- [ ] 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). - [x] Ajouter un test négatif : les messages, logs, causes, contextes et tracebacks complets ne fuient pas de secrets (`icalsecurise`, clés API, mots de passe).
### Critères d'acceptation ### Critères d'acceptation
- `pytest` passe et `pytest --cov` atteint ≥ 90 % (`fail_under = 90`). - `pytest` passe et `pytest --cov` atteint ≥ 90 % (`fail_under = 90`).

View File

@@ -10,10 +10,13 @@ from __future__ import annotations
from datetime import date from datetime import date
from typing import Literal from typing import Literal
from urllib.parse import urlparse
from pydantic import Field, SecretStr, field_serializer from pydantic import Field, SecretStr, ValidationInfo, field_serializer, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
from pronote_sync.utils.redaction import redact_url
class PronoteSettings(BaseSettings): class PronoteSettings(BaseSettings):
"""Paramètres d'accès à Pronote (flux iCal et API ``pronotepy``). """Paramètres d'accès à Pronote (flux iCal et API ``pronotepy``).
@@ -51,16 +54,75 @@ class CalDAVSettings(BaseSettings):
"""Paramètres d'accès au serveur CalDAV de destination. """Paramètres d'accès au serveur CalDAV de destination.
Les variables d'environnement correspondantes sont préfixées par Les variables d'environnement correspondantes sont préfixées par
``CALDAV_``. ``CALDAV_``. L'URL est traitée comme potentiellement sensible (au même
titre que ``PRONOTE_ICAL_URL``) : elle est de type ``SecretStr`` et
masquée lors de la sérialisation. Par défaut, seul HTTPS est accepté ;
HTTP n'est toléré que pour un hôte de boucle locale (``localhost``,
``127.0.0.1``, ``::1``) lorsque ``allow_insecure_http`` vaut ``True``.
""" """
model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="CALDAV_") model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="CALDAV_")
url: str | None = None allow_insecure_http: bool = False
url: SecretStr | None = None
username: str | None = None username: str | None = None
password: SecretStr | None = None password: SecretStr | None = None
calendar_path: str = "/pronote-sync/" calendar_path: str = "/pronote-sync/"
@field_serializer("url")
def _serialize_url(self, value: SecretStr | None) -> str | None:
"""Masque l'URL CalDAV lors de la sérialisation (repr, str, JSON).
:param value: Valeur du champ ``url`` (secret potentiel).
:return: URL avec les éléments sensibles remplacés par ``REDACTED``,
ou ``None`` si la valeur est absente.
:rtype: str | None
"""
if value is None:
return None
return redact_url(value.get_secret_value())
@field_validator("url")
@classmethod
def _validate_url_https(cls, v: SecretStr | None, info: ValidationInfo) -> SecretStr | None:
"""Valide le schéma de l'URL CalDAV (HTTPS obligatoire par défaut).
HTTPS est toujours accepté. HTTP n'est accepté que pour un hôte de
boucle locale (``localhost``, ``127.0.0.1``, ``::1``) et uniquement
lorsque ``allow_insecure_http`` vaut ``True``. Les messages d'erreur
ne contiennent jamais l'URL brute (susceptible de contenir des
identifiants).
:param v: Valeur du champ ``url`` à valider.
:param info: Contexte de validation (accès aux autres champs).
:return: La valeur validée inchangée.
:rtype: SecretStr | None
:raises ValueError: Si le schéma n'est pas supporté ou si l'URL HTTP
n'est pas autorisée.
"""
if v is None:
return v
raw_url = v.get_secret_value()
parsed = urlparse(raw_url)
if parsed.scheme not in ("http", "https"):
raise ValueError("URL CalDAV invalide : schéma non supporté") from None
if parsed.scheme == "https":
return v
# HTTP — check allow_insecure_http flag and loopback
allow_insecure = info.data.get("allow_insecure_http", False)
if not allow_insecure:
raise ValueError(
"URL CalDAV non sécurisée : HTTPS requis (ou activer "
"CALDAV_ALLOW_INSECURE_HTTP pour localhost)"
) from None
hostname = parsed.hostname or ""
loopback_hosts = {"localhost", "127.0.0.1", "::1"}
if hostname not in loopback_hosts:
raise ValueError(
"URL CalDAV non sécurisée : HTTP autorisé uniquement pour localhost"
) from None
return v
class XmppSettings(BaseSettings): class XmppSettings(BaseSettings):
"""Paramètres du canal de notifications XMPP (désactivé par défaut). """Paramètres du canal de notifications XMPP (désactivé par défaut).

View File

@@ -23,6 +23,7 @@ from pronote_sync.models.agenda import Lesson, LessonStatus
from pronote_sync.models.homework import Homework from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType from pronote_sync.models.message import Message, MessageType
from pronote_sync.utils.redaction import redact_exception from pronote_sync.utils.redaction import redact_exception
from pronote_sync.utils.uid import generate_deterministic_uid, normalize_pronote_uid
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -270,6 +271,12 @@ class PronoteClient:
def get_lessons(self, start: date, end: date) -> list[Lesson]: def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours via ``pronotepy`` (repli iCal). """Récupère les cours via ``pronotepy`` (repli iCal).
Les UIDs des cours sont normalisés comme ceux du flux iCal via
:func:`normalize_pronote_uid` afin que la même leçon produise le
même identifiant quelle que soit la source ; en l'absence d'UID
exploitable, un UID déterministe est généré via
:func:`generate_deterministic_uid`.
Les exceptions ne sont pas attrapées : elles se propagent afin que 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 l'appelant puisse détecter l'échec et déclencher le repli (ou une
erreur explicite). erreur explicite).
@@ -288,9 +295,21 @@ class PronoteClient:
lessons: list[Lesson] = [] lessons: list[Lesson] = []
for lesson in client.lessons(start, end): for lesson in client.lessons(start, end):
content = lesson.content content = lesson.content
raw_uid = lesson.id
if raw_uid:
uid = normalize_pronote_uid(raw_uid)
else:
uid = generate_deterministic_uid(
start=lesson.start,
end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "",
teachers=list(lesson.teacher_names or ()),
rooms=list(lesson.classrooms or ()),
group=lesson.group_name,
)
lessons.append( lessons.append(
Lesson( Lesson(
id=lesson.id, id=uid,
start=lesson.start, start=lesson.start,
end=lesson.end, end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "", subject=lesson.subject.name if lesson.subject is not None else "",

View File

@@ -20,6 +20,7 @@ from pronote_sync.sources.theoretical.holidays import SchoolHolidayCalendar
from pronote_sync.sources.theoretical.model import TheoreticalAgendaFile, TheoreticalLessonEntry from pronote_sync.sources.theoretical.model import TheoreticalAgendaFile, TheoreticalLessonEntry
from pronote_sync.sources.theoretical.parity import WeekParityService from pronote_sync.sources.theoretical.parity import WeekParityService
from pronote_sync.utils.redaction import redact_exception, redact_secrets from pronote_sync.utils.redaction import redact_exception, redact_secrets
from pronote_sync.utils.text import normalize_subject as normalize_subject
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -36,7 +37,7 @@ def _generate_id(entry: TheoreticalLessonEntry) -> str:
:return: Identifiant déterministe unique. :return: Identifiant déterministe unique.
:rtype: str :rtype: str
""" """
subject_slug = entry.subject.lower().strip().replace(" ", "-") subject_slug = normalize_subject(entry.subject).replace(" ", "-")
return f"theoretical:{entry.week}:{entry.day_of_week}:{entry.start_time}-{entry.end_time}:{subject_slug}" return f"theoretical:{entry.week}:{entry.day_of_week}:{entry.start_time}-{entry.end_time}:{subject_slug}"
@@ -111,7 +112,7 @@ class JsonTheoreticalAgendaProvider:
if effective_id in seen_ids: if effective_id in seen_ids:
raise PronoteSyncError( raise PronoteSyncError(
f"Conflit d'identifiant dans l'agenda théorique : " f"Conflit d'identifiant dans l'agenda théorique : "
f"l'identifiant '{effective_id}' est utilisé par plusieurs leçons. " f"l'identifiant '{redact_secrets(effective_id)}' est utilisé par plusieurs leçons. "
f"Fournissez des identifiants explicites uniques." f"Fournissez des identifiants explicites uniques."
) from None ) from None
seen_ids.add(effective_id) seen_ids.add(effective_id)

View File

@@ -0,0 +1,5 @@
"""Module de synchronisation CalDAV."""
from pronote_sync.sync.synchronizer import synchronize
__all__ = ["synchronize"]

352
pronote_sync/sync/caldav.py Normal file
View File

@@ -0,0 +1,352 @@
"""Passerelle d'accès au calendrier CalDAV.
Ce module fournit :class:`CalDAVGateway`, une passerelle qui isole la
bibliothèque ``caldav`` du reste du pipeline de synchronisation. Elle gère
la connexion au serveur CalDAV, la résolution du calendrier de destination,
la liste des événements gérés par l'outil, ainsi que l'écriture et la
suppression d'événements.
La passerelle applique des contraintes de sécurité strictes : le mot de
passe n'est extrait de son ``SecretStr`` que localement, au moment de créer
le client, et aucun secret (mot de passe, URL brute) n'est conservé sur
l'instance après la connexion. Toute exception de la bibliothèque ``caldav``
est interceptée puis re-levée sous la forme d'une
:class:`~pronote_sync.errors.PronoteSyncError` — sans chaînage — dont le
message ne contient aucune donnée sensible.
"""
from __future__ import annotations
import logging
from collections.abc import Callable
from datetime import datetime
from typing import Any, cast
from urllib.parse import urlparse
import caldav
from caldav.lib.error import NotFoundError
from icalendar import Component
from pydantic import SecretStr
from pronote_sync.config.settings import CalDAVSettings
from pronote_sync.errors import PronoteSyncError
from pronote_sync.sync.serialization import MANAGED_PROPERTY, MANAGED_VALUE
from pronote_sync.utils.redaction import redact_exception, redact_secrets, redact_url
from pronote_sync.utils.uid import normalize_pronote_uid
logger = logging.getLogger(__name__)
class CalDAVGateway:
"""Passerelle d'accès au calendrier CalDAV, isolant la bibliothèque caldav.
La passerelle gère la connexion, la résolution du calendrier de
destination, la récupération des événements portant le marqueur de
gestion (:data:`MANAGED_PROPERTY`), leur écriture et leur suppression.
Seuls les paramètres non sensibles nécessaires (``calendar_path``,
``username``) ainsi que l'URL rédigée sont mémorisés sur l'instance ; le
mot de passe et l'URL brute ne sont jamais conservés en clair.
L'usage typique se fait via le gestionnaire de contexte ::
with CalDAVGateway(settings) as gateway:
gateway.upsert_event(vcalendar_text, uid)
"""
def __init__(
self,
settings: CalDAVSettings,
client_factory: Callable[..., Any] | None = None,
) -> None:
"""Initialise la passerelle avec la configuration CalDAV.
Extrait uniquement les paramètres non sensibles nécessaires
(``calendar_path``, ``username``) ainsi que l'URL rédigée pour la
journalisation. Le mot de passe reste encapsulé dans son
``SecretStr`` et n'est jamais stocké en clair sur l'instance.
:param settings: Configuration CalDAV (url, username, password,
calendar_path).
:param client_factory: Appelable optionnel créant une instance
``DAVClient`` (permet l'injection de dépendances en test). Si
``None``, utilise ``caldav.DAVClient``.
"""
# ``cast`` nécessaire : mypy ne résout pas le ré-export du module
# ``caldav`` (le type de ``caldav.DAVClient`` est vu comme ``object``).
self._client_factory: Callable[..., Any] = (
client_factory
if client_factory is not None
else cast(Callable[..., Any], caldav.DAVClient)
)
self._calendar_path: str = settings.calendar_path
self._redacted_url: str | None = (
redact_url(settings.url.get_secret_value()) if settings.url else None
)
self._username: str | None = settings.username
self._url_secret: SecretStr | None = settings.url
self._password_secret: SecretStr | None = settings.password
self._client: Any = None
self._calendar: Any = None
def _resolve_calendar(self) -> Any:
"""Résout le calendrier cible via la découverte CalDAV.
Interroge le principal CalDAV puis sa liste de calendriers, et
sélectionne celui dont le chemin d'URL correspond au
``calendar_path`` configuré à la frontière d'un composant de
chemin (barres obliques finales ignorées, préfixe ``/`` garanti par
la normalisation).
:return: Le calendrier CalDAV correspondant au chemin configuré.
:rtype: Any
:raises PronoteSyncError: Si aucun calendrier ne correspond ou si
plusieurs calendriers correspondent au chemin configuré.
"""
principal = self._client.principal()
calendars = principal.calendars()
normalized_path = self._calendar_path.strip("/")
matches: list[Any] = []
for cal in calendars:
cal_url = str(cal.url) if hasattr(cal, "url") and cal.url else ""
cal_path = urlparse(cal_url).path.strip("/")
if cal_path == normalized_path or cal_path.endswith(f"/{normalized_path}"):
matches.append(cal)
if len(matches) == 0:
raise PronoteSyncError(
f"Calendrier CalDAV introuvable : {self._redacted_url}"
) from None
if len(matches) > 1:
raise PronoteSyncError(
f"Calendrier CalDAV ambigu : plusieurs calendriers "
f"correspondent à '{self._calendar_path}'"
) from None
return matches[0]
def connect(self) -> None:
"""Établit la connexion au serveur CalDAV et résout le calendrier.
Le mot de passe est extrait de son ``SecretStr`` uniquement pour la
création du ``DAVClient``, en variable locale, puis abandonné. Le
calendrier cible est résolu par découverte
(:meth:`_resolve_calendar`) plutôt que par concaténation d'URL. Toute
exception de la bibliothèque ``caldav`` est interceptée, journalisée
avec :func:`redact_exception` et re-levée en
:class:`PronoteSyncError` — hors du bloc ``except``, afin que
``__context__`` ne retienne aucune exception brute — sans chaînage ni
donnée sensible. Les erreurs de résolution du calendrier
(message « introuvable » ou « ambigu ») sont propagées telles quelles.
:raises PronoteSyncError: Si la configuration est incomplète ou si la
connexion au serveur CalDAV échoue.
"""
if self._url_secret is None or self._username is None or self._password_secret is None:
raise PronoteSyncError(
"Configuration CalDAV incomplète : url, username et password sont requis"
) from None
raw_url = self._url_secret.get_secret_value()
password = self._password_secret.get_secret_value()
error_msg: str | None = None
try:
self._client = self._client_factory(
url=raw_url, username=self._username, password=password
)
self._calendar = self._resolve_calendar()
except PronoteSyncError:
raise
except Exception as exc:
error_msg = redact_exception(exc)
logger.error("Échec de la connexion CalDAV : %s", error_msg)
if error_msg is not None:
raise PronoteSyncError(f"Échec de la connexion CalDAV : {self._redacted_url}") from None
def list_managed_events(self, start: datetime, end: datetime) -> list[tuple[str, str, Any]]:
"""Liste les événements gérés par l'outil dans la fenêtre donnée.
Interroge le serveur CalDAV sur la fenêtre ``[start, end]`` et ne
conserve que les VEVENT portant le marqueur de gestion
(:data:`MANAGED_PROPERTY` avec la valeur :data:`MANAGED_VALUE`).
Pour chaque VEVENT, l'UID brut tel que stocké sur le serveur est
conservé ainsi que sa forme canonique obtenue via
:func:`~pronote_sync.utils.uid.normalize_pronote_uid` — la même
normalisation que celle appliquée aux événements locaux — afin que
le planificateur puisse apparier les événements distants suffixés aux
événements Pronote normalisés.
:param start: Début de la fenêtre de recherche.
:param end: Fin de la fenêtre de recherche.
:return: Triplets ``(raw_uid, canonical_uid, vevent)`` pour chaque
événement géré trouvé.
:rtype: list[tuple[str, str, Any]]
:raises PronoteSyncError: Si la passerelle n'est pas connectée ou si
la récupération échoue.
"""
if self._calendar is None:
raise PronoteSyncError("Passerelle CalDAV non connectée") from None
result: list[tuple[str, str, Any]] = []
error_msg: str | None = None
try:
events = self._calendar.search(start=start, end=end, event=True, expand=True)
for event in events:
component: Component = event.icalendar_component
for vevent in component.walk("VEVENT"):
managed = vevent.get(MANAGED_PROPERTY)
if managed is not None and str(managed) == MANAGED_VALUE:
raw_uid = str(vevent.get("UID"))
canonical_uid = normalize_pronote_uid(raw_uid)
result.append((raw_uid, canonical_uid, vevent))
except Exception as exc:
error_msg = redact_exception(exc)
logger.error("Échec de la récupération des événements CalDAV : %s", error_msg)
if error_msg is not None:
raise PronoteSyncError(
f"Échec de la récupération des événements CalDAV : {self._redacted_url}"
) from None
return result
def _is_managed_event(self, event: Any) -> bool:
"""Détermine si un événement distant est géré par pronote-sync.
Vérifie la présence du marqueur de gestion (:data:`MANAGED_PROPERTY`
avec la valeur :data:`MANAGED_VALUE`) sur au moins un des composants
VEVENT de l'événement, selon le même motif que
:meth:`list_managed_events`.
:param event: Objet événement distant exposant la propriété
``icalendar_component`` retournant un ``icalendar.Calendar``.
:return: ``True`` si l'événement porte le marqueur de gestion,
``False`` sinon.
:rtype: bool
"""
component: Component = event.icalendar_component
for vevent in component.walk("VEVENT"):
managed = vevent.get(MANAGED_PROPERTY)
if managed is not None and str(managed) == MANAGED_VALUE:
return True
return False
def upsert_event(self, vcalendar_text: str, uid: str) -> None:
"""Crée ou met à jour un événement CalDAV identifié par son UID.
Recherche d'abord l'événement existant par UID via
``get_event_by_uid`` : s'il est introuvable (``NotFoundError``), un
nouvel événement est créé via ``add_event``. S'il existe, son contenu
est remplacé puis sauvegardé — uniquement si l'événement est géré par
l'outil (marqueur :data:`MANAGED_PROPERTY` avec la valeur
:data:`MANAGED_VALUE`). Un événement existant non géré provoque une
:class:`PronoteSyncError` explicite et n'est jamais modifié. Toute
autre exception est journalisée avec :func:`redact_exception` puis
re-levée en :class:`PronoteSyncError` — sans chaînage ni donnée
sensible.
:param vcalendar_text: Document iCalendar complet (VCALENDAR + VEVENT).
:param uid: UID stable de l'événement à créer ou mettre à jour.
:raises PronoteSyncError: Si la passerelle n'est pas connectée, si
l'événement distant n'est pas géré par l'outil, ou si l'opération
échoue.
"""
if self._calendar is None:
raise PronoteSyncError("Passerelle CalDAV non connectée") from None
error_msg: str | None = None
try:
try:
event = self._calendar.get_event_by_uid(uid)
except NotFoundError:
self._calendar.add_event(ical=vcalendar_text)
else:
if not self._is_managed_event(event):
raise PronoteSyncError(
"Conflit d'UID : l'événement distant n'est pas géré par pronote-sync"
) from None
event.data = vcalendar_text
event.save()
except PronoteSyncError:
raise
except Exception as exc:
error_msg = redact_exception(exc)
logger.error(
"Échec de l'écriture d'un événement CalDAV (uid=%s) : %s",
redact_secrets(uid),
error_msg,
)
if error_msg is not None:
raise PronoteSyncError(
f"Échec de l'écriture d'un événement CalDAV : {self._redacted_url}"
) from None
def delete_event(self, uid: str) -> None:
"""Supprime un événement du calendrier, identifié par son UID.
Récupère l'événement distant via ``get_event_by_uid`` : s'il est
introuvable (``NotFoundError``), la suppression est un succès
idempotent et la méthode retourne silencieusement. S'il existe, il
n'est supprimé que s'il est géré par l'outil (marqueur
:data:`MANAGED_PROPERTY` avec la valeur :data:`MANAGED_VALUE`) ; un
événement non géré est laissé intact, un avertissement est
journalisé et la méthode retourne sans erreur. Toute autre exception
est journalisée avec :func:`redact_exception` et re-levée en
:class:`PronoteSyncError` — sans chaînage ni donnée sensible.
:param uid: Identifiant UID de l'événement à supprimer.
:raises PronoteSyncError: Si la passerelle n'est pas connectée ou si
la suppression échoue.
"""
if self._calendar is None:
raise PronoteSyncError("Passerelle CalDAV non connectée") from None
error_msg: str | None = None
try:
event = self._calendar.get_event_by_uid(uid)
if not self._is_managed_event(event):
logger.warning(
"Suppression refusée : l'événement distant UID=%s n'est pas géré par "
"pronote-sync",
redact_secrets(uid),
)
return
event.delete()
except NotFoundError:
return
except Exception as exc:
error_msg = redact_exception(exc)
logger.error(
"Échec de la suppression d'un événement CalDAV (uid=%s) : %s",
redact_secrets(uid),
error_msg,
)
if error_msg is not None:
raise PronoteSyncError(
f"Échec de la suppression d'un événement CalDAV : {self._redacted_url}"
) from None
def close(self) -> None:
"""Libère les ressources : client, calendrier et secrets.
Réinitialise le client, le calendrier et les ``SecretStr`` conservés
afin de ne laisser aucune référence à des données sensibles sur
l'instance.
"""
self._client = None
self._calendar = None
self._url_secret = None
self._password_secret = None
def __enter__(self) -> CalDAVGateway:
"""Entre dans le contexte en établissant la connexion.
:return: La passerelle connectée.
:rtype: CalDAVGateway
:raises PronoteSyncError: Si la connexion échoue.
"""
self.connect()
return self
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
"""Quitte le contexte en libérant les ressources.
Les exceptions éventuellement en cours ne sont pas interceptées et
continuent leur propagation normale.
:param exc_type: Type de l'exception en cours, le cas échéant.
:param exc_val: Instance de l'exception en cours, le cas échéant.
:param exc_tb: Traceback de l'exception en cours, le cas échéant.
"""
self.close()

206
pronote_sync/sync/diff.py Normal file
View File

@@ -0,0 +1,206 @@
"""Comparaison entre l'agenda réel et l'agenda théorique.
Ce module définit :class:`AgendaComparator`, responsable de produire un
:class:`AgendaDiff` en appariant les cours réels (:class:`Lesson`) aux cours
théoriques (:class:`TheoreticalLesson`) fournis par un
:class:`TheoreticalAgendaProvider`. L'appariement est tolérant sur les horaires
(±15 minutes) et normalise les matières. Le résultat est déterministe : il ne
dépend ni de l'ordre des entrées du fournisseur, ni de l'ordre des cours réels
pour le matching.
"""
from __future__ import annotations
from datetime import date, datetime, time
from pronote_sync.models.agenda import Lesson, LessonStatus, TheoreticalLesson
from pronote_sync.models.diff import AgendaChange, AgendaChangeType, AgendaDiff
from pronote_sync.sources.theoretical.provider import TheoreticalAgendaProvider
from pronote_sync.utils.text import normalize_subject
#: Tolérance temporelle en minutes (valeur absolue) pour l'appariement.
_TOLERANCE_MINUTES = 15
def _minutes_since_midnight(dt: datetime) -> int:
"""Retourne le nombre de minutes écoulées depuis minuit pour un datetime.
Les secondes sont ignorées.
:param dt: Date/heure à convertir.
:return: Nombre de minutes (heure * 60 + minute).
:rtype: int
"""
return dt.hour * 60 + dt.minute
def _time_minutes(t: time) -> int:
"""Retourne le nombre de minutes écoulées depuis minuit pour un time.
Les secondes sont ignorées.
:param t: Heure à convertir.
:return: Nombre de minutes (heure * 60 + minute).
:rtype: int
"""
return t.hour * 60 + t.minute
class AgendaComparator:
"""Compare l'agenda réel à l'agenda théorique pour une date cible.
:class:`AgendaComparator` apparie chaque cours réel au cours théorique qui
lui correspond (tolérance temporelle ±15 minutes et matière normalisée),
détecte les cours ajoutés, supprimés et modifiés, puis produit un
:class:`AgendaDiff` ordonné de manière déterministe.
"""
def __init__(self, theoretical_provider: TheoreticalAgendaProvider) -> None:
"""Initialise le comparateur avec un fournisseur d'agenda théorique.
:param theoretical_provider: Fournisseur des cours théoriques.
:rtype: None
"""
self._theoretical_provider = theoretical_provider
def compare(self, real_lessons: list[Lesson], target_date: date) -> AgendaDiff:
"""Compare les cours réels aux cours théoriques pour la date cible.
Les changements sont émis dans un ordre déterministe : d'abord les cours
réels dans leur ordre d'entrée (ADDED ou MODIFIED), puis les cours
théoriques non appariés par existence (REMOVED) triés par identifiant.
:param real_lessons: Liste des cours réels (dans leur ordre d'entrée).
:param target_date: Date cible de la comparaison.
:return: Le diff entre l'agenda réel et l'agenda théorique.
:rtype: AgendaDiff
"""
theoretical_lessons = self._theoretical_provider.get_lessons(target_date)
#: Identifiants des cours théoriques candidats d'au moins un cours réel
#: (appariement par existence pour la détection des suppressions).
matched_by_existence: set[str] = set()
changes: list[AgendaChange] = []
for real in real_lessons:
candidates = [
theoretical
for theoretical in theoretical_lessons
if self._matches(real, theoretical, target_date)
]
matched_by_existence.update(candidate.id for candidate in candidates)
selected = min(candidates, key=lambda candidate: candidate.id) if candidates else None
if selected is None:
changes.append(
AgendaChange(
type=AgendaChangeType.ADDED,
lesson=real,
theoretical_lesson=None,
details="Cours ajouté par rapport à l'agenda théorique",
)
)
elif self._is_modified(real, selected):
changes.append(
AgendaChange(
type=AgendaChangeType.MODIFIED,
lesson=real,
theoretical_lesson=selected,
details=self._describe_changes(real, selected),
)
)
for theoretical in sorted(theoretical_lessons, key=lambda lesson: lesson.id):
if theoretical.id not in matched_by_existence:
changes.append(
AgendaChange(
type=AgendaChangeType.REMOVED,
lesson=None,
theoretical_lesson=theoretical,
details="Cours supprimé par rapport à l'agenda théorique",
)
)
return AgendaDiff(target_date=target_date, changes=tuple(changes))
def _matches(
self,
real: Lesson,
theoretical: TheoreticalLesson,
target_date: date,
) -> bool:
"""Détermine si un cours théorique est candidat d'un cours réel.
Un cours théorique est candidat d'un cours réel si le jour de la semaine
correspond, si les horaires de début et de fin coïncident à ±15 minutes
près et si les matières normalisées sont identiques.
:param real: Cours réel.
:param theoretical: Cours théorique candidat.
:param target_date: Date cible de la comparaison.
:return: ``True`` si le cours théorique correspond au cours réel.
:rtype: bool
"""
if theoretical.day_of_week != target_date.weekday():
return False
if abs(_time_minutes(theoretical.start_time) - _minutes_since_midnight(real.start)) > (
_TOLERANCE_MINUTES
):
return False
if abs(_time_minutes(theoretical.end_time) - _minutes_since_midnight(real.end)) > (
_TOLERANCE_MINUTES
):
return False
return normalize_subject(theoretical.subject) == normalize_subject(real.subject)
def _is_modified(self, real: Lesson, theoretical: TheoreticalLesson) -> bool:
"""Détermine si un cours réel apparié diffère de son cours théorique.
Un cours est considéré modifié si au moins un horaire diffère à la
minute près, si la matière normalisée diffère, si les professeurs ou les
salles diffèrent (comparaison par ensemble), ou si le statut n'est pas
``NORMAL``.
:param real: Cours réel apparié.
:param theoretical: Cours théorique apparié.
:return: ``True`` si le cours réel diffère du cours théorique.
:rtype: bool
"""
if real.start.time() != theoretical.start_time or real.end.time() != theoretical.end_time:
return True
if normalize_subject(real.subject) != normalize_subject(theoretical.subject):
return True
if set(real.teachers) != set(theoretical.teachers):
return True
if set(real.rooms) != set(theoretical.rooms):
return True
return real.status != LessonStatus.NORMAL
def _describe_changes(self, real: Lesson, theoretical: TheoreticalLesson) -> str:
"""Génère une description lisible des différences entre deux cours.
Les différences détectées sont décrites sous forme d'éléments séparés
par ``"; "``, en utilisant les valeurs originales (non normalisées) des
matières et des ensembles de professeurs/salles.
:param real: Cours réel apparié.
:param theoretical: Cours théorique apparié.
:return: Description lisible des différences.
:rtype: str
"""
parts: list[str] = []
if real.start.time() != theoretical.start_time or real.end.time() != theoretical.end_time:
parts.append(
f"horaires: {theoretical.start_time.strftime('%H:%M')}"
f"{theoretical.end_time.strftime('%H:%M')}"
f"{real.start.strftime('%H:%M')}{real.end.strftime('%H:%M')}"
)
if normalize_subject(real.subject) != normalize_subject(theoretical.subject):
parts.append(f"matière: {theoretical.subject}{real.subject}")
if set(real.teachers) != set(theoretical.teachers):
parts.append(f"professeurs: {set(theoretical.teachers)}{set(real.teachers)}")
if set(real.rooms) != set(theoretical.rooms):
parts.append(f"salles: {set(theoretical.rooms)}{set(real.rooms)}")
if real.status != LessonStatus.NORMAL:
parts.append(f"statut: {real.status.value}")
return "; ".join(parts)

View File

@@ -0,0 +1,225 @@
"""Exécuteur du plan de synchronisation CalDAV.
Ce module fournit :class:`CalDAVSyncExecutor`, qui applique un
:class:`~pronote_sync.models.sync.CalDAVSyncPlan` contre une
:class:`~pronote_sync.sync.caldav.CalDAVGateway` et produit un
:class:`~pronote_sync.models.sync.CalDAVSyncResult` avec des compteurs et
les éventuelles erreurs expurgées. Chaque opération (ajout, mise à jour,
suppression) est indépendante : l'échec d'un événement n'interrompt pas le
lot. En mode ``dry_run``, aucune écriture n'est envoyée à la passerelle,
mais le résultat reflète les opérations qui auraient été effectuées.
"""
from __future__ import annotations
import logging
from collections.abc import Mapping
from pronote_sync.errors import PronoteSyncError
from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.sync import (
CalDAVSyncPlan,
CalDAVSyncResult,
CalDAVSyncStatus,
)
from pronote_sync.sync.caldav import CalDAVGateway
from pronote_sync.sync.serialization import model_to_vcalendar_text
from pronote_sync.utils.redaction import redact_secrets
logger = logging.getLogger(__name__)
def _model_uid(model: Lesson | Homework | SchoolEvent) -> str:
"""Retourne l'UID iCalendar correspondant à un modèle Pronote.
L'UID reproduit la convention de :mod:`pronote_sync.sync.serialization`
(préfixes ``homework-`` et ``school-event-``) afin de journaliser des
identifiants stables, identiques à ceux envoyés à la passerelle.
:param model: Modèle Pronote concerné.
:return: UID iCalendar du modèle.
:rtype: str
"""
if isinstance(model, Lesson):
return str(model.id)
if isinstance(model, Homework):
return f"homework-{model.id}"
return f"school-event-{model.label}-{model.from_date.isoformat()}"
class CalDAVSyncExecutor:
"""Exécute un plan de synchronisation CalDAV contre une passerelle distante.
Les opérations du plan sont traitées une à une, indépendamment : une
erreur ``PronoteSyncError`` sur un événement est consignée dans le
résultat (message expurgé) sans interrompre le traitement du lot. En
mode ``dry_run``, la passerelle n'est jamais appelée en écriture ;
les compteurs du résultat reflètent néanmoins ce qui aurait été fait.
"""
def __init__(self, gateway: CalDAVGateway, dry_run: bool = False) -> None:
"""Initialise l'exécuteur avec la passerelle CalDAV.
:param gateway: Passerelle CalDAV connectée.
:param dry_run: Si ``True``, aucune écriture n'est effectuée ; seuls
les logs et le résultat sont renseignés.
"""
self._gateway = gateway
self._dry_run = dry_run
def execute(
self,
plan: CalDAVSyncPlan,
*,
remote_raw_by_canonical: Mapping[str, str] | None = None,
) -> CalDAVSyncResult:
"""Exécute le plan de synchronisation et retourne le résultat.
Les cours, devoirs et événements scolaires sont traités dans l'ordre
« ajouts, mises à jour, suppressions ». Une erreur
``PronoteSyncError`` sur une opération est consignée dans
``result.errors`` (message expurgé) sans stopper les autres
opérations ; toute autre exception (erreur de programmation) se
propage. Le statut final vaut ``FAILED`` si au moins une erreur a été
consignée, ``SKIPPED`` si aucune opération n'était à effectuer
(reprise idempotente), sinon ``SUCCESS``.
:param plan: Plan de synchronisation à appliquer.
:param remote_raw_by_canonical: Mapping canonical_uid -> raw_uid des
événements distants gérés, requis pour cibler l'UID brut lors des
mises à jour. ``None`` ou une clé absente entraîne une erreur
consignée dans ``result.errors`` pour chaque mise à jour concernée.
:return: Résultat de la synchronisation (statut, compteurs, erreurs).
:rtype: CalDAVSyncResult
"""
result = CalDAVSyncResult(status=CalDAVSyncStatus.SUCCESS, added=0, updated=0, removed=0)
lesson: Lesson
for lesson in plan.lessons_to_add:
self._do_save(lesson, result, is_update=False)
for lesson in plan.lessons_to_update:
self._do_save(
lesson,
result,
is_update=True,
remote_raw_by_canonical=remote_raw_by_canonical,
)
for uid in plan.lessons_to_remove:
self._do_delete(uid, result)
homework: Homework
for homework in plan.homeworks_to_add:
self._do_save(homework, result, is_update=False)
for homework in plan.homeworks_to_update:
self._do_save(
homework,
result,
is_update=True,
remote_raw_by_canonical=remote_raw_by_canonical,
)
for uid in plan.homeworks_to_remove:
self._do_delete(uid, result)
school_event: SchoolEvent
for school_event in plan.school_events_to_add:
self._do_save(school_event, result, is_update=False)
for school_event in plan.school_events_to_update:
self._do_save(
school_event,
result,
is_update=True,
remote_raw_by_canonical=remote_raw_by_canonical,
)
for uid in plan.school_events_to_remove:
self._do_delete(uid, result)
total = result.added + result.updated + result.removed
if result.errors:
result.status = CalDAVSyncStatus.FAILED
elif total == 0:
result.status = CalDAVSyncStatus.SKIPPED
else:
result.status = CalDAVSyncStatus.SUCCESS
return result
def _do_save(
self,
model: Lesson | Homework | SchoolEvent,
result: CalDAVSyncResult,
is_update: bool,
remote_raw_by_canonical: Mapping[str, str] | None = None,
) -> None:
"""Écrit un événement sur la passerelle, ou simule l'écriture.
En mode ``dry_run``, l'action est uniquement journalisée et le
compteur correspondant est incrémenté. Sinon, le modèle est
sérialisé en document iCalendar complet (``VCALENDAR``) via
:func:`model_to_vcalendar_text`, puis envoyé à la passerelle avec
l'UID dérivé via :func:`_model_uid` (l'upsert par UID permet la
création ou la mise à jour de l'événement) ; en cas d'erreur
``PronoteSyncError``, le message expurgé est ajouté à
``result.errors``. Pour une mise à jour, l'UID cible est l'UID brut
distant (via ``remote_raw_by_canonical``) afin de mettre à jour le
vrai événement distant au lieu d'en créer un doublon ; si le mapping
est absent, l'erreur est consignée dans ``result.errors`` sans
interrompre le lot.
:param model: Modèle Pronote à écrire (Lesson, Homework ou SchoolEvent).
:param result: Résultat à mettre à jour (compteurs et erreurs).
:param is_update: Si ``True``, l'opération est une mise à jour,
sinon un ajout.
:param remote_raw_by_canonical: Mapping canonical_uid -> raw_uid des
événements distants gérés, utilisé uniquement pour les mises à jour.
"""
action = "mise à jour" if is_update else "ajout"
uid = _model_uid(model)
if self._dry_run:
logger.info("DRY-RUN: %s de l'événement UID=%s", action, redact_secrets(uid))
if is_update:
result.updated += 1
else:
result.added += 1
return
try:
vcalendar_text = model_to_vcalendar_text(model)
target_uid = uid
if is_update:
if remote_raw_by_canonical is None or uid not in remote_raw_by_canonical:
result.errors.append(
f"UID canonique sans correspondant distant : {redact_secrets(uid)}"
)
return
target_uid = remote_raw_by_canonical[uid]
self._gateway.upsert_event(vcalendar_text, target_uid)
except PronoteSyncError as exc:
result.errors.append(redact_secrets(str(exc)))
return
logger.info("%s de l'événement UID=%s", action, redact_secrets(uid))
if is_update:
result.updated += 1
else:
result.added += 1
def _do_delete(self, uid: str, result: CalDAVSyncResult) -> None:
"""Supprime un événement de la passerelle, ou simule la suppression.
En mode ``dry_run``, l'action est uniquement journalisée et le
compteur des suppressions est incrémenté. Sinon, la passerelle est
appelée avec l'UID ; en cas d'erreur ``PronoteSyncError``, le
message expurgé est ajouté à ``result.errors``.
:param uid: Identifiant UID de l'événement à supprimer.
:param result: Résultat à mettre à jour (compteurs et erreurs).
"""
if self._dry_run:
logger.info("DRY-RUN: suppression de l'événement UID=%s", redact_secrets(uid))
result.removed += 1
return
try:
self._gateway.delete_event(uid)
except PronoteSyncError as exc:
result.errors.append(redact_secrets(str(exc)))
return
logger.info("suppression de l'événement UID=%s", redact_secrets(uid))
result.removed += 1

View File

@@ -0,0 +1,130 @@
"""Planification de la synchronisation CalDAV.
Ce module compare les données Pronote normalisées aux événements distants
marqués comme gérés par ``pronote-sync`` et produit un plan de synchronisation
CalDAV (ajouts, mises à jour, suppressions) pour chaque catégorie d'événement :
cours, devoirs et événements scolaires.
Le plan est calculé de manière pure et déterministe : deux entrées identiques
produisent un plan identique, et un événement dont la signature sémantique
n'a pas changé n'apparaît dans aucune liste du plan (idempotence).
"""
from __future__ import annotations
from typing import Any
from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.sync import CalDAVSyncPlan
from pronote_sync.sync.serialization import (
component_to_signature,
homework_to_vevent,
lesson_to_vevent,
school_event_to_vevent,
)
def compute_plan(
pronote_data: PronoteData,
remote_managed: list[tuple[str, str, Any]],
) -> tuple[CalDAVSyncPlan, dict[str, str]]:
"""Calcule le plan de synchronisation CalDAV et le mapping des UID distants.
L'appariement entre les événements locaux et distants se fait sur l'UID
canonique (forme normalisée, identique pour une même source Pronote,
suffixe temporel retiré) tandis que les mutations (suppressions, mises à
jour) ciblent l'UID brut tel que stocké sur le serveur. Le mapping
``canonical_uid -> raw_uid`` retourné permet à l'exécuteur de cibler le
bon objet distant lors des mises à jour.
:param pronote_data: Données Pronote normalisées (cours, devoirs, événements).
:param remote_managed: Liste de tuples (raw_uid, canonical_uid, vevent)
pour les événements distants marqués comme gérés par pronote-sync.
:return: Tuple (plan de synchronisation, mapping canonical_uid -> raw_uid).
Les listes ``*_to_remove`` contiennent l'UID brut distant, les autres
listes contiennent les modèles Pronote locaux.
:rtype: tuple[CalDAVSyncPlan, dict[str, str]]
"""
#: canonical_uid -> signature sémantique du VEVENT distant (pour l'appariement).
remote_signatures_by_canonical: dict[str, str] = {}
#: canonical_uid -> UID brut distant (pour cibler le bon objet lors des mutations).
remote_raw_by_canonical: dict[str, str] = {}
for raw_uid, canonical_uid, vevent in remote_managed:
remote_signatures_by_canonical[canonical_uid] = component_to_signature(vevent)
remote_raw_by_canonical[canonical_uid] = raw_uid
lessons_to_add: list[Lesson] = []
lessons_to_update: list[Lesson] = []
lessons_to_remove: list[str] = []
local_lessons_by_uid: dict[str, Lesson] = {lesson.id: lesson for lesson in pronote_data.lessons}
for lesson in pronote_data.lessons:
local_canonical = lesson.id
local_sig = component_to_signature(lesson_to_vevent(lesson))
if local_canonical not in remote_signatures_by_canonical:
lessons_to_add.append(lesson)
elif remote_signatures_by_canonical[local_canonical] != local_sig:
lessons_to_update.append(lesson)
homeworks_to_add: list[Homework] = []
homeworks_to_update: list[Homework] = []
homeworks_to_remove: list[str] = []
local_homeworks_by_uid: dict[str, Homework] = {
f"homework-{homework.id}": homework for homework in pronote_data.homeworks
}
for homework in pronote_data.homeworks:
local_canonical = f"homework-{homework.id}"
local_sig = component_to_signature(homework_to_vevent(homework))
if local_canonical not in remote_signatures_by_canonical:
homeworks_to_add.append(homework)
elif remote_signatures_by_canonical[local_canonical] != local_sig:
homeworks_to_update.append(homework)
school_events_to_add: list[SchoolEvent] = []
school_events_to_update: list[SchoolEvent] = []
school_events_to_remove: list[str] = []
local_school_events_by_uid: dict[str, SchoolEvent] = {
f"school-event-{event.label}-{event.from_date.isoformat()}": event
for event in pronote_data.school_events
}
for school_event in pronote_data.school_events:
local_canonical = f"school-event-{school_event.label}-{school_event.from_date.isoformat()}"
local_sig = component_to_signature(school_event_to_vevent(school_event))
if local_canonical not in remote_signatures_by_canonical:
school_events_to_add.append(school_event)
elif remote_signatures_by_canonical[local_canonical] != local_sig:
school_events_to_update.append(school_event)
# Détection des événements distants orphelins : un UID canonique distant
# absent des données locales est supprimé en ciblant l'UID brut stocké sur
# le serveur. L'acheminement vers la bonne liste de suppression se fait sur
# le préfixe de l'UID canonique.
for canonical_uid in remote_signatures_by_canonical:
raw_uid = remote_raw_by_canonical[canonical_uid]
if canonical_uid.startswith("homework-"):
if canonical_uid not in local_homeworks_by_uid:
homeworks_to_remove.append(raw_uid)
elif canonical_uid.startswith("school-event-"):
if canonical_uid not in local_school_events_by_uid:
school_events_to_remove.append(raw_uid)
elif canonical_uid not in local_lessons_by_uid:
lessons_to_remove.append(raw_uid)
return (
CalDAVSyncPlan(
lessons_to_add=lessons_to_add,
lessons_to_update=lessons_to_update,
lessons_to_remove=lessons_to_remove,
homeworks_to_add=homeworks_to_add,
homeworks_to_update=homeworks_to_update,
homeworks_to_remove=homeworks_to_remove,
school_events_to_add=school_events_to_add,
school_events_to_update=school_events_to_update,
school_events_to_remove=school_events_to_remove,
),
remote_raw_by_canonical,
)

View File

@@ -0,0 +1,196 @@
"""Sérialisation des modèles Pronote en composants iCalendar (VEVENT).
Ce module convertit les modèles métier (:class:`Lesson`, :class:`Homework`,
:class:`SchoolEvent`) en composants :class:`icalendar.Event` destinés à la
synchronisation CalDAV, fournit un enveloppement en document ``VCALENDAR``
complet (avec ``VERSION`` et ``PRODID``), et extrait une signature sémantique
déterministe d'un composant distant pour permettre une comparaison
idempotente.
Les composants produits portent le marqueur :data:`MANAGED_PROPERTY` avec la
valeur :data:`MANAGED_VALUE` afin d'identifier les événements gérés par
l'outil et de ne jamais toucher aux événements étrangers du calendrier.
"""
from __future__ import annotations
from datetime import datetime, time
from typing import cast
from icalendar import Calendar, Component, Event, vDate, vDatetime
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.utils.uid import normalize_datetime_to_utc
#: Propriété iCalendar marquant un événement géré par ``pronote-sync``.
MANAGED_PROPERTY = "X-PRONOTE-SYNC-MANAGED"
#: Valeur du marqueur de gestion (version du format de signature).
MANAGED_VALUE = "v1"
#: Identifiant du produit pour la propriété ``PRODID`` des documents CalDAV.
PRODID = "-//pronote-sync//NONSGML v1.0//EN"
#: Propriétés prises en compte dans la signature sémantique d'un composant.
_SIGNATURE_KEYS: tuple[str, ...] = ("UID", "SUMMARY", "DTSTART", "DTEND", "STATUS", "DESCRIPTION")
def lesson_to_vevent(lesson: Lesson) -> Event:
"""Convertit un cours Pronote en composant VEVENT iCalendar.
La description contient une ligne par champ renseigné (matière,
professeur(s), salle(s), contenu). Un cours annulé est marqué
``STATUS:CANCELLED`` et classé dans la catégorie « Annulé », un cours
déplacé dans la catégorie « Déplacé ».
:param lesson: Cours Pronote à sérialiser.
:return: Composant :class:`icalendar.Event` marqué comme géré par l'outil.
:rtype: icalendar.Event
"""
event = Event()
event.add("uid", lesson.id)
event.add("summary", lesson.subject)
event.add("dtstart", vDatetime(lesson.start))
event.add("dtend", vDatetime(lesson.end))
parts: list[str] = []
if lesson.subject:
parts.append(f"Matière: {lesson.subject}")
if lesson.teachers:
parts.append(f"Professeur(s): {', '.join(lesson.teachers)}")
if lesson.rooms:
parts.append(f"Salle(s): {', '.join(lesson.rooms)}")
if lesson.content:
parts.append(f"Contenu: {lesson.content}")
event.add("description", "\n".join(parts))
if lesson.status == LessonStatus.CANCELLED:
event.add("status", "CANCELLED")
else:
event.add("status", "CONFIRMED")
categories = ["Pronote"]
if lesson.status == LessonStatus.CANCELLED:
categories.append("Annulé")
elif lesson.status == LessonStatus.MOVED:
categories.append("Déplacé")
event.add("categories", categories)
event.add(MANAGED_PROPERTY, MANAGED_VALUE)
return event
def homework_to_vevent(homework: Homework) -> Event:
"""Convertit un devoir Pronote en composant VEVENT iCalendar.
Le devoir est représenté comme une tâche (``STATUS:NEEDS-ACTION``) sur la
journée d'échéance, entre 08:00 et 18:00.
:param homework: Devoir Pronote à sérialiser.
:return: Composant :class:`icalendar.Event` marqué comme géré par l'outil.
:rtype: icalendar.Event
"""
event = Event()
event.add("uid", f"homework-{homework.id}")
event.add("summary", f"Devoir: {homework.subject}")
event.add("dtstart", vDatetime(datetime.combine(homework.due_on, time(8, 0))))
event.add("dtend", vDatetime(datetime.combine(homework.due_on, time(18, 0))))
event.add("description", homework.text)
event.add("status", "NEEDS-ACTION")
event.add("categories", ["Pronote", "Devoir"])
event.add(MANAGED_PROPERTY, MANAGED_VALUE)
return event
def school_event_to_vevent(school_event: SchoolEvent) -> Event:
"""Convertit un événement scolaire en composant VEVENT iCalendar.
:param school_event: Événement scolaire (vacances, jour férié) à sérialiser.
:return: Composant :class:`icalendar.Event` marqué comme géré par l'outil.
:rtype: icalendar.Event
"""
event = Event()
event.add("uid", f"school-event-{school_event.label}-{school_event.from_date.isoformat()}")
event.add("summary", school_event.label)
event.add("dtstart", vDate(school_event.from_date))
event.add("dtend", vDate(school_event.to_date))
event.add("status", "CONFIRMED")
event.add("categories", ["Pronote", school_event.kind.value])
event.add(MANAGED_PROPERTY, MANAGED_VALUE)
return event
def model_to_vcalendar_text(model: Lesson | Homework | SchoolEvent) -> str:
"""Sérialise un modèle Pronote en document iCalendar complet (VCALENDAR).
Produit un document ``VCALENDAR`` valide contenant un seul ``VEVENT``,
avec les propriétés ``VERSION:2.0`` et ``PRODID`` requises par le protocole
CalDAV.
:param model: Modèle Pronote à sérialiser (Lesson, Homework ou SchoolEvent).
:return: Document iCalendar complet en texte.
:rtype: str
:raises ValueError: Si le type de modèle n'est pas supporté.
"""
if isinstance(model, Lesson):
vevent = lesson_to_vevent(model)
elif isinstance(model, Homework):
vevent = homework_to_vevent(model)
elif isinstance(model, SchoolEvent):
vevent = school_event_to_vevent(model)
else:
raise ValueError(f"Type de modèle non supporté : {type(model).__name__}")
cal = Calendar()
cal.add("prodid", PRODID)
cal.add("version", "2.0")
cal.add_component(vevent)
# ``to_ical()`` n'est pas typé dans icalendar : le cast documente le
# décodage UTF-8 en texte et satisfait mypy strict.
return cast(str, cal.to_ical().decode("utf-8"))
def component_to_signature(component: Component) -> str:
"""Extrait une signature sémantique déterministe d'un composant iCalendar.
La signature couvre l'UID, le résumé, les dates de début et de fin, le
statut, la description, les catégories (triées) et le marqueur de gestion.
Les propriétés volatiles (``DTSTAMP``, ``CREATED``, ``LAST-MODIFIED``,
``SEQUENCE``) sont volontairement exclues : elles changent à chaque
écriture serveur et ne reflètent aucun changement des données Pronote.
Les valeurs textuelles sont normalisées (espaces rognés, minuscules) et
les dates/heures sérialisées via ``isoformat()``, de sorte que deux
composants au contenu sémantiquement identique produisent la même
signature.
:param component: Composant iCalendar (généralement un VEVENT distant).
:return: Paires ``clé=valeur`` triées et jointes par ``|``.
:rtype: str
"""
props: list[str] = []
for key in _SIGNATURE_KEYS:
raw = component.get(key)
if raw is None:
continue
value = getattr(raw, "dt", raw)
if hasattr(value, "isoformat"):
if isinstance(value, datetime):
rendered = normalize_datetime_to_utc(value).isoformat()
else:
rendered = value.isoformat()
else:
rendered = str(value).strip().lower()
props.append(f"{key.lower()}={rendered}")
categories = component.get("CATEGORIES")
if categories is not None:
cats = sorted(str(c).strip().lower() for c in categories.cats)
props.append(f"categories={','.join(cats)}")
managed = component.get(MANAGED_PROPERTY)
if managed is not None:
props.append(f"managed={str(managed).strip().lower()}")
return "|".join(sorted(props))

View File

@@ -0,0 +1,148 @@
"""Orchestrateur de la synchronisation CalDAV.
Ce module fournit :func:`synchronize`, le point d'entrée haut niveau qui
enchaîne les trois phases de la synchronisation : connexion à la passerelle
CalDAV, scan des événements distants gérés et calcul du plan, puis exécution
du plan (ou simulation en mode ``dry_run``). Il s'appuie sur
:class:`~pronote_sync.sync.caldav.CalDAVGateway`,
:func:`~pronote_sync.sync.planner.compute_plan` et
:class:`~pronote_sync.sync.executor.CalDAVSyncExecutor`.
"""
from __future__ import annotations
import logging
from collections.abc import Callable
from datetime import datetime, time, timedelta
from typing import Any
from pronote_sync.config.settings import Settings
from pronote_sync.errors import PronoteSyncError
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
from pronote_sync.sync.caldav import CalDAVGateway
from pronote_sync.sync.executor import CalDAVSyncExecutor
from pronote_sync.sync.planner import compute_plan
logger = logging.getLogger(__name__)
def synchronize(
pronote_data: PronoteData,
settings: Settings,
client_factory: Callable[..., Any] | None = None,
*,
now: datetime | None = None,
) -> CalDAVSyncResult:
"""Synchronise les données Pronote vers le calendrier CalDAV.
Enchaîne les trois phases : connexion à la passerelle, scan distant et
calcul du plan, puis exécution (ou simulation dry-run).
La fenêtre de synchronisation est calculée en **journées complètes** :
elle commence à minuit de ``aujourd'hui - sync_past_days`` (inclusive) et
se termine, de façon exclusive, à minuit de ``aujourd'hui +
sync_future_days + 1``, afin que le dernier jour de la fenêtre soit couvert
en entier. Cette fenêtre s'applique aux **deux** côtés de la
synchronisation : les événements distants gérés scannés sur la passerelle
et les données Pronote locales filtrées (cours filtrés sur ``start``,
devoirs sur ``due_on`` en jour, événements scolaires sur le chevauchement
de leur période) avant d'être transmises au planificateur. Aucune
abstraction d'horloge n'existe encore dans le dépôt (les données Pronote
sont par convention naïves en heure locale) : ``datetime.now()`` est
l'instant de référence par défaut, remplaçable via ``now`` pour les tests.
:param pronote_data: Données Pronote normalisées à synchroniser.
:param settings: Configuration racine du pipeline.
:param client_factory: Fabrique optionnelle de client DAV (pour les tests).
:param now: Instant de référence pour le calcul de la fenêtre ; par défaut
``datetime.now()``.
:return: Résultat de la synchronisation (statut, compteurs, erreurs).
:rtype: CalDAVSyncResult
:raises PronoteSyncError: Si la configuration CalDAV est incomplète ou si la
connexion échoue.
"""
now = now or datetime.now()
window_start = datetime.combine(
(now - timedelta(days=settings.app.sync_past_days)).date(),
time(0, 0),
)
window_end = datetime.combine(
(now + timedelta(days=settings.app.sync_future_days + 1)).date(),
time(0, 0),
)
# Application de la fenêtre aux données Pronote locales avant le passage au
# planificateur : les événements hors fenêtre ne doivent être ni écrits sur
# le calendrier ni déclencher de suppression d'un événement distant géré.
filtered_lessons = [
lesson for lesson in pronote_data.lessons if window_start <= lesson.start < window_end
]
filtered_homeworks = [
hw for hw in pronote_data.homeworks if window_start.date() <= hw.due_on < window_end.date()
]
filtered_school_events = [
se
for se in pronote_data.school_events
if se.from_date < window_end.date() and se.to_date > window_start.date()
]
filtered_pronote = PronoteData(
lessons=filtered_lessons,
homeworks=filtered_homeworks,
school_events=filtered_school_events,
messages=pronote_data.messages,
target_date=pronote_data.target_date,
generated_at=pronote_data.generated_at,
)
if (
settings.caldav.url is None
or settings.caldav.username is None
or settings.caldav.password is None
):
logger.info("CalDAV non configuré — synchronisation ignorée")
return CalDAVSyncResult(status=CalDAVSyncStatus.SKIPPED, added=0, updated=0, removed=0)
gateway = CalDAVGateway(settings.caldav, client_factory=client_factory)
try:
with gateway:
remote_managed = gateway.list_managed_events(start=window_start, end=window_end)
logger.info(
"Synchronisation CalDAV : %d événements distants gérés trouvés",
len(remote_managed),
)
plan, remote_raw_by_canonical = compute_plan(filtered_pronote, remote_managed)
n_add = (
len(plan.lessons_to_add)
+ len(plan.homeworks_to_add)
+ len(plan.school_events_to_add)
)
n_update = (
len(plan.lessons_to_update)
+ len(plan.homeworks_to_update)
+ len(plan.school_events_to_update)
)
n_remove = (
len(plan.lessons_to_remove)
+ len(plan.homeworks_to_remove)
+ len(plan.school_events_to_remove)
)
logger.info(
"Plan : %d ajouts, %d mises à jour, %d suppressions",
n_add,
n_update,
n_remove,
)
if settings.app.dry_run:
logger.info("DRY-RUN : aucune écriture ne sera effectuée sur le calendrier")
executor = CalDAVSyncExecutor(gateway, dry_run=settings.app.dry_run)
return executor.execute(plan, remote_raw_by_canonical=remote_raw_by_canonical)
except PronoteSyncError:
logger.error("Échec de la synchronisation CalDAV")
# Re-lève la même exception de domaine sans en créer de nouvelle.
# ``PronoteSyncError`` a déjà été levée avec ``from None`` en amont
# (passerelle CalDAV), donc ``__cause__`` et ``__context__`` restent
# propres : un ``raise`` nu préserve cet état sans ajouter de chaînage.
raise

View File

@@ -0,0 +1,32 @@
"""Utilitaires de normalisation et de traitement du texte.
Ce module centralise les transformations de texte partagées par plusieurs
couches du pipeline ``pronote-sync`` (sources, synchronisation) afin que les
modules de logique de domaine ne dépendent pas d'adaptateurs concrets.
"""
from __future__ import annotations
import re
import unicodedata
__all__ = ["normalize_subject"]
def normalize_subject(subject: str) -> str:
"""Normalise une matière pour le matching déterministe.
Applique la normalisation Unicode NFKC, unifie les espaces (y compris
tabulations et espaces insécables), supprime la ponctuation et met la
chaîne en minuscules. Deux représentations visuellement identiques d'une
même matière produisent ainsi la même forme normalisée.
:param subject: La matière brute.
:return: La forme normalisée (NFKC, espaces unifiés, sans ponctuation, minuscule).
:rtype: str
"""
normalized = unicodedata.normalize("NFKC", subject)
normalized = re.sub(r"\s+", " ", normalized).strip()
normalized = re.sub(r"[^\w\s]", "", normalized)
normalized = re.sub(r"\s+", " ", normalized).strip()
return normalized.lower()

View File

@@ -11,6 +11,7 @@ from __future__ import annotations
import hashlib import hashlib
import re import re
from datetime import datetime from datetime import datetime
from zoneinfo import ZoneInfo
_TEMPORAL_SUFFIX_PATTERN = re.compile(r"-\d{8}T\d{6}Z-Index-Education$") _TEMPORAL_SUFFIX_PATTERN = re.compile(r"-\d{8}T\d{6}Z-Index-Education$")
_EDUCATION_SUFFIX_PATTERN = re.compile(r"-Index-Education$") _EDUCATION_SUFFIX_PATTERN = re.compile(r"-Index-Education$")
@@ -31,6 +32,21 @@ def normalize_pronote_uid(uid: str) -> str:
return _EDUCATION_SUFFIX_PATTERN.sub("", normalized) return _EDUCATION_SUFFIX_PATTERN.sub("", normalized)
def normalize_datetime_to_utc(dt: datetime) -> datetime:
"""Normalise une datetime vers UTC pour les signatures et hachages.
Les datetimes naïves sont interprétées comme Europe/Paris puis converties
vers UTC. Les datetimes conscientes sont converties vers UTC.
:param dt: Datetime à normaliser (naïve ou consciente).
:return: Datetime en UTC.
:rtype: datetime
"""
if dt.tzinfo is None:
return dt.replace(tzinfo=ZoneInfo("Europe/Paris")).astimezone(ZoneInfo("UTC"))
return dt.astimezone(ZoneInfo("UTC"))
def generate_deterministic_uid( def generate_deterministic_uid(
start: datetime, start: datetime,
end: datetime, end: datetime,
@@ -56,8 +72,8 @@ def generate_deterministic_uid(
:rtype: str :rtype: str
""" """
parts = [ parts = [
start.isoformat(), normalize_datetime_to_utc(start).isoformat(),
end.isoformat(), normalize_datetime_to_utc(end).isoformat(),
subject, subject,
",".join(sorted(teachers)), ",".join(sorted(teachers)),
",".join(sorted(rooms)), ",".join(sorted(rooms)),

View File

@@ -0,0 +1,99 @@
"""Fixtures partagées pour les tests de pronote-sync."""
from __future__ import annotations
from datetime import date, datetime
import pytest
from pydantic import SecretStr
from pronote_sync.config.settings import CalDAVSettings
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.models.homework import Homework
from pronote_sync.models.pronote import PronoteData
@pytest.fixture
def sample_lesson() -> Lesson:
"""Cours normal pour les tests."""
return Lesson(
id="L-1234-Normal",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
@pytest.fixture
def sample_cancelled_lesson() -> Lesson:
"""Cours annulé pour les tests."""
return Lesson(
id="L-5678-Cancelled",
start=datetime(2026, 1, 16, 10, 0),
end=datetime(2026, 1, 16, 11, 0),
subject="Français",
teachers=("Prof Martin",),
rooms=("Salle 202",),
status=LessonStatus.CANCELLED,
group=None,
content=None,
)
@pytest.fixture
def sample_homework() -> Homework:
"""Devoir pour les tests."""
return Homework(
id="hw-001",
subject="Histoire",
teachers=("Prof Bernard",),
assigned_on=date(2026, 1, 15),
due_on=date(2026, 1, 20),
text="Lire le chapitre 5",
html="<p>Lire le chapitre 5</p>",
)
@pytest.fixture
def sample_school_event() -> SchoolEvent:
"""Événement scolaire pour les tests."""
return SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Vacances de Noël",
from_date=date(2026, 12, 20),
to_date=date(2027, 1, 5),
)
@pytest.fixture
def caldav_settings() -> CalDAVSettings:
"""Configuration CalDAV de test avec une URL HTTPS."""
return CalDAVSettings(
url=SecretStr("https://caldav.example.com/remote.php/dav/"),
username="test-user",
password=SecretStr("test-secret-password-12345"),
calendar_path="/pronote-sync/",
)
@pytest.fixture
def pronote_data(
sample_lesson: Lesson,
sample_cancelled_lesson: Lesson,
sample_homework: Homework,
sample_school_event: SchoolEvent,
) -> PronoteData:
"""Données Pronote de test avec des cours, devoirs et événements."""
return PronoteData(
lessons=[sample_lesson, sample_cancelled_lesson],
homeworks=[sample_homework],
school_events=[sample_school_event],
messages=[],
target_date=date(2026, 1, 15),
generated_at=datetime(2026, 1, 15, 0, 0),
)

View File

View File

@@ -0,0 +1,929 @@
"""Tests d'intégration pour la synchronisation CalDAV.
Ce module teste le flux complet de synchronisation avec un faux serveur
CalDAV en mémoire, en vérifiant l'idempotence, le mode dry-run, les
ajouts/mises à jour/suppressions, et la préservation des événements non gérés.
"""
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING, Any
import pytest
from caldav.lib.error import NotFoundError
from icalendar import Calendar, Event
from pydantic import SecretStr
from pronote_sync.config.settings import AppSettings, CalDAVSettings, Settings
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.models.homework import Homework
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.sync import CalDAVSyncStatus
from pronote_sync.sync.synchronizer import synchronize
if TYPE_CHECKING:
pass
class FakeCalendarEvent:
"""Faux événement calendrier pour les tests.
Reproduit le contrat minimal de ``CalendarObjectResource`` utilisé par
la passerelle : contenu brut accessible via ``data`` (lecture/écriture),
enregistrement via ``save()``, composant iCalendar et suppression.
"""
def __init__(
self, ical_text: str, uid: str | None = None, server: FakeCalDAVServer | None = None
):
self._ical_text = ical_text
self._uid = uid
self._server = server
@property
def data(self) -> str:
"""Retourne le contenu iCalendar brut de l'événement.
:return: Texte iCalendar de l'événement.
:rtype: str
"""
return self._ical_text
@data.setter
def data(self, value: str) -> None:
"""Remplace le contenu iCalendar brut de l'événement.
:param value: Nouveau texte iCalendar de l'événement.
"""
self._ical_text = value
def save(self) -> None:
"""Enregistre le contenu courant de l'événement sur le serveur."""
if self._server and self._uid:
self._server._events[self._uid] = self._ical_text
@property
def icalendar_component(self) -> Calendar:
"""Retourne le composant iCalendar de l'événement.
:return: Composant iCalendar parsé.
:rtype: Calendar
"""
return Calendar.from_ical(self._ical_text)
def delete(self) -> None:
"""Supprime l'événement du serveur."""
if self._server and self._uid:
self._server._events.pop(self._uid, None)
class FakeCalendar:
"""Faux calendrier CalDAV pour les tests."""
def __init__(self, server: FakeCalDAVServer):
self._server = server
self.url = "https://caldav.example.com/remote.php/dav/calendars/test-user/pronote-sync/"
def search(
self,
*,
start: datetime | None = None,
end: datetime | None = None,
event: bool = True,
expand: bool = True,
**kwargs: Any,
) -> list[FakeCalendarEvent]:
"""Retourne les événements dont la période chevauche la fenêtre donnée.
Reproduit le comportement d'un vrai serveur CalDAV : seuls les
événements dont la période chevauche la fenêtre ``[start, end)`` sont
retournés, comme de faux objets calendrier.
:param start: Début de la fenêtre de recherche (``None`` : pas de
borne inférieure).
:param end: Fin de la fenêtre de recherche (``None`` : pas de borne
supérieure).
:param event: Non utilisé (le faux ne traite que des VEVENT).
:param expand: Non utilisé (compatibilité avec la passerelle).
:param kwargs: Paramètres additionnels ignorés (compatibilité avec
l'API ``caldav``).
:return: Liste des événements chevauchant la fenêtre.
:rtype: list[FakeCalendarEvent]
"""
results: list[FakeCalendarEvent] = []
for uid, ical_text in self._server._events.items():
cal = Calendar.from_ical(ical_text)
for component in cal.walk("VEVENT"):
raw_start = component.get("dtstart")
raw_end = component.get("dtend")
if raw_start is None:
continue
event_start = raw_start.dt
event_end = raw_end.dt if raw_end is not None else event_start
overlaps = True
if start is not None:
overlaps = overlaps and event_end > start
if end is not None:
overlaps = overlaps and event_start < end
if overlaps:
results.append(FakeCalendarEvent(ical_text, uid=uid, server=self._server))
break
return results
def get_event_by_uid(self, uid: str) -> FakeCalendarEvent:
"""Retourne un événement par son UID, ou lève NotFoundError s'il est absent."""
if uid in self._server._events:
return FakeCalendarEvent(self._server._events[uid], uid=uid, server=self._server)
raise NotFoundError(f"Event {uid} not found")
def add_event(self, *, ical: str) -> None:
"""Ajoute un événement en extrayant l'UID du texte iCalendar."""
cal = Calendar.from_ical(ical)
for component in cal.walk("VEVENT"):
uid = str(component.get("UID"))
self._server._events[uid] = ical
@property
def icalendar_component(self) -> Calendar:
"""Propriété non utilisée pour le calendrier lui-même."""
raise NotImplementedError
class FakePrincipal:
"""Faux principal CalDAV pour les tests."""
def __init__(self, server: FakeCalDAVServer):
self._server = server
def calendars(self) -> list[FakeCalendar]:
"""Retourne les calendriers du principal.
:return: Liste contenant le faux calendrier cible.
:rtype: list[FakeCalendar]
"""
return [FakeCalendar(self._server)]
class FakeDAVClient:
"""Faux client DAV pour les tests."""
def __init__(self, server: FakeCalDAVServer):
self._server = server
def principal(self) -> FakePrincipal:
"""Retourne un faux principal CalDAV pour la découverte."""
return FakePrincipal(self._server)
class FakeCalDAVServer:
"""Faux serveur CalDAV en mémoire pour les tests d'intégration."""
def __init__(self) -> None:
self._events: dict[str, str] = {} # uid -> ical text
def client_factory(self, **kwargs: Any) -> FakeDAVClient:
"""Retourne un faux client DAV."""
return FakeDAVClient(self)
def get_events(self) -> dict[str, str]:
"""Retourne tous les événements stockés."""
return self._events.copy()
def clear(self) -> None:
"""Efface tous les événements."""
self._events.clear()
@pytest.fixture
def fake_caldav_server() -> FakeCalDAVServer:
"""Fournit un faux serveur CalDAV vide pour les tests."""
return FakeCalDAVServer()
@pytest.fixture
def caldav_settings() -> CalDAVSettings:
"""Configuration CalDAV de test."""
return CalDAVSettings(
url=SecretStr("https://caldav.example.com/remote.php/dav/"),
username="test-user",
password=SecretStr("test-password"),
calendar_path="/pronote-sync/",
)
@pytest.fixture
def app_settings() -> AppSettings:
"""Paramètres d'application de test."""
return AppSettings(
dry_run=False,
log_level="INFO",
sync_past_days=7,
sync_future_days=30,
)
@pytest.fixture
def full_settings(caldav_settings: CalDAVSettings, app_settings: AppSettings) -> Settings:
"""Configuration complète de test."""
return Settings(
caldav=caldav_settings,
app=app_settings,
)
def _create_vevent_text(
uid: str,
summary: str,
start: datetime,
end: datetime,
status: str = "CONFIRMED",
managed: bool = True,
) -> str:
"""Crée un texte iCalendar pour un VEVENT.
:param uid: Identifiant unique de l'événement.
:param summary: Résumé de l'événement.
:param start: Date/heure de début.
:param end: Date/heure de fin.
:param status: Statut de l'événement.
:param managed: Si True, ajoute le marqueur de gestion.
:return: Texte iCalendar de l'événement.
:rtype: str
"""
cal = Calendar()
cal.add("prodid", "-//pronote-sync//test//FR")
cal.add("version", "2.0")
event = Event()
event.add("uid", uid)
event.add("summary", summary)
event.add("dtstart", start)
event.add("dtend", end)
event.add("status", status)
if managed:
event.add("X-PRONOTE-SYNC-MANAGED", "v1")
cal.add_component(event)
return cal.to_ical().decode("utf-8") # type: ignore[no-any-return]
class TestCalDAVSynchronize:
"""Tests d'intégration pour la synchronisation CalDAV."""
def test_full_sync_add(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste l'ajout d'un cours lors d'une synchronisation complète."""
# Données Pronote avec un cours
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 1
assert result.updated == 0
assert result.removed == 0
assert len(fake_caldav_server.get_events()) == 1
def test_full_sync_idempotent(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste l'idempotence : deux synchronisations identiques."""
# Données Pronote avec un cours
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Première synchronisation
result1 = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert result1.added == 1
# Deuxième synchronisation avec les mêmes données
result2 = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert result2.status == CalDAVSyncStatus.SKIPPED
assert result2.added == 0
assert result2.updated == 0
assert result2.removed == 0
def test_full_sync_dry_run(
self,
fake_caldav_server: FakeCalDAVServer,
caldav_settings: CalDAVSettings,
) -> None:
"""Teste le mode dry-run : aucune écriture sur le serveur."""
app_settings = AppSettings(
dry_run=True,
log_level="INFO",
sync_past_days=7,
sync_future_days=30,
)
full_settings = Settings(
caldav=caldav_settings,
app=app_settings,
)
# Données Pronote avec un cours
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation en mode dry-run
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications : le résultat indique un ajout, mais le serveur reste vide
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 0
def test_full_sync_update(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste la mise à jour d'un cours existant."""
# Pré-remplir le serveur avec un événement existant
existing_event = _create_vevent_text(
uid="L-001",
summary="Mathématiques",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
managed=True,
)
fake_caldav_server._events["L-001"] = existing_event
# Données Pronote avec le même cours mais avec un sujet modifié
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 0
assert result.updated == 1
assert result.removed == 0
def test_full_sync_remove(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste la suppression d'un événement distant non présent localement."""
# Pré-remplir le serveur avec un événement existant
existing_event = _create_vevent_text(
uid="L-001",
summary="Mathématiques",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
managed=True,
)
fake_caldav_server._events["L-001"] = existing_event
# Données Pronote sans ce cours
pronote_data = PronoteData(
lessons=[],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 0
assert result.updated == 0
assert result.removed == 1
assert len(fake_caldav_server.get_events()) == 0
def test_cancelled_lesson_preserved(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste qu'un cours annulé est synchronisé avec le statut CANCELLED."""
# Données Pronote avec un cours annulé
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.CANCELLED,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 1
# Vérifie que l'événement a le statut CANCELLED
event_text = list(fake_caldav_server.get_events().values())[0]
assert "STATUS:CANCELLED" in event_text
def test_unmanaged_event_untouched(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste qu'un événement non géré n'est pas supprimé."""
# Pré-remplir le serveur avec un événement NON géré
unmanaged_event = _create_vevent_text(
uid="UNMANAGED-001",
summary="Événement personnel",
start=datetime(2026, 1, 15, 10, 0),
end=datetime(2026, 1, 15, 11, 0),
managed=False, # Non géré
)
fake_caldav_server._events["UNMANAGED-001"] = unmanaged_event
# Données Pronote sans ce cours
pronote_data = PronoteData(
lessons=[],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications : l'événement non géré est toujours présent
assert result.status == CalDAVSyncStatus.SKIPPED
assert len(fake_caldav_server.get_events()) == 1
assert "UNMANAGED-001" in fake_caldav_server.get_events()
def test_caldav_not_configured(
self,
fake_caldav_server: FakeCalDAVServer,
) -> None:
"""Teste que la synchronisation est ignorée si CalDAV n'est pas configuré."""
# Configuration sans CalDAV
full_settings = Settings(
caldav=CalDAVSettings(
url=None,
username=None,
password=None,
calendar_path="/pronote-sync/",
),
app=AppSettings(
dry_run=False,
log_level="INFO",
sync_past_days=7,
sync_future_days=30,
),
)
# Données Pronote
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications
assert result.status == CalDAVSyncStatus.SKIPPED
assert result.added == 0
assert result.updated == 0
assert result.removed == 0
def test_homework_sync(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste la synchronisation des devoirs."""
# Données Pronote avec un devoir
pronote_data = PronoteData(
lessons=[],
homeworks=[
Homework(
id="hw-001",
subject="Histoire",
teachers=("Prof Bernard",),
assigned_on=datetime(2026, 1, 15).date(),
due_on=datetime(2026, 1, 20).date(),
text="Lire le chapitre 5",
html="<p>Lire le chapitre 5</p>",
)
],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 1
def test_school_event_sync(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste la synchronisation des événements scolaires."""
# Données Pronote avec un événement scolaire
pronote_data = PronoteData(
lessons=[],
homeworks=[],
school_events=[
SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Vacances de Noël",
from_date=datetime(2026, 12, 20).date(),
to_date=datetime(2027, 1, 5).date(),
)
],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 12, 20, 12, 0),
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 1
def test_multiple_operations(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste une synchronisation avec plusieurs opérations (ajout, mise à jour, suppression)."""
# Pré-remplir le serveur avec des événements existants
existing_lesson = _create_vevent_text(
uid="L-001",
summary="Mathématiques",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
managed=True,
)
existing_homework = _create_vevent_text(
uid="homework-hw-001",
summary="Devoir: Histoire",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 18, 0),
managed=True,
)
existing_unmanaged = _create_vevent_text(
uid="UNMANAGED-001",
summary="Événement personnel",
start=datetime(2026, 1, 15, 10, 0),
end=datetime(2026, 1, 15, 11, 0),
managed=False,
)
fake_caldav_server._events["L-001"] = existing_lesson
fake_caldav_server._events["homework-hw-001"] = existing_homework
fake_caldav_server._events["UNMANAGED-001"] = existing_unmanaged
# Données Pronote :
# - L-001 : modifié (mise à jour)
# - L-002 : nouveau (ajout)
# - hw-001 : supprimé (suppression)
# - hw-002 : nouveau (ajout)
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Physique", # Modifié
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
),
Lesson(
id="L-002",
start=datetime(2026, 1, 16, 8, 0),
end=datetime(2026, 1, 16, 9, 0),
subject="Français",
teachers=("Prof Martin",),
rooms=("Salle 202",),
status=LessonStatus.NORMAL,
group=None,
content=None,
),
],
homeworks=[
Homework(
id="hw-002",
subject="Mathématiques",
teachers=("Prof Dupont",),
assigned_on=datetime(2026, 1, 15).date(),
due_on=datetime(2026, 1, 20).date(),
text="Exercices page 45",
html="<p>Exercices page 45</p>",
)
],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 2 # L-002 et hw-002
assert result.updated == 1 # L-001
assert result.removed == 1 # hw-001
# Vérifie que l'événement non géré est toujours présent
events = fake_caldav_server.get_events()
assert "UNMANAGED-001" in events
assert len(events) == 4 # L-001 (mis à jour), L-002, hw-002, UNMANAGED-001
def test_sync_filters_out_of_window_events(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste qu'un cours hors fenêtre n'est pas écrit sur le calendrier.
Avec ``now=2026-01-14`` et la configuration par défaut (7 jours dans
le passé, 30 dans le futur), la fenêtre couvre les journées complètes
du 2026-01-07 au 2026-02-14 (fin exclusive) : un cours le 2026-02-15
doit être filtré et ne jamais atteindre le planificateur ni le
calendrier.
"""
# Données Pronote avec un cours hors de la fenêtre
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 2, 15, 8, 0),
end=datetime(2026, 2, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 2, 15).date(),
generated_at=datetime(2026, 2, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
# Vérifications : le cours hors fenêtre n'est jamais écrit
assert result.status == CalDAVSyncStatus.SKIPPED
assert result.added == 0
assert result.updated == 0
assert result.removed == 0
assert len(fake_caldav_server.get_events()) == 0
def test_sync_idempotent_at_window_edge(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste l'idempotence pour un cours tardif en fin de fenêtre.
Un cours à 23:00 le dernier jour complet de la fenêtre (2026-02-13 ;
la fenêtre se termine de façon exclusive le 2026-02-14) doit être
ajouté une seule fois : une seconde exécution identique doit le
détecter comme déjà présent et ne rien réécrire.
"""
# Données Pronote avec un cours tardif le dernier jour de la fenêtre
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 2, 13, 23, 0),
end=datetime(2026, 2, 13, 23, 30),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 2, 13).date(),
generated_at=datetime(2026, 2, 13, 0, 0),
)
# Première synchronisation : le cours est ajouté
result1 = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert result1.status == CalDAVSyncStatus.SUCCESS
assert result1.added == 1
assert len(fake_caldav_server.get_events()) == 1
# Seconde synchronisation identique : rien n'est réécrit
result2 = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert result2.status == CalDAVSyncStatus.SKIPPED
assert result2.added == 0
assert result2.updated == 0
assert result2.removed == 0
assert len(fake_caldav_server.get_events()) == 1

View File

@@ -0,0 +1,682 @@
"""Tests unitaires pour l'exécuteur de synchronisation CalDAV.
Ce module vérifie que :class:`CalDAVSyncExecutor` applique correctement
un plan de synchronisation contre une passerelle CalDAV mockée, avec
la bonne gestion du mode dry-run, des compteurs et des erreurs.
"""
from __future__ import annotations
from datetime import date, datetime
from typing import TYPE_CHECKING
from unittest.mock import MagicMock
import pytest
from pronote_sync.errors import PronoteSyncError
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.models.homework import Homework
from pronote_sync.models.sync import CalDAVSyncPlan, CalDAVSyncStatus
from pronote_sync.sync.executor import CalDAVSyncExecutor
if TYPE_CHECKING:
pass
# --- Helper fixtures ---
def _make_lesson(
lesson_id: str = "L-1234",
subject: str = "Mathématiques",
start: datetime | None = None,
end: datetime | None = None,
status: LessonStatus = LessonStatus.NORMAL,
content: str | None = None,
group: str | None = None,
) -> Lesson:
"""Fabrique un cours Pronote pour les tests.
:param lesson_id: Identifiant du cours.
:param subject: Matière.
:param start: Date/heure de début.
:param end: Date/heure de fin.
:param status: Statut du cours.
:param content: Contenu pédagogique.
:param group: Groupe.
:return: Instance de Lesson.
:rtype: Lesson
"""
if start is None:
start = datetime(2026, 1, 15, 8, 0)
if end is None:
end = datetime(2026, 1, 15, 9, 0)
return Lesson(
id=lesson_id,
start=start,
end=end,
subject=subject,
status=status,
content=content,
group=group,
)
def _make_homework(
homework_id: str = "HW-5678",
subject: str = "Mathématiques",
due_on: date | None = None,
text: str = "Exercice 1 à 5",
assigned_on: date | None = None,
) -> Homework:
"""Fabrique un devoir Pronote pour les tests.
:param homework_id: Identifiant du devoir.
:param subject: Matière.
:param due_on: Date d'échéance.
:param text: Texte du devoir.
:param assigned_on: Date de distribution.
:return: Instance de Homework.
:rtype: Homework
"""
if due_on is None:
due_on = date(2026, 1, 20)
return Homework(
id=homework_id,
subject=subject,
due_on=due_on,
text=text,
assigned_on=assigned_on,
)
def _make_school_event(
label: str = "Vacances de Noël",
from_date: date | None = None,
to_date: date | None = None,
kind: SchoolEventKind = SchoolEventKind.HOLIDAY,
) -> SchoolEvent:
"""Fabrique un événement scolaire pour les tests.
:param label: Libellé de l'événement.
:param from_date: Date de début.
:param to_date: Date de fin.
:param kind: Type d'événement.
:return: Instance de SchoolEvent.
:rtype: SchoolEvent
"""
if from_date is None:
from_date = date(2026, 12, 20)
if to_date is None:
to_date = date(2027, 1, 5)
return SchoolEvent(
label=label,
from_date=from_date,
to_date=to_date,
kind=kind,
)
# --- Fixtures ---
@pytest.fixture
def mock_gateway() -> MagicMock:
"""Fournit une passerelle CalDAV mockée.
:return: MagicMock configuré comme une CalDAVGateway.
:rtype: MagicMock
"""
gateway = MagicMock()
gateway.upsert_event = MagicMock()
gateway.delete_event = MagicMock()
return gateway
@pytest.fixture
def empty_plan() -> CalDAVSyncPlan:
"""Fournit un plan de synchronisation vide.
:return: CalDAVSyncPlan vide.
:rtype: CalDAVSyncPlan
"""
return CalDAVSyncPlan()
@pytest.fixture
def plan_with_lesson_add() -> CalDAVSyncPlan:
"""Fournit un plan avec un cours à ajouter.
:return: CalDAVSyncPlan avec un cours à ajouter.
:rtype: CalDAVSyncPlan
"""
lesson = _make_lesson(lesson_id="L-1234")
return CalDAVSyncPlan(lessons_to_add=[lesson])
@pytest.fixture
def plan_with_lesson_update() -> CalDAVSyncPlan:
"""Fournit un plan avec un cours à mettre à jour.
:return: CalDAVSyncPlan avec un cours à mettre à jour.
:rtype: CalDAVSyncPlan
"""
lesson = _make_lesson(lesson_id="L-1234")
return CalDAVSyncPlan(lessons_to_update=[lesson])
@pytest.fixture
def plan_with_lesson_remove() -> CalDAVSyncPlan:
"""Fournit un plan avec un cours à supprimer.
:return: CalDAVSyncPlan avec un cours à supprimer.
:rtype: CalDAVSyncPlan
"""
return CalDAVSyncPlan(lessons_to_remove=["L-1234"])
@pytest.fixture
def plan_with_homework_add() -> CalDAVSyncPlan:
"""Fournit un plan avec un devoir à ajouter.
:return: CalDAVSyncPlan avec un devoir à ajouter.
:rtype: CalDAVSyncPlan
"""
homework = _make_homework(homework_id="HW-5678")
return CalDAVSyncPlan(homeworks_to_add=[homework])
@pytest.fixture
def plan_with_homework_remove() -> CalDAVSyncPlan:
"""Fournit un plan avec un devoir à supprimer.
:return: CalDAVSyncPlan avec un devoir à supprimer.
:rtype: CalDAVSyncPlan
"""
return CalDAVSyncPlan(homeworks_to_remove=["homework-HW-5678"])
@pytest.fixture
def plan_with_school_event_add() -> CalDAVSyncPlan:
"""Fournit un plan avec un événement scolaire à ajouter.
:return: CalDAVSyncPlan avec un événement scolaire à ajouter.
:rtype: CalDAVSyncPlan
"""
school_event = _make_school_event()
return CalDAVSyncPlan(school_events_to_add=[school_event])
@pytest.fixture
def plan_with_school_event_remove() -> CalDAVSyncPlan:
"""Fournit un plan avec un événement scolaire à supprimer.
:return: CalDAVSyncPlan avec un événement scolaire à supprimer.
:rtype: CalDAVSyncPlan
"""
return CalDAVSyncPlan(school_events_to_remove=["school-event-Vacances-2026-12-20"])
# --- Dry-run tests ---
def test_dry_run_add_does_not_call_gateway(
mock_gateway: MagicMock, plan_with_lesson_add: CalDAVSyncPlan
) -> None:
"""Vérifie que dry_run=True n'appelle pas gateway.upsert_event pour un ajout.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_add: Plan avec un cours à ajouter.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=True)
result = executor.execute(plan_with_lesson_add)
mock_gateway.upsert_event.assert_not_called()
assert result.added == 1
assert result.updated == 0
assert result.removed == 0
def test_dry_run_update_does_not_call_gateway(
mock_gateway: MagicMock, plan_with_lesson_update: CalDAVSyncPlan
) -> None:
"""Vérifie que dry_run=True n'appelle pas gateway.upsert_event pour une mise à jour.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_update: Plan avec un cours à mettre à jour.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=True)
result = executor.execute(plan_with_lesson_update)
mock_gateway.upsert_event.assert_not_called()
assert result.added == 0
assert result.updated == 1
assert result.removed == 0
def test_dry_run_delete_does_not_call_gateway(
mock_gateway: MagicMock, plan_with_lesson_remove: CalDAVSyncPlan
) -> None:
"""Vérifie que dry_run=True n'appelle pas gateway.delete_event.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_remove: Plan avec un cours à supprimer.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=True)
result = executor.execute(plan_with_lesson_remove)
mock_gateway.delete_event.assert_not_called()
assert result.added == 0
assert result.updated == 0
assert result.removed == 1
def test_dry_run_all_operations(
mock_gateway: MagicMock,
) -> None:
"""Vérifie que dry_run=True fonctionne pour toutes les catégories.
:param mock_gateway: Passerelle mockée.
:return: None
"""
plan = CalDAVSyncPlan(
lessons_to_add=[_make_lesson(lesson_id="L-0001")],
lessons_to_update=[_make_lesson(lesson_id="L-0002")],
lessons_to_remove=["L-0003"],
homeworks_to_add=[_make_homework(homework_id="HW-0001")],
homeworks_to_update=[_make_homework(homework_id="HW-0002")],
homeworks_to_remove=["homework-HW-0003"],
school_events_to_add=[_make_school_event()],
school_events_to_update=[_make_school_event(label="Événement 2")],
school_events_to_remove=["school-event-Événement-2026-01-01"],
)
executor = CalDAVSyncExecutor(mock_gateway, dry_run=True)
result = executor.execute(plan)
mock_gateway.upsert_event.assert_not_called()
mock_gateway.delete_event.assert_not_called()
assert result.added == 3 # 1 lesson + 1 homework + 1 school event
assert result.updated == 3 # 1 lesson + 1 homework + 1 school event
assert result.removed == 3 # 1 lesson + 1 homework + 1 school event
# --- Real execution tests ---
def test_real_add_calls_gateway(
mock_gateway: MagicMock, plan_with_lesson_add: CalDAVSyncPlan
) -> None:
"""Vérifie que dry_run=False appelle gateway.upsert_event pour un ajout.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_add: Plan avec un cours à ajouter.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan_with_lesson_add)
mock_gateway.upsert_event.assert_called_once()
# Vérifier que les arguments sont (texte iCalendar, uid)
call_args = mock_gateway.upsert_event.call_args
assert isinstance(call_args[0][0], str)
assert "BEGIN:VEVENT" in call_args[0][0]
assert call_args[0][1] == "L-1234"
assert result.added == 1
def test_real_update_calls_gateway(
mock_gateway: MagicMock, plan_with_lesson_update: CalDAVSyncPlan
) -> None:
"""Vérifie que dry_run=False appelle gateway.upsert_event pour une mise à jour.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_update: Plan avec un cours à mettre à jour.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(
plan_with_lesson_update,
remote_raw_by_canonical={"L-1234": "L-1234"},
)
mock_gateway.upsert_event.assert_called_once()
assert result.updated == 1
def test_update_uses_raw_uid_from_mapping(
mock_gateway: MagicMock, plan_with_lesson_update: CalDAVSyncPlan
) -> None:
"""Vérifie que la mise à jour cible l'UID brut distant du mapping.
Pour une mise à jour, l'exécuteur doit appeler ``upsert_event`` avec
l'UID brut fourni par ``remote_raw_by_canonical`` (celui stocké sur le
serveur) et non l'UID canonique, afin d'éviter la création d'un doublon.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_update: Plan avec un cours à mettre à jour.
:return: None
"""
raw_uid = "L-1234-20260905T080000Z-Index-Education"
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(
plan_with_lesson_update,
remote_raw_by_canonical={"L-1234": raw_uid},
)
call_args = mock_gateway.upsert_event.call_args
assert call_args[0][0] is not None
# Le contenu VEVENT porte l'UID canonique du modèle.
assert "UID:L-1234" in call_args[0][0]
# La cible (2e argument) est l'UID brut distant, pas l'UID canonique.
assert call_args[0][1] == raw_uid
assert result.updated == 1
assert len(result.errors) == 0
def test_update_without_mapping_raises_error(
mock_gateway: MagicMock, plan_with_lesson_update: CalDAVSyncPlan
) -> None:
"""Vérifie qu'une mise à jour sans mapping distant consigne une erreur.
Si ``remote_raw_by_canonical`` est absent (ou sans clé pour l'UID
canonique), l'exécuteur ne doit pas retomber silencieusement sur l'UID
canonique (créant un doublon) : l'erreur est consignée dans
``result.errors`` et le lot continue sans appeler la passerelle.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_update: Plan avec un cours à mettre à jour.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan_with_lesson_update)
mock_gateway.upsert_event.assert_not_called()
assert len(result.errors) == 1
assert "UID canonique sans correspondant distant" in result.errors[0]
assert result.updated == 0
assert result.status == CalDAVSyncStatus.FAILED
def test_real_delete_calls_gateway(
mock_gateway: MagicMock, plan_with_lesson_remove: CalDAVSyncPlan
) -> None:
"""Vérifie que dry_run=False appelle gateway.delete_event.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_remove: Plan avec un cours à supprimer.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan_with_lesson_remove)
mock_gateway.delete_event.assert_called_once_with("L-1234")
assert result.removed == 1
def test_real_execution_all_operations(
mock_gateway: MagicMock,
) -> None:
"""Vérifie que dry_run=False appelle la passerelle pour toutes les opérations.
:param mock_gateway: Passerelle mockée.
:return: None
"""
plan = CalDAVSyncPlan(
lessons_to_add=[_make_lesson(lesson_id="L-0001")],
lessons_to_update=[_make_lesson(lesson_id="L-0002")],
lessons_to_remove=["L-0003"],
homeworks_to_add=[_make_homework(homework_id="HW-0001")],
homeworks_to_remove=["homework-HW-0002"],
school_events_to_add=[_make_school_event()],
school_events_to_remove=["school-event-Vacances-2026-12-20"],
)
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(
plan,
remote_raw_by_canonical={"L-0002": "L-0002"},
)
# upsert_event appelé pour les ajouts et mises à jour
# 1 lesson add + 1 lesson update + 1 homework add + 1 school event add = 4
assert mock_gateway.upsert_event.call_count == 4
# delete_event appelé pour les suppressions
assert mock_gateway.delete_event.call_count == 3 # 1 lesson + 1 homework + 1 school event
assert result.added == 3 # 1 lesson + 1 homework + 1 school event
assert result.updated == 1 # 1 lesson
assert result.removed == 3 # 1 lesson + 1 homework + 1 school event
# --- Error handling tests ---
def test_error_isolation_save_failure(
mock_gateway: MagicMock,
) -> None:
"""Vérifie qu'une erreur sur upsert_event est capturée et le batch continue.
:param mock_gateway: Passerelle mockée.
:return: None
"""
lesson1 = _make_lesson(lesson_id="L-0001")
lesson2 = _make_lesson(lesson_id="L-0002")
# Configurer le mock pour lever une erreur sur le premier appel
mock_gateway.upsert_event.side_effect = [
PronoteSyncError("Échec de l'écriture"),
None, # Le deuxième appel réussit
]
plan = CalDAVSyncPlan(lessons_to_add=[lesson1, lesson2])
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan)
# Les deux appels ont été tentés
assert mock_gateway.upsert_event.call_count == 2
# Une erreur a été capturée
assert len(result.errors) == 1
assert "Échec de l'écriture" in result.errors[0]
# Le compteur d'ajouts est à 1 (seul le deuxième a réussi)
assert result.added == 1
# Le statut est FAILED car il y a des erreurs
assert result.status == CalDAVSyncStatus.FAILED
def test_error_isolation_delete_failure(
mock_gateway: MagicMock,
) -> None:
"""Vérifie qu'une erreur sur delete_event est capturée et le batch continue.
:param mock_gateway: Passerelle mockée.
:return: None
"""
mock_gateway.delete_event.side_effect = PronoteSyncError("Échec de la suppression")
plan = CalDAVSyncPlan(
lessons_to_remove=["L-0001", "L-0002"],
)
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan)
# Les deux appels ont été tentés
assert mock_gateway.delete_event.call_count == 2
# Deux erreurs ont été capturées
assert len(result.errors) == 2
assert all("Échec de la suppression" in err for err in result.errors)
# Aucun compteur de suppression n'a été incrémenté
assert result.removed == 0
# Le statut est FAILED
assert result.status == CalDAVSyncStatus.FAILED
def test_error_isolation_mixed_operations(
mock_gateway: MagicMock,
) -> None:
"""Vérifie que les erreurs sont isolées entre différentes opérations.
:param mock_gateway: Passerelle mockée.
:return: None
"""
lesson = _make_lesson(lesson_id="L-0001")
# upsert_event échoue, delete_event réussit
mock_gateway.upsert_event.side_effect = PronoteSyncError("Échec save")
mock_gateway.delete_event.return_value = None
plan = CalDAVSyncPlan(
lessons_to_add=[lesson],
lessons_to_remove=["L-0002"],
)
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan)
assert mock_gateway.upsert_event.call_count == 1
assert mock_gateway.delete_event.call_count == 1
assert len(result.errors) == 1
assert "Échec save" in result.errors[0]
assert result.added == 0
assert result.removed == 1
assert result.status == CalDAVSyncStatus.FAILED
# --- Status tests ---
def test_status_skipped_when_no_operations(
mock_gateway: MagicMock, empty_plan: CalDAVSyncPlan
) -> None:
"""Vérifie que le statut est SKIPPED quand aucune opération n'est à effectuer.
:param mock_gateway: Passerelle mockée.
:param empty_plan: Plan vide.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(empty_plan)
assert result.status == CalDAVSyncStatus.SKIPPED
assert result.added == 0
assert result.updated == 0
assert result.removed == 0
def test_status_success_when_no_errors(
mock_gateway: MagicMock, plan_with_lesson_add: CalDAVSyncPlan
) -> None:
"""Vérifie que le statut est SUCCESS quand les opérations réussissent sans erreur.
:param mock_gateway: Passerelle mockée.
:param plan_with_lesson_add: Plan avec un cours à ajouter.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan_with_lesson_add)
assert result.status == CalDAVSyncStatus.SUCCESS
def test_status_failed_when_errors_present(
mock_gateway: MagicMock,
) -> None:
"""Vérifie que le statut est FAILED quand des erreurs sont présentes.
:param mock_gateway: Passerelle mockée.
:return: None
"""
lesson = _make_lesson(lesson_id="L-0001")
mock_gateway.upsert_event.side_effect = PronoteSyncError("Échec")
plan = CalDAVSyncPlan(lessons_to_add=[lesson])
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan)
assert result.status == CalDAVSyncStatus.FAILED
# --- Homework and School Event tests ---
def test_homework_add_real_execution(
mock_gateway: MagicMock, plan_with_homework_add: CalDAVSyncPlan
) -> None:
"""Vérifie l'exécution réelle pour l'ajout d'un devoir.
:param mock_gateway: Passerelle mockée.
:param plan_with_homework_add: Plan avec un devoir à ajouter.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan_with_homework_add)
mock_gateway.upsert_event.assert_called_once()
call_args = mock_gateway.upsert_event.call_args
vcalendar_text = call_args[0][0]
assert "homework-HW-5678" in vcalendar_text
assert call_args[0][1] == "homework-HW-5678"
assert "Devoir: Mathématiques" in vcalendar_text
assert result.added == 1
def test_homework_delete_real_execution(
mock_gateway: MagicMock, plan_with_homework_remove: CalDAVSyncPlan
) -> None:
"""Vérifie l'exécution réelle pour la suppression d'un devoir.
:param mock_gateway: Passerelle mockée.
:param plan_with_homework_remove: Plan avec un devoir à supprimer.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan_with_homework_remove)
mock_gateway.delete_event.assert_called_once_with("homework-HW-5678")
assert result.removed == 1
def test_school_event_add_real_execution(
mock_gateway: MagicMock, plan_with_school_event_add: CalDAVSyncPlan
) -> None:
"""Vérifie l'exécution réelle pour l'ajout d'un événement scolaire.
:param mock_gateway: Passerelle mockée.
:param plan_with_school_event_add: Plan avec un événement scolaire à ajouter.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan_with_school_event_add)
mock_gateway.upsert_event.assert_called_once()
call_args = mock_gateway.upsert_event.call_args
vcalendar_text = call_args[0][0]
assert "school-event-Vacances de Noël-2026-12-20" in vcalendar_text
assert call_args[0][1] == "school-event-Vacances de Noël-2026-12-20"
assert result.added == 1
def test_school_event_delete_real_execution(
mock_gateway: MagicMock, plan_with_school_event_remove: CalDAVSyncPlan
) -> None:
"""Vérifie l'exécution réelle pour la suppression d'un événement scolaire.
:param mock_gateway: Passerelle mockée.
:param plan_with_school_event_remove: Plan avec un événement scolaire à supprimer.
:return: None
"""
executor = CalDAVSyncExecutor(mock_gateway, dry_run=False)
result = executor.execute(plan_with_school_event_remove)
mock_gateway.delete_event.assert_called_once_with("school-event-Vacances-2026-12-20")
assert result.removed == 1
# Ensure trailing newline

View File

@@ -0,0 +1,815 @@
"""Tests unitaires pour la passerelle CalDAV.
Ce module vérifie que :class:`CalDAVGateway` gère correctement la connexion,
la liste des événements gérés, l'écriture et la suppression d'événements,
avec une attention particulière à la sécurité (masquage des secrets).
"""
from __future__ import annotations
import logging
from datetime import datetime
from typing import TYPE_CHECKING
from unittest.mock import MagicMock
import pytest
from caldav.lib.error import NotFoundError
from icalendar import Calendar, Event
from pydantic import SecretStr
from pronote_sync.config.settings import CalDAVSettings
from pronote_sync.errors import PronoteSyncError
from pronote_sync.sync.caldav import CalDAVGateway
from pronote_sync.sync.serialization import MANAGED_PROPERTY, MANAGED_VALUE
if TYPE_CHECKING:
from _pytest.logging import LogCaptureFixture
# --- Fixtures ---
@pytest.fixture
def caldav_settings() -> CalDAVSettings:
"""Fournit des paramètres CalDAV valides pour les tests.
:return: Instance de CalDAVSettings.
:rtype: CalDAVSettings
"""
return CalDAVSettings(
url=SecretStr("https://caldav.example.com"),
username="testuser",
password=SecretStr("testpass123"),
calendar_path="/pronote-sync/",
)
@pytest.fixture
def caldav_settings_http_localhost() -> CalDAVSettings:
"""Fournit des paramètres CalDAV avec HTTP pour localhost.
:return: Instance de CalDAVSettings.
:rtype: CalDAVSettings
"""
return CalDAVSettings(
url=SecretStr("http://localhost:5232"),
username="testuser",
password=SecretStr("testpass123"),
calendar_path="/pronote-sync/",
allow_insecure_http=True,
)
@pytest.fixture
def mock_client_factory() -> MagicMock:
"""Fournit une usine de clients CalDAV mockée.
Le client retourné expose un principal dont la liste de calendriers
contient un calendrier dont le chemin se termine par ``pronote-sync``,
ce qui permet à :meth:`~pronote_sync.sync.caldav.CalDAVGateway.connect`
de le résoudre par découverte.
:return: MagicMock configuré pour retourner un client, un principal et
un calendrier mockés.
:rtype: MagicMock
"""
mock_factory = MagicMock()
# Configurer le mock pour retourner un client
mock_client = MagicMock()
mock_factory.return_value = mock_client
# Configurer le client pour exposer un principal avec un calendrier
mock_principal = MagicMock()
mock_client.principal.return_value = mock_principal
mock_calendar = MagicMock()
mock_calendar.url = "https://caldav.example.com/calendars/testuser/pronote-sync/"
mock_principal.calendars.return_value = [mock_calendar]
return mock_factory
@pytest.fixture
def mock_managed_event() -> MagicMock:
"""Fournit un événement CalDAV mocké avec le marqueur de gestion.
:return: MagicMock configuré comme un événement géré.
:rtype: MagicMock
"""
mock_event = MagicMock()
# Créer un VEVENT avec le marqueur de gestion
vevent = Event()
vevent.add("UID", "test-uid-123")
vevent.add("SUMMARY", "Test Event")
vevent.add(MANAGED_PROPERTY, MANAGED_VALUE)
mock_event.icalendar_component = Calendar()
mock_event.icalendar_component.add_component(vevent)
return mock_event
@pytest.fixture
def mock_unmanaged_event() -> MagicMock:
"""Fournit un événement CalDAV mocké sans le marqueur de gestion.
:return: MagicMock configuré comme un événement non géré.
:rtype: MagicMock
"""
mock_event = MagicMock()
# Créer un VEVENT sans le marqueur de gestion
vevent = Event()
vevent.add("UID", "unmanaged-uid-456")
vevent.add("SUMMARY", "Unmanaged Event")
mock_event.icalendar_component = Calendar()
mock_event.icalendar_component.add_component(vevent)
return mock_event
# --- Connection tests ---
def test_connect_success(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que connect() configure le client et résout le calendrier.
La résolution passe par la découverte CalDAV (``principal()`` puis
``calendars()``) et non par une concaténation d'URL.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
# Vérifier que l'usine a été appelée avec les bons paramètres
mock_client_factory.assert_called_once()
call_kwargs = mock_client_factory.call_args[1]
assert call_kwargs["url"] == "https://caldav.example.com"
assert call_kwargs["username"] == "testuser"
assert call_kwargs["password"] == "testpass123"
# Vérifier que le calendrier a été résolu par découverte
mock_client = mock_client_factory.return_value
mock_client.principal.assert_called_once_with()
mock_principal = mock_client.principal.return_value
mock_principal.calendars.assert_called_once_with()
mock_calendar = mock_principal.calendars.return_value[0]
assert gateway._calendar is mock_calendar
def test_connect_calendar_not_found_raises(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que connect() lève PronoteSyncError si aucun calendrier ne correspond.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
mock_principal = mock_client_factory.return_value.principal.return_value
other_calendar = MagicMock()
other_calendar.url = "https://caldav.example.com/calendars/testuser/autre-calendrier/"
mock_principal.calendars.return_value = [other_calendar]
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
with pytest.raises(PronoteSyncError) as exc_info:
gateway.connect()
assert "introuvable" in str(exc_info.value)
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_connect_calendar_ambiguous_raises(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que connect() lève PronoteSyncError si plusieurs calendriers correspondent.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
mock_principal = mock_client_factory.return_value.principal.return_value
cal1 = MagicMock()
cal1.url = "https://caldav.example.com/calendars/testuser/pronote-sync/"
cal2 = MagicMock()
cal2.url = "https://caldav.example.com/calendars/autreuser/pronote-sync/"
mock_principal.calendars.return_value = [cal1, cal2]
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
with pytest.raises(PronoteSyncError) as exc_info:
gateway.connect()
assert "ambigu" in str(exc_info.value)
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_connect_calendar_path_boundary_not_matched(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que connect() ne résout pas un chemin partageant un préfixe.
Un calendrier dont le chemin se termine par ``pronote-sync`` sans
frontière de composant (ex : ``.../not-pronote-sync/``) ne doit pas
correspondre au ``calendar_path`` configuré ``/pronote-sync/``.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
mock_principal = mock_client_factory.return_value.principal.return_value
boundary_calendar = MagicMock()
boundary_calendar.url = "https://caldav.example.com/calendars/testuser/not-pronote-sync/"
mock_principal.calendars.return_value = [boundary_calendar]
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
with pytest.raises(PronoteSyncError) as exc_info:
gateway.connect()
assert "introuvable" in str(exc_info.value)
def test_connect_failure_raises_pronote_sync_error(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que connect() lève PronoteSyncError en cas d'échec de connexion.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
# Créer une usine qui lève une exception
def failing_factory(*args: object, **kwargs: object) -> None:
raise ConnectionError("Connection failed")
gateway = CalDAVGateway(caldav_settings, client_factory=failing_factory)
with pytest.raises(PronoteSyncError) as exc_info:
gateway.connect()
# Vérifier que le message ne contient pas l'URL brute
error_message = str(exc_info.value)
assert "caldav.example.com" in error_message # L'URL rédigée doit être présente
assert "testpass123" not in error_message # Le mot de passe ne doit pas être présent
assert "from None" in str(exc_info.typename) or exc_info.value.__cause__ is None
def test_connect_missing_credentials_raises(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que connect() lève PronoteSyncError si les identifiants sont manquants.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
# Créer des paramètres sans URL
incomplete_settings = CalDAVSettings(
url=None,
username=None,
password=None,
calendar_path="/pronote-sync/",
)
gateway = CalDAVGateway(incomplete_settings)
with pytest.raises(PronoteSyncError) as exc_info:
gateway.connect()
assert "incomplète" in str(exc_info.value)
# --- list_managed_events tests ---
def test_list_managed_events_returns_only_managed(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
mock_managed_event: MagicMock,
mock_unmanaged_event: MagicMock,
) -> None:
"""Vérifie que list_managed_events ne retourne que les événements gérés.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:param mock_managed_event: Événement géré mocké.
:param mock_unmanaged_event: Événement non géré mocké.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
# Configurer le calendrier pour retourner les deux événements
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.search.return_value = [mock_managed_event, mock_unmanaged_event]
start = datetime(2026, 1, 1)
end = datetime(2026, 12, 31)
result = gateway.list_managed_events(start, end)
# Seuls les événements gérés doivent être retournés
assert len(result) == 1
raw_uid, canonical_uid, vevent = result[0]
assert raw_uid == "test-uid-123"
assert canonical_uid == "test-uid-123" # Pas de suffixe Pronote : canonique == brut
assert str(vevent.get("UID")) == "test-uid-123"
def test_list_managed_events_not_connected_raises(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que list_managed_events lève PronoteSyncError si non connecté.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
gateway = CalDAVGateway(caldav_settings)
# Ne pas appeler connect()
with pytest.raises(PronoteSyncError) as exc_info:
gateway.list_managed_events(datetime(2026, 1, 1), datetime(2026, 12, 31))
assert "non connectée" in str(exc_info.value)
def test_list_managed_events_caldav_error_raises(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que list_managed_events lève PronoteSyncError en cas d'erreur CalDAV.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
# Configurer le calendrier pour lever une exception
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.search.side_effect = Exception("CalDAV error")
with pytest.raises(PronoteSyncError) as exc_info:
gateway.list_managed_events(datetime(2026, 1, 1), datetime(2026, 12, 31))
# Vérifier que le message ne contient pas de secret
error_message = str(exc_info.value)
assert "testpass123" not in error_message
assert exc_info.value.__context__ is None
# --- upsert_event tests ---
def test_upsert_event_creates_new_event_when_uid_missing(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que upsert_event crée un événement si l'UID est introuvable.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.get_event_by_uid.side_effect = NotFoundError("Event not found")
vcalendar_text = "BEGIN:VCALENDAR\nBEGIN:VEVENT\nUID:test-123\nEND:VEVENT\nEND:VCALENDAR"
gateway.upsert_event(vcalendar_text, "test-123")
mock_calendar.get_event_by_uid.assert_called_once_with("test-123")
mock_calendar.add_event.assert_called_once_with(ical=vcalendar_text)
def test_upsert_event_updates_existing_event_by_uid(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
mock_managed_event: MagicMock,
) -> None:
"""Vérifie que upsert_event remplace le contenu d'un événement géré existant.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:param mock_managed_event: Événement géré mocké.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.get_event_by_uid.return_value = mock_managed_event
vcalendar_text = "BEGIN:VCALENDAR\nBEGIN:VEVENT\nUID:test-123\nEND:VEVENT\nEND:VCALENDAR"
gateway.upsert_event(vcalendar_text, "test-123")
mock_calendar.get_event_by_uid.assert_called_once_with("test-123")
assert mock_managed_event.data == vcalendar_text
mock_managed_event.save.assert_called_once()
mock_calendar.add_event.assert_not_called()
def test_upsert_event_not_connected_raises(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que upsert_event lève PronoteSyncError si non connecté.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
gateway = CalDAVGateway(caldav_settings)
with pytest.raises(PronoteSyncError) as exc_info:
gateway.upsert_event("BEGIN:VCALENDAR\nEND:VCALENDAR", "test-123")
assert "non connectée" in str(exc_info.value)
def test_upsert_event_caldav_error_raises(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que upsert_event lève PronoteSyncError en cas d'erreur CalDAV.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.get_event_by_uid.side_effect = Exception("Save error")
with pytest.raises(PronoteSyncError) as exc_info:
gateway.upsert_event("BEGIN:VCALENDAR\nEND:VCALENDAR", "test-123")
error_message = str(exc_info.value)
assert "testpass123" not in error_message
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
# --- delete_event tests ---
def test_upsert_event_refuses_unmanaged_event(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
mock_unmanaged_event: MagicMock,
) -> None:
"""Vérifie que upsert_event refuse de modifier un événement non géré.
Un événement distant existant sans le marqueur de gestion ne doit jamais
être écrasé : la méthode lève PronoteSyncError avec un message « Conflit »
et n'appelle pas ``add_event``.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:param mock_unmanaged_event: Événement non géré mocké.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.get_event_by_uid.return_value = mock_unmanaged_event
vcalendar_text = "BEGIN:VCALENDAR\nBEGIN:VEVENT\nUID:test-123\nEND:VEVENT\nEND:VCALENDAR"
with pytest.raises(PronoteSyncError) as exc_info:
gateway.upsert_event(vcalendar_text, "test-123")
assert "Conflit" in str(exc_info.value)
mock_calendar.get_event_by_uid.assert_called_once_with("test-123")
mock_calendar.add_event.assert_not_called()
mock_unmanaged_event.save.assert_not_called()
def test_delete_event_calls_calendar(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
mock_managed_event: MagicMock,
) -> None:
"""Vérifie que delete_event appelle event.delete() sur un événement géré.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:param mock_managed_event: Événement géré mocké.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
# Configurer le calendrier pour retourner un événement géré mocké
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.get_event_by_uid.return_value = mock_managed_event
gateway.delete_event("test-uid-123")
mock_calendar.get_event_by_uid.assert_called_once_with("test-uid-123")
mock_managed_event.delete.assert_called_once()
def test_delete_event_not_connected_raises(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que delete_event lève PronoteSyncError si non connecté.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
gateway = CalDAVGateway(caldav_settings)
with pytest.raises(PronoteSyncError) as exc_info:
gateway.delete_event("test-uid-123")
assert "non connectée" in str(exc_info.value)
def test_delete_event_caldav_error_raises(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que delete_event lève PronoteSyncError en cas d'erreur CalDAV.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.get_event_by_uid.side_effect = Exception("Delete error")
with pytest.raises(PronoteSyncError) as exc_info:
gateway.delete_event("test-uid-123")
error_message = str(exc_info.value)
assert "testpass123" not in error_message
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_delete_event_refuses_unmanaged_event(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
mock_unmanaged_event: MagicMock,
caplog: LogCaptureFixture,
) -> None:
"""Vérifie que delete_event refuse de supprimer un événement non géré.
L'événement distant sans le marqueur de gestion ne doit jamais être
supprimé : la méthode retourne sans erreur, journalise un avertissement
et n'appelle pas ``event.delete()``.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:param mock_unmanaged_event: Événement non géré mocké.
:param caplog: Capture des journaux pytest.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.get_event_by_uid.return_value = mock_unmanaged_event
with caplog.at_level(logging.WARNING):
gateway.delete_event("test-uid-123")
mock_calendar.get_event_by_uid.assert_called_once_with("test-uid-123")
mock_unmanaged_event.delete.assert_not_called()
assert any("Suppression refusée" in record.getMessage() for record in caplog.records)
def test_delete_event_idempotent_when_not_found(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que delete_event est idempotent quand l'UID est introuvable.
Une suppression d'un événement déjà absent est un succès silencieux :
aucune exception n'est levée.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
mock_calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[
0
]
mock_calendar.get_event_by_uid.side_effect = NotFoundError("Event not found")
gateway.delete_event("test-uid-123")
mock_calendar.get_event_by_uid.assert_called_once_with("test-uid-123")
# --- Context manager tests ---
def test_context_manager_calls_connect_and_close(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que le gestionnaire de contexte appelle connect() et close().
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
with gateway:
# À l'intérieur du contexte, le client doit être configuré
assert gateway._client is not None
assert gateway._calendar is not None
# Après la sortie du contexte, le client doit être réinitialisé
assert gateway._client is None
assert gateway._calendar is None
def test_context_manager_connect_failure(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que le gestionnaire de contexte propage l'erreur de connexion.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
def failing_factory(*args: object, **kwargs: object) -> None:
raise ConnectionError("Connection failed")
gateway = CalDAVGateway(caldav_settings, client_factory=failing_factory)
with pytest.raises(PronoteSyncError):
with gateway:
pass # Ne doit pas être atteint
# --- Security tests ---
def test_no_plaintext_password_in_vars(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que vars(gateway) ne contient pas le mot de passe en clair.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
gateway = CalDAVGateway(caldav_settings)
gateway_vars = vars(gateway)
# Vérifier que le mot de passe n'est pas en clair
assert "testpass123" not in str(gateway_vars)
# Vérifier que l'URL brute avec credentials n'est pas en clair
# Note: _redacted_url contient l'URL sans credentials, ce qui est acceptable
assert "testuser:testpass123@" not in str(gateway_vars)
def test_repr_does_not_leak_password(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que repr(gateway) ne fuit pas le mot de passe.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
gateway = CalDAVGateway(caldav_settings)
repr_str = repr(gateway)
assert "testpass123" not in repr_str
def test_str_does_not_leak_password(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que str(gateway) ne fuit pas le mot de passe.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
gateway = CalDAVGateway(caldav_settings)
str_str = str(gateway)
assert "testpass123" not in str_str
def test_redacted_url_in_error_message(
caldav_settings: CalDAVSettings,
) -> None:
"""Vérifie que les messages d'erreur contiennent l'URL rédigée, pas l'URL brute.
:param caldav_settings: Paramètres CalDAV valides.
:return: None
"""
def failing_factory(*args: object, **kwargs: object) -> None:
raise ConnectionError("Connection failed")
gateway = CalDAVGateway(caldav_settings, client_factory=failing_factory)
try:
gateway.connect()
except PronoteSyncError as exc:
error_message = str(exc)
# L'URL doit être rédigée (sans credentials)
assert "caldav.example.com" in error_message
# Le mot de passe ne doit pas être présent
assert "testpass123" not in error_message
# L'URL complète avec credentials ne doit pas être présente
assert "https://testuser:testpass123@caldav.example.com" not in error_message
def test_close_clears_secrets(
caldav_settings: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que close() réinitialise les secrets.
:param caldav_settings: Paramètres CalDAV valides.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
gateway.connect()
# Avant close(), les secrets sont présents
assert gateway._url_secret is not None
assert gateway._password_secret is not None
gateway.close()
# Après close(), les secrets sont réinitialisés
assert gateway._url_secret is None
assert gateway._password_secret is None
assert gateway._client is None
assert gateway._calendar is None
# --- HTTP localhost tests ---
def test_http_localhost_allowed_with_flag(
caldav_settings_http_localhost: CalDAVSettings,
mock_client_factory: MagicMock,
) -> None:
"""Vérifie que HTTP est autorisé pour localhost avec allow_insecure_http=True.
:param caldav_settings_http_localhost: Paramètres CalDAV avec HTTP pour localhost.
:param mock_client_factory: Usine de clients mockée.
:return: None
"""
gateway = CalDAVGateway(caldav_settings_http_localhost, client_factory=mock_client_factory)
# La validation doit réussir (pas d'erreur levée)
# Le client_factory est appelé avec l'URL HTTP
gateway.connect()
call_kwargs = mock_client_factory.call_args[1]
assert call_kwargs["url"] == "http://localhost:5232"
# Ensure trailing newline

View File

@@ -0,0 +1,653 @@
"""Tests unitaires pour le planificateur de synchronisation CalDAV.
Ce module vérifie que la fonction :func:`compute_plan` produit correctement
les listes d'ajouts, mises à jour et suppressions pour chaque catégorie
(cours, devoirs, événements scolaires) en comparant les données Pronote
normalisées aux événements distants marqués comme gérés.
"""
from __future__ import annotations
from datetime import date, datetime
from icalendar import Event
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.models.homework import Homework
from pronote_sync.models.pronote import PronoteData
from pronote_sync.sync.planner import compute_plan
from pronote_sync.sync.serialization import (
lesson_to_vevent,
)
# --- Helper fixtures ---
def _make_lesson(
lesson_id: str = "L-1234",
subject: str = "Mathématiques",
start: datetime | None = None,
end: datetime | None = None,
status: LessonStatus = LessonStatus.NORMAL,
content: str | None = None,
group: str | None = None,
) -> Lesson:
"""Fabrique un cours Pronote pour les tests.
:param lesson_id: Identifiant du cours.
:param subject: Matière.
:param start: Date/heure de début.
:param end: Date/heure de fin.
:param status: Statut du cours.
:param content: Contenu pédagogique.
:param group: Groupe.
:return: Instance de Lesson.
:rtype: Lesson
"""
if start is None:
start = datetime(2026, 1, 15, 8, 0)
if end is None:
end = datetime(2026, 1, 15, 9, 0)
return Lesson(
id=lesson_id,
start=start,
end=end,
subject=subject,
status=status,
content=content,
group=group,
)
def _make_homework(
homework_id: str = "HW-5678",
subject: str = "Mathématiques",
due_on: date | None = None,
text: str = "Exercice 1 à 5",
assigned_on: date | None = None,
) -> Homework:
"""Fabrique un devoir Pronote pour les tests.
:param homework_id: Identifiant du devoir.
:param subject: Matière.
:param due_on: Date d'échéance.
:param text: Texte du devoir.
:param assigned_on: Date de distribution.
:return: Instance de Homework.
:rtype: Homework
"""
if due_on is None:
due_on = date(2026, 1, 20)
return Homework(
id=homework_id,
subject=subject,
due_on=due_on,
text=text,
assigned_on=assigned_on,
)
def _make_school_event(
label: str = "Vacances de Noël",
from_date: date | None = None,
to_date: date | None = None,
kind: SchoolEventKind = SchoolEventKind.HOLIDAY,
) -> SchoolEvent:
"""Fabrique un événement scolaire pour les tests.
:param label: Libellé de l'événement.
:param from_date: Date de début.
:param to_date: Date de fin.
:param kind: Type d'événement.
:return: Instance de SchoolEvent.
:rtype: SchoolEvent
"""
if from_date is None:
from_date = date(2026, 12, 20)
if to_date is None:
to_date = date(2027, 1, 5)
return SchoolEvent(
label=label,
from_date=from_date,
to_date=to_date,
kind=kind,
)
def _make_vevent(
uid: str,
summary: str,
dtstart: datetime,
dtend: datetime,
status: str = "CONFIRMED",
categories: list[str] | None = None,
) -> Event:
"""Fabrique un VEVENT iCalendar pour les tests.
:param uid: UID de l'événement.
:param summary: Résumé.
:param dtstart: Date/heure de début.
:param dtend: Date/heure de fin.
:param status: Statut.
:param categories: Catégories.
:return: Instance de Event.
:rtype: Event
"""
from icalendar import vDatetime
event = Event()
event.add("UID", uid)
event.add("SUMMARY", summary)
event.add("DTSTART", vDatetime(dtstart))
event.add("DTEND", vDatetime(dtend))
event.add("STATUS", status)
if categories:
event.add("CATEGORIES", categories)
# Ajouter le marqueur de gestion
from pronote_sync.sync.serialization import MANAGED_PROPERTY, MANAGED_VALUE
event.add(MANAGED_PROPERTY, MANAGED_VALUE)
return event
def _make_pronote_data(
lessons: list[Lesson] | None = None,
homeworks: list[Homework] | None = None,
school_events: list[SchoolEvent] | None = None,
) -> PronoteData:
"""Fabrique des données Pronote pour les tests.
:param lessons: Liste des cours.
:param homeworks: Liste des devoirs.
:param school_events: Liste des événements scolaires.
:return: Instance de PronoteData.
:rtype: PronoteData
"""
return PronoteData(
lessons=lessons or [],
homeworks=homeworks or [],
school_events=school_events or [],
messages=[],
target_date=date(2026, 1, 15),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# --- Tests for lessons ---
def test_lesson_add_when_not_in_remote() -> None:
"""Vérifie qu'un cours non présent à distance va dans lessons_to_add.
:return: None
"""
lesson = _make_lesson(lesson_id="L-1234")
pronote_data = _make_pronote_data(lessons=[lesson])
remote_managed: list[tuple[str, str, Event]] = []
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.lessons_to_add) == 1
assert plan.lessons_to_add[0].id == "L-1234"
assert len(plan.lessons_to_update) == 0
assert len(plan.lessons_to_remove) == 0
def test_lesson_update_when_signature_differs() -> None:
"""Vérifie qu'un cours présent à distance avec une signature différente va dans lessons_to_update.
:return: None
"""
lesson = _make_lesson(lesson_id="L-1234", subject="Mathématiques")
pronote_data = _make_pronote_data(lessons=[lesson])
# Créer un VEVENT distant avec un sujet différent
remote_event = _make_vevent(
uid="L-1234",
summary="Physique", # Différent
dtstart=datetime(2026, 1, 15, 8, 0),
dtend=datetime(2026, 1, 15, 9, 0),
)
remote_managed: list[tuple[str, str, Event]] = [("L-1234", "L-1234", remote_event)]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.lessons_to_add) == 0
assert len(plan.lessons_to_update) == 1
assert plan.lessons_to_update[0].id == "L-1234"
assert len(plan.lessons_to_remove) == 0
def test_lesson_idempotent_when_signature_same() -> None:
"""Vérifie qu'un cours présent à distance avec la même signature n'apparaît dans aucune liste.
:return: None
"""
lesson = _make_lesson(lesson_id="L-1234", subject="Mathématiques")
pronote_data = _make_pronote_data(lessons=[lesson])
# Créer un VEVENT distant avec les mêmes propriétés
remote_event = lesson_to_vevent(lesson)
remote_managed: list[tuple[str, str, Event]] = [("L-1234", "L-1234", remote_event)]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.lessons_to_add) == 0
assert len(plan.lessons_to_update) == 0
assert len(plan.lessons_to_remove) == 0
def test_lesson_remove_when_not_in_local() -> None:
"""Vérifie qu'un UID distant non présent en local va dans lessons_to_remove.
:return: None
"""
pronote_data = _make_pronote_data(lessons=[])
remote_event = _make_vevent(
uid="L-9999",
summary="Ancien cours",
dtstart=datetime(2026, 1, 15, 8, 0),
dtend=datetime(2026, 1, 15, 9, 0),
)
remote_managed: list[tuple[str, str, Event]] = [("L-9999", "L-9999", remote_event)]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.lessons_to_add) == 0
assert len(plan.lessons_to_update) == 0
assert len(plan.lessons_to_remove) == 1
assert plan.lessons_to_remove[0] == "L-9999"
def test_plan_with_suffixed_remote_uid_matches_canonical() -> None:
"""Vérifie qu'un UID distant suffixé apparié par UID canonique ne produit rien.
L'événement distant porte un UID brut suffixé
(``L-1234-20260905T080000Z-Index-Education``) dont la forme canonique
(``L-1234``) correspond au cours local ; les signatures étant identiques,
le plan doit être vide — sans ajout, mise à jour ou suppression artificiels.
:return: None
"""
lesson = _make_lesson(lesson_id="L-1234", subject="Mathématiques")
pronote_data = _make_pronote_data(lessons=[lesson])
# VEVENT distant construit à partir du même cours : contenu sémantique
# identique (seul l'UID brut stocké diffère, capturé par le triplet).
remote_event = lesson_to_vevent(lesson)
remote_managed: list[tuple[str, str, Event]] = [
("L-1234-20260905T080000Z-Index-Education", "L-1234", remote_event)
]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.lessons_to_add) == 0
assert len(plan.lessons_to_update) == 0
assert len(plan.lessons_to_remove) == 0
def test_plan_with_suffixed_remote_uid_and_no_local_adds_to_remove() -> None:
"""Vérifie que la suppression d'un UID distant suffixé utilise l'UID brut.
Un événement distant orphelin (aucun cours local) dont l'UID brut est
suffixé doit être supprimé en ciblant l'UID brut stocké sur le serveur,
et non sa forme canonique.
:return: None
"""
pronote_data = _make_pronote_data(lessons=[])
raw_uid = "L-9999-20260101T080000Z-Index-Education"
remote_event = _make_vevent(
uid=raw_uid,
summary="Ancien cours",
dtstart=datetime(2026, 1, 15, 8, 0),
dtend=datetime(2026, 1, 15, 9, 0),
)
remote_managed: list[tuple[str, str, Event]] = [(raw_uid, "L-9999", remote_event)]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.lessons_to_add) == 0
assert len(plan.lessons_to_update) == 0
assert len(plan.lessons_to_remove) == 1
# La liste de suppression contient l'UID brut, pas la forme canonique.
assert plan.lessons_to_remove[0] == raw_uid
def test_lesson_cancelled_preserved() -> None:
"""Vérifie qu'un cours annulé est traité normalement (ajout/mise à jour).
:return: None
"""
lesson = _make_lesson(lesson_id="L-1234", status=LessonStatus.CANCELLED)
pronote_data = _make_pronote_data(lessons=[lesson])
remote_managed: list[tuple[str, str, Event]] = []
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
# Un cours annulé doit aller dans lessons_to_add comme n'importe quel autre cours
assert len(plan.lessons_to_add) == 1
assert plan.lessons_to_add[0].id == "L-1234"
assert plan.lessons_to_add[0].status == LessonStatus.CANCELLED
# --- Tests for homeworks ---
def test_homework_add_when_not_in_remote() -> None:
"""Vérifie qu'un devoir non présent à distance va dans homeworks_to_add.
:return: None
"""
homework = _make_homework(homework_id="HW-5678")
pronote_data = _make_pronote_data(homeworks=[homework])
remote_managed: list[tuple[str, str, Event]] = []
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.homeworks_to_add) == 1
assert plan.homeworks_to_add[0].id == "HW-5678"
assert len(plan.homeworks_to_update) == 0
assert len(plan.homeworks_to_remove) == 0
def test_homework_update_when_signature_differs() -> None:
"""Vérifie qu'un devoir présent à distance avec une signature différente va dans homeworks_to_update.
:return: None
"""
homework = _make_homework(homework_id="HW-5678", subject="Mathématiques")
pronote_data = _make_pronote_data(homeworks=[homework])
# Créer un VEVENT distant avec un sujet différent
remote_event = _make_vevent(
uid="homework-HW-5678",
summary="Devoir: Physique", # Différent
dtstart=datetime(2026, 1, 20, 8, 0),
dtend=datetime(2026, 1, 20, 18, 0),
)
remote_managed: list[tuple[str, str, Event]] = [
("homework-HW-5678", "homework-HW-5678", remote_event)
]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.homeworks_to_add) == 0
assert len(plan.homeworks_to_update) == 1
assert plan.homeworks_to_update[0].id == "HW-5678"
assert len(plan.homeworks_to_remove) == 0
def test_homework_remove_when_not_in_local() -> None:
"""Vérifie qu'un UID de devoir distant non présent en local va dans homeworks_to_remove.
:return: None
"""
pronote_data = _make_pronote_data(homeworks=[])
remote_event = _make_vevent(
uid="homework-HW-9999",
summary="Devoir: Ancien devoir",
dtstart=datetime(2026, 1, 20, 8, 0),
dtend=datetime(2026, 1, 20, 18, 0),
)
remote_managed: list[tuple[str, str, Event]] = [
("homework-HW-9999", "homework-HW-9999", remote_event)
]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.homeworks_to_add) == 0
assert len(plan.homeworks_to_update) == 0
assert len(plan.homeworks_to_remove) == 1
assert plan.homeworks_to_remove[0] == "homework-HW-9999"
# --- Tests for school events ---
def test_school_event_add_when_not_in_remote() -> None:
"""Vérifie qu'un événement scolaire non présent à distance va dans school_events_to_add.
:return: None
"""
school_event = _make_school_event(
label="Vacances de Noël",
from_date=date(2026, 12, 20),
)
pronote_data = _make_pronote_data(school_events=[school_event])
remote_managed: list[tuple[str, str, Event]] = []
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.school_events_to_add) == 1
assert plan.school_events_to_add[0].label == "Vacances de Noël"
assert len(plan.school_events_to_update) == 0
assert len(plan.school_events_to_remove) == 0
def test_school_event_update_when_signature_differs() -> None:
"""Vérifie qu'un événement scolaire présent à distance avec une signature différente
va dans school_events_to_update.
:return: None
"""
school_event = _make_school_event(
label="Vacances de Noël",
from_date=date(2026, 12, 20),
to_date=date(2027, 1, 5),
kind=SchoolEventKind.HOLIDAY,
)
pronote_data = _make_pronote_data(school_events=[school_event])
# Créer un VEVENT distant avec le même UID mais un libellé différent
# L'UID doit correspondre à celui généré par school_event_to_vevent
remote_event = _make_vevent(
uid="school-event-Vacances de Noël-2026-12-20",
summary="Vacances d'hiver", # Différent du local "Vacances de Noël"
dtstart=datetime(2026, 12, 20, 0, 0),
dtend=datetime(2027, 1, 5, 0, 0),
)
remote_managed: list[tuple[str, str, Event]] = [
(
"school-event-Vacances de Noël-2026-12-20",
"school-event-Vacances de Noël-2026-12-20",
remote_event,
)
]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.school_events_to_add) == 0
assert len(plan.school_events_to_update) == 1
assert len(plan.school_events_to_remove) == 0
def test_school_event_remove_when_not_in_local() -> None:
"""Vérifie qu'un UID d'événement scolaire distant non présent en local
va dans school_events_to_remove.
:return: None
"""
pronote_data = _make_pronote_data(school_events=[])
remote_event = _make_vevent(
uid="school-event-Ancien événement-2026-01-01",
summary="Ancien événement",
dtstart=datetime(2026, 1, 1, 0, 0),
dtend=datetime(2026, 1, 2, 0, 0),
)
remote_managed: list[tuple[str, str, Event]] = [
(
"school-event-Ancien événement-2026-01-01",
"school-event-Ancien événement-2026-01-01",
remote_event,
)
]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.school_events_to_add) == 0
assert len(plan.school_events_to_update) == 0
assert len(plan.school_events_to_remove) == 1
assert plan.school_events_to_remove[0] == "school-event-Ancien événement-2026-01-01"
# --- Tests for UID routing ---
def test_uid_routing_homework_to_remove() -> None:
"""Vérifie qu'un UID distant commençant par 'homework-' va dans homeworks_to_remove.
:return: None
"""
pronote_data = _make_pronote_data(lessons=[], homeworks=[], school_events=[])
remote_event = _make_vevent(
uid="homework-HW-9999",
summary="Devoir à supprimer",
dtstart=datetime(2026, 1, 20, 8, 0),
dtend=datetime(2026, 1, 20, 18, 0),
)
remote_managed: list[tuple[str, str, Event]] = [
("homework-HW-9999", "homework-HW-9999", remote_event)
]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
# Ne doit PAS aller dans lessons_to_remove
assert len(plan.lessons_to_remove) == 0
assert len(plan.homeworks_to_remove) == 1
assert plan.homeworks_to_remove[0] == "homework-HW-9999"
def test_uid_routing_school_event_to_remove() -> None:
"""Vérifie qu'un UID distant commençant par 'school-event-' va dans school_events_to_remove.
:return: None
"""
pronote_data = _make_pronote_data(lessons=[], homeworks=[], school_events=[])
remote_event = _make_vevent(
uid="school-event-Vacances-2026-12-20",
summary="Événement à supprimer",
dtstart=datetime(2026, 12, 20, 0, 0),
dtend=datetime(2027, 1, 5, 0, 0),
)
remote_managed: list[tuple[str, str, Event]] = [
("school-event-Vacances-2026-12-20", "school-event-Vacances-2026-12-20", remote_event)
]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
# Ne doit PAS aller dans lessons_to_remove
assert len(plan.lessons_to_remove) == 0
assert len(plan.school_events_to_remove) == 1
assert plan.school_events_to_remove[0] == "school-event-Vacances-2026-12-20"
# --- Tests for empty inputs ---
def test_empty_inputs_empty_plan() -> None:
"""Vérifie que des entrées vides produisent un plan vide.
:return: None
"""
pronote_data = _make_pronote_data(lessons=[], homeworks=[], school_events=[])
remote_managed: list[tuple[str, str, Event]] = []
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
assert len(plan.lessons_to_add) == 0
assert len(plan.lessons_to_update) == 0
assert len(plan.lessons_to_remove) == 0
assert len(plan.homeworks_to_add) == 0
assert len(plan.homeworks_to_update) == 0
assert len(plan.homeworks_to_remove) == 0
assert len(plan.school_events_to_add) == 0
assert len(plan.school_events_to_update) == 0
assert len(plan.school_events_to_remove) == 0
# --- Tests for mixed scenarios ---
def test_mixed_scenario() -> None:
"""Vérifie un scénario mixte avec ajouts, mises à jour et suppressions.
:return: None
"""
# Données locales
lesson1 = _make_lesson(lesson_id="L-0001") # Nouveau
lesson2 = _make_lesson(lesson_id="L-0002", subject="Mathématiques") # À mettre à jour
homework1 = _make_homework(homework_id="HW-0001") # Nouveau
pronote_data = _make_pronote_data(
lessons=[lesson1, lesson2],
homeworks=[homework1],
school_events=[],
)
# Événements distants
# L-0002 existe mais avec un sujet différent
remote_lesson2 = _make_vevent(
uid="L-0002",
summary="Physique",
dtstart=datetime(2026, 1, 15, 8, 0),
dtend=datetime(2026, 1, 15, 9, 0),
)
# L-0003 n'existe plus localement
remote_lesson3 = _make_vevent(
uid="L-0003",
summary="Ancien cours",
dtstart=datetime(2026, 1, 15, 8, 0),
dtend=datetime(2026, 1, 15, 9, 0),
)
remote_managed: list[tuple[str, str, Event]] = [
("L-0002", "L-0002", remote_lesson2),
("L-0003", "L-0003", remote_lesson3),
]
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
# Ajouts
assert len(plan.lessons_to_add) == 1
assert plan.lessons_to_add[0].id == "L-0001"
assert len(plan.homeworks_to_add) == 1
assert plan.homeworks_to_add[0].id == "HW-0001"
# Mises à jour
assert len(plan.lessons_to_update) == 1
assert plan.lessons_to_update[0].id == "L-0002"
# Suppressions
assert len(plan.lessons_to_remove) == 1
assert plan.lessons_to_remove[0] == "L-0003"
def test_unmanaged_events_not_in_remote_managed() -> None:
"""Vérifie que remote_managed ne contient que des événements gérés.
Le contrat indique que remote_managed ne contient déjà que des événements
marqués comme gérés. Le planner ne doit pas filtrer.
:return: None
"""
lesson = _make_lesson(lesson_id="L-1234")
pronote_data = _make_pronote_data(lessons=[lesson])
# remote_managed ne contient que des événements gérés (par hypothèse)
# Donc pas besoin de tester le filtrage ici - c'est la responsabilité de list_managed_events
remote_managed: list[tuple[str, str, Event]] = []
plan, raw_mapping = compute_plan(pronote_data, remote_managed)
# Le cours doit être dans lessons_to_add
assert len(plan.lessons_to_add) == 1
# Ensure trailing newline

View File

@@ -0,0 +1,379 @@
"""Tests de sécurité pour la passerelle CalDAV et sa configuration.
Ce module vérifie que les secrets (URL, mot de passe) ne fuient jamais dans
les représentations textuelles, les logs, les messages d'erreur ou les
chaînages d'exceptions de la configuration CalDAV et de la passerelle.
"""
from __future__ import annotations
import logging
from datetime import datetime
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock
import pytest
from pydantic import SecretStr, ValidationError
from pronote_sync.config.settings import CalDAVSettings
from pronote_sync.errors import PronoteSyncError
from pronote_sync.sync.caldav import CalDAVGateway
if TYPE_CHECKING:
from _pytest.logging import LogCaptureFixture
# Sentinelles pour détecter les fuites de secrets dans les tests
SENTINEL_URL = "https://user:pass-super-secret-12345@caldav.example.com/secret-path/"
SENTINEL_PASSWORD = "super-secret-password-67890"
SENTINEL_HTTP_URL = "http://caldav.example.com/"
SENTINEL_HTTP_LOCALHOST = "http://localhost:5232/caldav/"
SENTINEL_HTTP_NON_LOOPBACK = "http://insecure.example.com/caldav/"
class TestCalDAVSettingsSecurity:
"""Tests de sécurité pour la configuration CalDAV (CalDAVSettings)."""
def test_url_redaction_in_repr(self) -> None:
"""Vérifie que l'URL brute n'apparaît pas dans repr(settings)."""
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
repr_str = repr(settings)
assert SENTINEL_URL not in repr_str
assert "pass-super-secret-12345" not in repr_str
# L'URL est masquée par redact_url qui retourne REDACTED_URL ou une URL avec REDACTED
assert "REDACTED" in repr_str or "**********" in repr_str
def test_url_redaction_in_str(self) -> None:
"""Vérifie que l'URL brute n'apparaît pas dans str(settings)."""
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
str_str = str(settings)
assert SENTINEL_URL not in str_str
assert "pass-super-secret-12345" not in str_str
# L'URL est masquée par redact_url
assert "REDACTED" in str_str or "**********" in str_str
def test_url_redaction_in_model_dump(self) -> None:
"""Vérifie que l'URL brute n'apparaît pas dans model_dump()."""
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
dumped = settings.model_dump()
# Vérifie que l'URL n'est pas dans les valeurs du dict
for value in dumped.values():
if isinstance(value, str):
assert SENTINEL_URL not in value
assert "pass-super-secret-12345" not in value
# Vérifie que la version rédigée est présente
assert "REDACTED" in str(dumped)
def test_password_not_in_repr(self) -> None:
"""Vérifie que le mot de passe n'apparaît pas dans repr(settings)."""
settings = CalDAVSettings(
url=SecretStr("https://caldav.example.com/"),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
repr_str = repr(settings)
assert SENTINEL_PASSWORD not in repr_str
assert "**********" in repr_str
def test_password_not_in_str(self) -> None:
"""Vérifie que le mot de passe n'apparaît pas dans str(settings)."""
settings = CalDAVSettings(
url=SecretStr("https://caldav.example.com/"),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
str_str = str(settings)
assert SENTINEL_PASSWORD not in str_str
assert "**********" in str_str
def test_https_enforcement(self) -> None:
"""Vérifie que HTTP (non-localhost) est rejeté par défaut."""
with pytest.raises(ValidationError) as exc_info:
CalDAVSettings(
url=SecretStr(SENTINEL_HTTP_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
assert exc_info.value.error_count() >= 1
def test_https_accepted(self) -> None:
"""Vérifie que HTTPS est accepté sans erreur."""
settings = CalDAVSettings(
url=SecretStr("https://caldav.example.com/"),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
assert settings.url is not None
def test_http_localhost_without_flag(self) -> None:
"""Vérifie que HTTP localhost est rejeté sans allow_insecure_http."""
with pytest.raises(ValidationError) as exc_info:
CalDAVSettings(
url=SecretStr(SENTINEL_HTTP_LOCALHOST),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
assert exc_info.value.error_count() >= 1
def test_http_localhost_with_flag(self) -> None:
"""Vérifie que HTTP localhost est accepté avec allow_insecure_http=True."""
settings = CalDAVSettings(
url=SecretStr(SENTINEL_HTTP_LOCALHOST),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
allow_insecure_http=True,
)
assert settings.url is not None
def test_http_non_loopback_with_flag(self) -> None:
"""Vérifie que HTTP non-loopback est rejeté même avec allow_insecure_http=True."""
with pytest.raises(ValidationError) as exc_info:
CalDAVSettings(
url=SecretStr(SENTINEL_HTTP_NON_LOOPBACK),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
allow_insecure_http=True,
)
assert exc_info.value.error_count() >= 1
def test_validation_error_message_safe(self) -> None:
"""Vérifie que les messages d'erreur de validation ne contiennent pas l'URL brute."""
with pytest.raises(ValidationError) as exc_info:
CalDAVSettings(
url=SecretStr(SENTINEL_HTTP_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
error_str = str(exc_info.value)
assert SENTINEL_HTTP_URL not in error_str
assert "caldav.example.com" not in error_str
class TestCalDAVGatewaySecurity:
"""Tests de sécurité pour la passerelle CalDAV (CalDAVGateway)."""
def test_password_not_stored_in_plaintext(self) -> None:
"""Vérifie que le mot de passe n'est pas stocké en clair sur l'instance."""
settings = CalDAVSettings(
url=SecretStr("https://caldav.example.com/"),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
gateway = CalDAVGateway(settings)
# Vérifie que le mot de passe en clair n'est dans aucun attribut
for attr_name in vars(gateway):
attr_value = getattr(gateway, attr_name)
if isinstance(attr_value, str):
assert SENTINEL_PASSWORD not in attr_value
elif isinstance(attr_value, SecretStr):
# SecretStr peut contenir le secret, mais pas en clair
assert SENTINEL_PASSWORD not in str(attr_value)
def test_error_messages_redacted(self, caplog: LogCaptureFixture) -> None:
"""Vérifie que les messages d'erreur ne contiennent pas de secrets."""
def _leaky_client_factory(**kwargs: Any) -> None:
# Utiliser un message d'erreur qui contient des secrets dans un format détectable
raise Exception(f"Connection failed to {SENTINEL_URL}?token={SENTINEL_PASSWORD}")
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
gateway = CalDAVGateway(settings, client_factory=_leaky_client_factory)
with caplog.at_level(logging.ERROR):
with pytest.raises(PronoteSyncError) as exc_info:
gateway.connect()
# Vérifie que le message d'erreur ne contient pas les sentinelles
error_msg = str(exc_info.value)
assert SENTINEL_URL not in error_msg
assert SENTINEL_PASSWORD not in error_msg
# Vérifie que les logs ne contiennent pas les sentinelles
for record in caplog.records:
log_msg = record.getMessage()
assert SENTINEL_URL not in log_msg
assert SENTINEL_PASSWORD not in log_msg
def test_exception_cause_and_context_is_none(self) -> None:
"""Vérifie que PronoteSyncError.__cause__ et __context__ sont None."""
def _leaky_client_factory(**kwargs: Any) -> None:
raise Exception(f"Connection failed to {SENTINEL_URL}?token={SENTINEL_PASSWORD}")
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
gateway = CalDAVGateway(settings, client_factory=_leaky_client_factory)
with pytest.raises(PronoteSyncError) as exc_info:
gateway.connect()
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_logs_redacted_on_list_managed_events_error(self, caplog: LogCaptureFixture) -> None:
"""Vérifie que les logs sont expurgés lors d'une erreur dans list_managed_events."""
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
# Créer un mock de client qui lève une exception avec des secrets
mock_client = MagicMock()
mock_principal = MagicMock()
mock_calendar = MagicMock()
mock_calendar.url = "https://caldav.example.com/cal/"
mock_calendar.search.side_effect = Exception(
f"Search failed at {SENTINEL_URL}?token={SENTINEL_PASSWORD}"
)
mock_principal.calendars.return_value = [mock_calendar]
mock_client.principal.return_value = mock_principal
gateway = CalDAVGateway(settings, client_factory=lambda **kw: mock_client)
gateway._client = mock_client
gateway._calendar = mock_calendar
with caplog.at_level(logging.ERROR):
with pytest.raises(PronoteSyncError) as exc_info:
gateway.list_managed_events(start=datetime(2026, 1, 15), end=datetime(2026, 1, 20))
# Vérifie que le message d'erreur ne contient pas les sentinelles
error_msg = str(exc_info.value)
assert SENTINEL_URL not in error_msg
assert SENTINEL_PASSWORD not in error_msg
# Vérifie que les logs ne contiennent pas les sentinelles
for record in caplog.records:
log_msg = record.getMessage()
assert SENTINEL_URL not in log_msg
assert SENTINEL_PASSWORD not in log_msg
# Vérifie que l'exception n'est chaînée à aucune exception brute
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_logs_redacted_on_upsert_event_error(self, caplog: LogCaptureFixture) -> None:
"""Vérifie que les logs sont expurgés lors d'une erreur dans upsert_event."""
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
# Créer un mock de calendrier dont la recherche par UID lève une
# exception avec des secrets (simule une fuite de la bibliothèque caldav)
mock_calendar = MagicMock()
mock_calendar.get_event_by_uid.side_effect = Exception(
f"Save failed at {SENTINEL_URL}?token={SENTINEL_PASSWORD}"
)
gateway = CalDAVGateway(settings)
gateway._calendar = mock_calendar
with caplog.at_level(logging.ERROR):
with pytest.raises(PronoteSyncError) as exc_info:
gateway.upsert_event("BEGIN:VCALENDAR\nEND:VCALENDAR", "test-uid")
# Vérifie que le message d'erreur ne contient pas les sentinelles
error_msg = str(exc_info.value)
assert SENTINEL_URL not in error_msg
assert SENTINEL_PASSWORD not in error_msg
# Vérifie que les logs ne contiennent pas les sentinelles
for record in caplog.records:
log_msg = record.getMessage()
assert SENTINEL_URL not in log_msg
assert SENTINEL_PASSWORD not in log_msg
# Vérifie que l'exception n'est chaînée à aucune exception brute
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_logs_redacted_on_delete_event_error(self, caplog: LogCaptureFixture) -> None:
"""Vérifie que les logs sont expurgés lors d'une erreur dans delete_event."""
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
# Créer un mock de calendrier qui lève une exception avec des secrets
mock_calendar = MagicMock()
mock_calendar.get_event_by_uid.side_effect = Exception(
f"Delete failed at {SENTINEL_URL}?token={SENTINEL_PASSWORD}"
)
gateway = CalDAVGateway(settings)
gateway._calendar = mock_calendar
with caplog.at_level(logging.ERROR):
with pytest.raises(PronoteSyncError) as exc_info:
gateway.delete_event("test-uid")
# Vérifie que le message d'erreur ne contient pas les sentinelles
error_msg = str(exc_info.value)
assert SENTINEL_URL not in error_msg
assert SENTINEL_PASSWORD not in error_msg
# Vérifie que les logs ne contiennent pas les sentinelles
for record in caplog.records:
log_msg = record.getMessage()
assert SENTINEL_URL not in log_msg
assert SENTINEL_PASSWORD not in log_msg
# Vérifie que l'exception n'est chaînée à aucune exception brute
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_redacted_url_stored_in_gateway(self) -> None:
"""Vérifie que l'URL rédigée est stockée sur l'instance de la passerelle."""
settings = CalDAVSettings(
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
gateway = CalDAVGateway(settings)
# Vérifie que l'URL rédigée est stockée
assert gateway._redacted_url is not None
assert SENTINEL_URL not in gateway._redacted_url
assert "REDACTED" in gateway._redacted_url
# Vérifie que l'URL brute n'est pas stockée en clair
assert gateway._url_secret is not None
assert SENTINEL_URL not in str(gateway._url_secret)

640
tests/unit/test_diff.py Normal file
View File

@@ -0,0 +1,640 @@
"""Unit tests for AgendaComparator in pronote_sync/sync/diff.py."""
from __future__ import annotations
from datetime import date, datetime, time
from typing import override
import pytest
from pronote_sync.models.agenda import Lesson, LessonStatus, TheoreticalLesson
from pronote_sync.models.diff import AgendaChangeType
from pronote_sync.sources.theoretical.provider import TheoreticalAgendaProvider
from pronote_sync.sync.diff import AgendaComparator
class _StubProvider(TheoreticalAgendaProvider):
"""Stub implementation of TheoreticalAgendaProvider for testing."""
def __init__(self, lessons: list[TheoreticalLesson]) -> None:
"""Initialize with a fixed list of theoretical lessons."""
self._lessons = lessons
@override
def get_lessons(self, target_date: date) -> list[TheoreticalLesson]:
"""Return the stub lessons regardless of target_date."""
return self._lessons.copy()
@override
def get_lessons_for_range(self, start_date: date, end_date: date) -> list[TheoreticalLesson]:
"""Return the stub lessons regardless of date range."""
return self._lessons.copy()
# Target date: Monday, 2025-09-15 (weekday() = 0)
TARGET_DATE = date(2025, 9, 15)
@pytest.fixture(name="empty_provider")
def fixture_empty_provider() -> _StubProvider:
"""Provider with no theoretical lessons."""
return _StubProvider([])
@pytest.fixture(name="comparator")
def fixture_comparator(empty_provider: _StubProvider) -> AgendaComparator:
"""AgendaComparator with empty provider."""
return AgendaComparator(empty_provider)
# ==================== Test Case 1: Empty agendas ====================
def test_empty_agendas(comparator: AgendaComparator) -> None:
"""No real, no theoretical → AgendaDiff with empty changes."""
result = comparator.compare([], TARGET_DATE)
assert result.target_date == TARGET_DATE
assert result.changes == ()
# ==================== Test Case 2: Empty theoretical, real lessons present ====================
def test_empty_theoretical_real_present(comparator: AgendaComparator) -> None:
"""Empty theoretical, real lessons present → all real → ADDED."""
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
group=None,
content=None,
),
Lesson(
id="real_2",
start=datetime(2025, 9, 15, 14, 0, 0),
end=datetime(2025, 9, 15, 15, 0, 0),
subject="Français",
group=None,
content=None,
),
]
result = comparator.compare(real_lessons, TARGET_DATE)
assert len(result.changes) == 2
assert result.changes[0].type == AgendaChangeType.ADDED
assert result.changes[0].lesson == real_lessons[0]
assert result.changes[0].theoretical_lesson is None
assert result.changes[0].details == "Cours ajouté par rapport à l'agenda théorique"
assert result.changes[1].type == AgendaChangeType.ADDED
assert result.changes[1].lesson == real_lessons[1]
# ==================== Test Case 3: Empty real, theoretical present ====================
def test_empty_real_theoretical_present() -> None:
"""Empty real, theoretical present → all theoretical → REMOVED, sorted by id."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_b",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
TheoreticalLesson(
id="theo_a",
day_of_week=0,
start_time=time(14, 0),
end_time=time(15, 0),
subject="Français",
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare([], TARGET_DATE)
assert len(result.changes) == 2
assert result.changes[0].type == AgendaChangeType.REMOVED
assert result.changes[0].theoretical_lesson == theoretical_lessons[1] # theo_a first
assert result.changes[0].lesson is None
assert result.changes[0].details == "Cours supprimé par rapport à l'agenda théorique"
assert result.changes[1].type == AgendaChangeType.REMOVED
assert result.changes[1].theoretical_lesson == theoretical_lessons[0] # theo_b second
# ==================== Test Case 4: Exact match ====================
def test_exact_match() -> None:
"""Real and theoretical at same time, same subject → no changes."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert result.changes == ()
# ==================== Test Case 5: Within tolerance (±14 min) ====================
def test_within_tolerance_14min() -> None:
"""Real start 14 min before theoretical → match, MODIFIED (horaires different)."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 9, 46, 0),
end=datetime(2025, 9, 15, 10, 46, 0),
subject="Mathématiques",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert len(result.changes) == 1
assert result.changes[0].type == AgendaChangeType.MODIFIED
assert result.changes[0].lesson == real_lessons[0]
assert result.changes[0].theoretical_lesson == theoretical_lessons[0]
assert "horaires: 10:0011:00 → 09:4610:46" in result.changes[0].details
# ==================== Test Case 6: At tolerance boundary (exactly 15 min) ====================
def test_at_tolerance_boundary_15min() -> None:
"""Real start exactly 15 min from theoretical → match (inclusive), MODIFIED."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 9, 45, 0),
end=datetime(2025, 9, 15, 10, 45, 0),
subject="Mathématiques",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert len(result.changes) == 1
assert result.changes[0].type == AgendaChangeType.MODIFIED
assert result.changes[0].lesson == real_lessons[0]
assert result.changes[0].theoretical_lesson == theoretical_lessons[0]
assert "horaires: 10:0011:00 → 09:4510:45" in result.changes[0].details
# ==================== Test Case 7: Outside tolerance (16 min) ====================
def test_outside_tolerance_16min() -> None:
"""Real start 16 min from theoretical → no match → ADDED + REMOVED."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 9, 44, 0),
end=datetime(2025, 9, 15, 10, 44, 0),
subject="Mathématiques",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert len(result.changes) == 2
assert result.changes[0].type == AgendaChangeType.ADDED
assert result.changes[0].lesson == real_lessons[0]
assert result.changes[1].type == AgendaChangeType.REMOVED
assert result.changes[1].theoretical_lesson == theoretical_lessons[0]
# ==================== Test Case 8: Subject normalization match ====================
def test_subject_normalization_match() -> None:
"""Real '\\u212BNGSTRÖM' (angstrom sign), theoretical 'ångström' → NFKC → same form, match, no change."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="ångström",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="\u212bNGSTRÖM",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert result.changes == ()
# ==================== Test Case 9: Different normalized subjects ====================
def test_different_normalized_subjects() -> None:
"""Real 'Mathématiques', theoretical 'Français' → no match → ADDED + REMOVED."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Français",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert len(result.changes) == 2
assert result.changes[0].type == AgendaChangeType.ADDED
assert result.changes[0].lesson == real_lessons[0]
assert result.changes[1].type == AgendaChangeType.REMOVED
assert result.changes[1].theoretical_lesson == theoretical_lessons[0]
# ==================== Test Case 10: Multi-candidate selection by id ====================
def test_multi_candidate_selection_by_id() -> None:
"""Two theoretical candidates match one real → select the smaller id (theo_a).
theo_a (smaller id) has teachers identical to the real lesson (no MODIFIED);
theo_b (larger id) has different teachers and would trigger MODIFIED if selected.
A zero-change diff therefore proves theo_a was selected.
"""
theoretical_lessons = [
TheoreticalLesson(
id="theo_b",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
teachers=("Mme Martin",),
),
TheoreticalLesson(
id="theo_a",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
teachers=("M. Dupont",),
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
teachers=("M. Dupont",),
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
# theo_a (smaller id) selected with identical teachers → no MODIFIED; theo_b matched by existence → not REMOVED
assert result.changes == ()
# ==================== Test Case 11: MODIFIED — teachers differ (order-insensitive) ====================
def test_teachers_differ_order_insensitive() -> None:
"""Real teachers ('M. Dupont', 'Mme Martin'), theoretical ('Mme Martin', 'M. Dupont') → match, NOT modified (same set)."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
teachers=("Mme Martin", "M. Dupont"),
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
teachers=("M. Dupont", "Mme Martin"),
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert result.changes == ()
# ==================== Test Case 12: MODIFIED — teachers differ (different sets) ====================
def test_teachers_differ_different_sets() -> None:
"""Real ('M. Dupont',), theoretical ('Mme Martin',) → match, MODIFIED."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
teachers=("Mme Martin",),
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
teachers=("M. Dupont",),
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert len(result.changes) == 1
assert result.changes[0].type == AgendaChangeType.MODIFIED
assert "professeurs: {'Mme Martin'} → {'M. Dupont'}" in result.changes[0].details
# ==================== Test Case 13: MODIFIED — rooms differ ====================
def test_rooms_differ() -> None:
"""Real ('Salle 12',), theoretical ('Salle 15',) → match, MODIFIED."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
rooms=("Salle 15",),
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
rooms=("Salle 12",),
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert len(result.changes) == 1
assert result.changes[0].type == AgendaChangeType.MODIFIED
assert "salles: {'Salle 15'} → {'Salle 12'}" in result.changes[0].details
# ==================== Test Case 14: MODIFIED — status != NORMAL ====================
def test_status_not_normal() -> None:
"""Real status=CANCELLED, otherwise identical → match, MODIFIED with statut in details."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
status=LessonStatus.CANCELLED,
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
assert len(result.changes) == 1
assert result.changes[0].type == AgendaChangeType.MODIFIED
assert "statut: cancelled" in result.changes[0].details
# ==================== Test Case 15: REMOVED by existence, not selection ====================
def test_removed_by_existence_not_selection() -> None:
"""Two theoretical match one real; real selects the smaller id; the other theoretical is a candidate (exists) → NOT REMOVED."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_a",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
TheoreticalLesson(
id="theo_b",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
# Both theoretical lessons are candidates (matched by existence), so neither is REMOVED
assert len(result.changes) == 0
# ==================== Test Case 16: Deterministic order ====================
def test_deterministic_order() -> None:
"""Multiple ADDED, MODIFIED, REMOVED in same run → verify exact order (reals in input order, then theoreticals sorted by id)."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_c",
day_of_week=0,
start_time=time(15, 0),
end_time=time(16, 0),
subject="Histoire",
),
TheoreticalLesson(
id="theo_a",
day_of_week=0,
start_time=time(8, 0),
end_time=time(9, 0),
subject="Physique",
),
TheoreticalLesson(
id="theo_b",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
teachers=("M. Dupont",),
group=None,
content=None,
),
Lesson(
id="real_2",
start=datetime(2025, 9, 15, 14, 0, 0),
end=datetime(2025, 9, 15, 15, 0, 0),
subject="Informatique",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result = comparator.compare(real_lessons, TARGET_DATE)
# real_1 matches theo_b but has different teachers → MODIFIED
# real_2 has no match → ADDED
# theo_a and theo_c are not matched by existence → REMOVED (sorted by id: theo_a, theo_c)
assert len(result.changes) == 4
# First: real_1 MODIFIED
assert result.changes[0].type == AgendaChangeType.MODIFIED
assert result.changes[0].lesson == real_lessons[0]
# Second: real_2 ADDED
assert result.changes[1].type == AgendaChangeType.ADDED
assert result.changes[1].lesson == real_lessons[1]
# Third: theo_a REMOVED
assert result.changes[2].type == AgendaChangeType.REMOVED
assert result.changes[2].theoretical_lesson == theoretical_lessons[1] # theo_a
# Fourth: theo_c REMOVED
assert result.changes[3].type == AgendaChangeType.REMOVED
assert result.changes[3].theoretical_lesson == theoretical_lessons[0] # theo_c
# ==================== Test Case 17: Idempotence ====================
def test_idempotence() -> None:
"""Call compare twice with same inputs → identical AgendaDiff."""
theoretical_lessons = [
TheoreticalLesson(
id="theo_1",
day_of_week=0,
start_time=time(10, 0),
end_time=time(11, 0),
subject="Mathématiques",
),
]
real_lessons = [
Lesson(
id="real_1",
start=datetime(2025, 9, 15, 10, 0, 0),
end=datetime(2025, 9, 15, 11, 0, 0),
subject="Mathématiques",
group=None,
content=None,
),
]
provider = _StubProvider(theoretical_lessons)
comparator = AgendaComparator(provider)
result1 = comparator.compare(real_lessons, TARGET_DATE)
result2 = comparator.compare(real_lessons, TARGET_DATE)
assert result1 == result2

View File

@@ -0,0 +1,488 @@
"""Tests unitaires pour la sérialisation des modèles Pronote en VEVENT iCalendar.
Ce module vérifie que les fonctions de conversion des modèles Pronote
(:class:`Lesson`, :class:`Homework`, :class:`SchoolEvent`) en composants
:class:`icalendar.Event` produisent les propriétés attendues (UID, SUMMARY,
DTSTART, DTEND, STATUS, CATEGORIES, etc.) et que la signature sémantique
déterministe est correctement calculée.
"""
from __future__ import annotations
from datetime import date, datetime
from icalendar import Calendar, Event
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.models.homework import Homework
from pronote_sync.sync.serialization import (
MANAGED_PROPERTY,
MANAGED_VALUE,
component_to_signature,
homework_to_vevent,
lesson_to_vevent,
model_to_vcalendar_text,
school_event_to_vevent,
)
# --- Helper fixtures ---
def _make_lesson(
lesson_id: str = "L-1234",
subject: str = "Mathématiques",
start: datetime | None = None,
end: datetime | None = None,
status: LessonStatus = LessonStatus.NORMAL,
teachers: tuple[str, ...] = ("Prof Dupont",),
rooms: tuple[str, ...] = ("Salle 101",),
content: str | None = None,
group: str | None = None,
) -> Lesson:
"""Fabrique un cours Pronote pour les tests.
:param lesson_id: Identifiant du cours.
:param subject: Matière.
:param start: Date/heure de début.
:param end: Date/heure de fin.
:param status: Statut du cours.
:param teachers: Professeurs.
:param rooms: Salles.
:param content: Contenu pédagogique.
:param group: Groupe.
:return: Instance de Lesson.
:rtype: Lesson
"""
if start is None:
start = datetime(2026, 1, 15, 8, 0)
if end is None:
end = datetime(2026, 1, 15, 9, 0)
return Lesson(
id=lesson_id,
start=start,
end=end,
subject=subject,
teachers=teachers,
rooms=rooms,
status=status,
content=content,
group=group,
)
def _make_homework(
homework_id: str = "HW-5678",
subject: str = "Mathématiques",
due_on: date | None = None,
text: str = "Exercice 1 à 5",
teachers: tuple[str, ...] = ("Prof Dupont",),
assigned_on: date | None = None,
) -> Homework:
"""Fabrique un devoir Pronote pour les tests.
:param homework_id: Identifiant du devoir.
:param subject: Matière.
:param due_on: Date d'échéance.
:param text: Texte du devoir.
:param teachers: Professeurs.
:param assigned_on: Date de distribution.
:return: Instance de Homework.
:rtype: Homework
"""
if due_on is None:
due_on = date(2026, 1, 20)
return Homework(
id=homework_id,
subject=subject,
due_on=due_on,
text=text,
teachers=teachers,
assigned_on=assigned_on,
)
def _make_school_event(
label: str = "Vacances de Noël",
from_date: date | None = None,
to_date: date | None = None,
kind: SchoolEventKind = SchoolEventKind.HOLIDAY,
) -> SchoolEvent:
"""Fabrique un événement scolaire pour les tests.
:param label: Libellé de l'événement.
:param from_date: Date de début.
:param to_date: Date de fin.
:param kind: Type d'événement.
:return: Instance de SchoolEvent.
:rtype: SchoolEvent
"""
if from_date is None:
from_date = date(2026, 12, 20)
if to_date is None:
to_date = date(2027, 1, 5)
return SchoolEvent(
label=label,
from_date=from_date,
to_date=to_date,
kind=kind,
)
# --- lesson_to_vevent tests ---
def test_model_to_vcalendar_text_produces_complete_vcalendar() -> None:
"""Vérifie que la sérialisation produit un VCALENDAR complet avec un VEVENT.
:return: None
"""
lesson = _make_lesson()
output = model_to_vcalendar_text(lesson)
calendar = Calendar.from_ical(output)
assert "VERSION:2.0" in output
assert "PRODID" in output
assert output.startswith("BEGIN:VCALENDAR")
vevents = calendar.walk("VEVENT")
assert len(vevents) == 1
def test_lesson_to_vevent_normal() -> None:
"""Vérifie qu'un cours normal produit un VEVENT avec les bonnes propriétés.
Un cours normal doit avoir :
- UID = id du cours
- SUMMARY = matière
- DTSTART/DTEND = dates de début/fin
- STATUS = CONFIRMED
- CATEGORIES contient "Pronote"
- MANAGED_PROPERTY présent avec MANAGED_VALUE
"""
lesson = _make_lesson()
event = lesson_to_vevent(lesson)
assert str(event.get("UID")) == "L-1234"
assert str(event.get("SUMMARY")) == "Mathématiques"
assert event.get("DTSTART").dt == datetime(2026, 1, 15, 8, 0)
assert event.get("DTEND").dt == datetime(2026, 1, 15, 9, 0)
assert str(event.get("STATUS")) == "CONFIRMED"
categories = event.get("CATEGORIES")
assert categories is not None
assert "Pronote" in categories.cats
assert str(event.get(MANAGED_PROPERTY)) == MANAGED_VALUE
def test_lesson_to_vevent_cancelled() -> None:
"""Vérifie qu'un cours annulé a STATUS=CANCELLED et CATEGORIES contient 'Annulé'.
:return: None
"""
lesson = _make_lesson(status=LessonStatus.CANCELLED)
event = lesson_to_vevent(lesson)
assert str(event.get("STATUS")) == "CANCELLED"
categories = event.get("CATEGORIES")
assert categories is not None
assert "Pronote" in categories.cats
assert "Annulé" in categories.cats
def test_lesson_to_vevent_moved() -> None:
"""Vérifie qu'un cours déplacé a CATEGORIES contient 'Déplacé'.
:return: None
"""
lesson = _make_lesson(status=LessonStatus.MOVED)
event = lesson_to_vevent(lesson)
assert str(event.get("STATUS")) == "CONFIRMED"
categories = event.get("CATEGORIES")
assert categories is not None
assert "Pronote" in categories.cats
assert "Déplacé" in categories.cats
def test_lesson_to_vevent_description() -> None:
"""Vérifie que la description contient tous les champs renseignés.
:return: None
"""
lesson = _make_lesson(
subject="Mathématiques",
teachers=("Prof Dupont", "Prof Martin"),
rooms=("Salle 101", "Salle 102"),
content="Chapitre 1",
)
event = lesson_to_vevent(lesson)
description = str(event.get("DESCRIPTION"))
assert "Matière: Mathématiques" in description
assert "Professeur(s): Prof Dupont, Prof Martin" in description
assert "Salle(s): Salle 101, Salle 102" in description
assert "Contenu: Chapitre 1" in description
# --- homework_to_vevent tests ---
def test_homework_to_vevent_uid_prefix() -> None:
"""Vérifie que l'UID d'un devoir est préfixé par 'homework-'.
:return: None
"""
homework = _make_homework()
event = homework_to_vevent(homework)
assert str(event.get("UID")) == "homework-HW-5678"
def test_homework_to_vevent_status() -> None:
"""Vérifie qu'un devoir a STATUS=NEEDS-ACTION.
:return: None
"""
homework = _make_homework()
event = homework_to_vevent(homework)
assert str(event.get("STATUS")) == "NEEDS-ACTION"
def test_homework_to_vevent_categories() -> None:
"""Vérifie que les CATEGORIES d'un devoir contiennent 'Pronote' et 'Devoir'.
:return: None
"""
homework = _make_homework()
event = homework_to_vevent(homework)
categories = event.get("CATEGORIES")
assert categories is not None
assert "Pronote" in categories.cats
assert "Devoir" in categories.cats
def test_homework_to_vevent_dtstart_dtend() -> None:
"""Vérifie que DTSTART et DTEND couvrent la journée d'échéance (08:00-18:00).
:return: None
"""
homework = _make_homework(due_on=date(2026, 1, 20))
event = homework_to_vevent(homework)
assert event.get("DTSTART").dt == datetime(2026, 1, 20, 8, 0)
assert event.get("DTEND").dt == datetime(2026, 1, 20, 18, 0)
def test_homework_to_vevent_summary() -> None:
"""Vérifie que le SUMMARY d'un devoir est préfixé par 'Devoir: '.
:return: None
"""
homework = _make_homework(subject="Mathématiques")
event = homework_to_vevent(homework)
assert str(event.get("SUMMARY")) == "Devoir: Mathématiques"
def test_homework_to_vevent_managed_marker() -> None:
"""Vérifie que le marqueur MANAGED_PROPERTY est présent.
:return: None
"""
homework = _make_homework()
event = homework_to_vevent(homework)
assert str(event.get(MANAGED_PROPERTY)) == MANAGED_VALUE
# --- school_event_to_vevent tests ---
def test_school_event_to_vevent_uid_prefix() -> None:
"""Vérifie que l'UID d'un événement scolaire est préfixé correctement.
:return: None
"""
school_event = _make_school_event(
label="Vacances de Noël",
from_date=date(2026, 12, 20),
)
event = school_event_to_vevent(school_event)
assert str(event.get("UID")) == "school-event-Vacances de Noël-2026-12-20"
def test_school_event_to_vevent_status() -> None:
"""Vérifie qu'un événement scolaire a STATUS=CONFIRMED.
:return: None
"""
school_event = _make_school_event()
event = school_event_to_vevent(school_event)
assert str(event.get("STATUS")) == "CONFIRMED"
def test_school_event_to_vevent_categories() -> None:
"""Vérifie que les CATEGORIES contiennent 'Pronote' et la valeur du kind.
:return: None
"""
school_event = _make_school_event(kind=SchoolEventKind.HOLIDAY)
event = school_event_to_vevent(school_event)
categories = event.get("CATEGORIES")
assert categories is not None
assert "Pronote" in categories.cats
assert "holiday" in categories.cats
def test_school_event_to_vevent_dtstart_dtend() -> None:
"""Vérifie que DTSTART et DTEND sont des vDate (pas vDatetime).
:return: None
"""
school_event = _make_school_event(
from_date=date(2026, 12, 20),
to_date=date(2027, 1, 5),
)
event = school_event_to_vevent(school_event)
from icalendar import vDate
assert isinstance(event.get("DTSTART"), vDate)
assert isinstance(event.get("DTEND"), vDate)
assert event.get("DTSTART").dt == date(2026, 12, 20)
assert event.get("DTEND").dt == date(2027, 1, 5)
def test_school_event_to_vevent_managed_marker() -> None:
"""Vérifie que le marqueur MANAGED_PROPERTY est présent.
:return: None
"""
school_event = _make_school_event()
event = school_event_to_vevent(school_event)
assert str(event.get(MANAGED_PROPERTY)) == MANAGED_VALUE
# --- component_to_signature tests ---
def test_component_to_signature_ignores_volatile_properties() -> None:
"""Vérifie que deux VEVENTs ne différant que par DTSTAMP/CREATED/LAST-MODIFIED/SEQUENCE
produisent la même signature.
:return: None
"""
# Créer deux événements identiques sauf pour les propriétés volatiles
event1 = Event()
event1.add("UID", "test-uid")
event1.add("SUMMARY", "Test Event")
event1.add("DTSTART", datetime(2026, 1, 15, 8, 0))
event1.add("DTEND", datetime(2026, 1, 15, 9, 0))
event1.add("STATUS", "CONFIRMED")
event1.add("DTSTAMP", datetime(2026, 1, 1, 0, 0)) # Différent
event1.add("CREATED", datetime(2026, 1, 1, 0, 0)) # Différent
event2 = Event()
event2.add("UID", "test-uid")
event2.add("SUMMARY", "Test Event")
event2.add("DTSTART", datetime(2026, 1, 15, 8, 0))
event2.add("DTEND", datetime(2026, 1, 15, 9, 0))
event2.add("STATUS", "CONFIRMED")
event2.add("DTSTAMP", datetime(2026, 1, 2, 0, 0)) # Différent
event2.add("LAST-MODIFIED", datetime(2026, 1, 2, 0, 0)) # Différent
event2.add("SEQUENCE", 1) # Différent
sig1 = component_to_signature(event1)
sig2 = component_to_signature(event2)
assert sig1 == sig2
def test_component_to_signature_different_summary() -> None:
"""Vérifie que deux VEVENTs avec SUMMARY différent produisent des signatures différentes.
:return: None
"""
event1 = Event()
event1.add("UID", "test-uid")
event1.add("SUMMARY", "Event 1")
event1.add("DTSTART", datetime(2026, 1, 15, 8, 0))
event1.add("DTEND", datetime(2026, 1, 15, 9, 0))
event1.add("STATUS", "CONFIRMED")
event2 = Event()
event2.add("UID", "test-uid")
event2.add("SUMMARY", "Event 2") # Différent
event2.add("DTSTART", datetime(2026, 1, 15, 8, 0))
event2.add("DTEND", datetime(2026, 1, 15, 9, 0))
event2.add("STATUS", "CONFIRMED")
sig1 = component_to_signature(event1)
sig2 = component_to_signature(event2)
assert sig1 != sig2
def test_component_to_signature_deterministic() -> None:
"""Vérifie que la signature est déterministe (même entrée → même sortie).
:return: None
"""
event = Event()
event.add("UID", "test-uid")
event.add("SUMMARY", "Test Event")
event.add("DTSTART", datetime(2026, 1, 15, 8, 0))
event.add("DTEND", datetime(2026, 1, 15, 9, 0))
event.add("STATUS", "CONFIRMED")
sig1 = component_to_signature(event)
sig2 = component_to_signature(event)
assert sig1 == sig2
def test_component_to_signature_includes_categories() -> None:
"""Vérifie que les CATEGORIES sont incluses dans la signature (triées).
:return: None
"""
event = Event()
event.add("UID", "test-uid")
event.add("SUMMARY", "Test Event")
event.add("DTSTART", datetime(2026, 1, 15, 8, 0))
event.add("DTEND", datetime(2026, 1, 15, 9, 0))
event.add("STATUS", "CONFIRMED")
event.add("CATEGORIES", ["Pronote", "Devoir"])
sig = component_to_signature(event)
# Les catégories doivent apparaître dans la signature
assert "categories=devoir,pronote" in sig
def test_component_to_signature_includes_managed_property() -> None:
"""Vérifie que MANAGED_PROPERTY est incluse dans la signature.
:return: None
"""
event = Event()
event.add("UID", "test-uid")
event.add("SUMMARY", "Test Event")
event.add("DTSTART", datetime(2026, 1, 15, 8, 0))
event.add("DTEND", datetime(2026, 1, 15, 9, 0))
event.add("STATUS", "CONFIRMED")
event.add(MANAGED_PROPERTY, MANAGED_VALUE)
sig = component_to_signature(event)
assert f"managed={MANAGED_VALUE.lower()}" in sig
# Ensure trailing newline

View File

@@ -13,7 +13,7 @@ from pathlib import Path
import pytest import pytest
from pronote_sync.errors import PronoteSyncError from pronote_sync.errors import PronoteSyncError
from pronote_sync.sources.theoretical.file import JsonTheoreticalAgendaProvider from pronote_sync.sources.theoretical.file import JsonTheoreticalAgendaProvider, normalize_subject
from pronote_sync.sources.theoretical.holidays import SchoolHolidayCalendar from pronote_sync.sources.theoretical.holidays import SchoolHolidayCalendar
from pronote_sync.sources.theoretical.parity import WeekParityService from pronote_sync.sources.theoretical.parity import WeekParityService
@@ -232,10 +232,10 @@ class TestJsonTheoreticalAgendaProvider:
lessons2 = provider_no_parity_no_holidays.get_lessons(target_date) lessons2 = provider_no_parity_no_holidays.get_lessons(target_date)
assert lessons1 == lessons2 assert lessons1 == lessons2
def test_even_odd_same_slot_different_ids( def test_even_odd_different_subjects_different_ids(
self, provider_with_parity: JsonTheoreticalAgendaProvider self, provider_with_parity: JsonTheoreticalAgendaProvider
) -> None: ) -> None:
"""Teste que les cours even/odd sur le même créneau ont des IDs différents. """Teste que les cours even/odd sur le même créneau avec des matières différentes ont des IDs différents.
:assert: Les IDs des cours even et odd sont différents. :assert: Les IDs des cours even et odd sont différents.
""" """
@@ -257,6 +257,59 @@ class TestJsonTheoreticalAgendaProvider:
odd_ids = {lesson.id for lesson in odd_lessons} odd_ids = {lesson.id for lesson in odd_lessons}
assert even_ids.isdisjoint(odd_ids) assert even_ids.isdisjoint(odd_ids)
def test_even_odd_same_subject_different_ids(self, tmp_path: Path) -> None:
"""Vérifie que les leçons paire/impaire sur le même créneau avec la même matière ont des IDs distincts.
Utilise la même matière pour isoler la parité comme seul différenciateur.
:assert: Les IDs des cours even et odd sont différents, avec la même matière.
"""
json_content = json.dumps(
{
"version": 1,
"lessons": [
{
"week": "even",
"day_of_week": 1,
"start_time": "10:00",
"end_time": "11:00",
"subject": "Langue vivante",
"teachers": [],
"rooms": [],
},
{
"week": "odd",
"day_of_week": 1,
"start_time": "10:00",
"end_time": "11:00",
"subject": "Langue vivante",
"teachers": [],
"rooms": [],
},
],
}
)
file_path = tmp_path / "theoretical.json"
file_path.write_text(json_content, encoding="utf-8")
anchor_date = date(2026, 9, 1) # Tuesday
parity = WeekParityService(anchor_date, "even")
provider = JsonTheoreticalAgendaProvider(file_path=str(file_path), parity_service=parity)
# Tuesday in even week
even_tuesday = date(2026, 9, 1) # Same week as anchor (even)
odd_tuesday = date(2026, 9, 8) # One week later (odd)
even_lessons = provider.get_lessons(even_tuesday)
odd_lessons = provider.get_lessons(odd_tuesday)
assert len(even_lessons) == 1
assert len(odd_lessons) == 1
assert even_lessons[0].id != odd_lessons[0].id
# The only difference in the ID should be the week type
assert "even" in even_lessons[0].id
assert "odd" in odd_lessons[0].id
def test_explicit_id_preserved( def test_explicit_id_preserved(
self, provider_no_parity_no_holidays: JsonTheoreticalAgendaProvider self, provider_no_parity_no_holidays: JsonTheoreticalAgendaProvider
) -> None: ) -> None:
@@ -589,3 +642,81 @@ class TestJsonTheoreticalAgendaProvider:
# Seule la leçon du 15 octobre (jeudi) devrait être retournée # Seule la leçon du 15 octobre (jeudi) devrait être retournée
assert len(lessons) == 1 assert len(lessons) == 1
assert lessons[0].subject == "Sciences" assert lessons[0].subject == "Sciences"
def test_normalize_subject_variants_produce_same_id(self, tmp_path: Path) -> None:
"""Vérifie que des variantes de casse, d'espacement et d'Unicode produisent le même ID.
:assert: Les variantes de la même matière produisent le même ID normalisé.
"""
# Test the normalize_subject function directly
# Note: hyphens are removed entirely (not replaced with spaces) by normalize_subject
variants = [
"Mathématiques avancées",
"mathématiques avancées",
"Mathématiques avancées",
"MATHÉMATIQUES AVANCÉES",
]
normalized = [normalize_subject(variant) for variant in variants]
# All should normalize to the same value
assert all(n == normalized[0] for n in normalized)
# Should be lowercase, no extra spaces, no punctuation
assert normalized[0] == "mathématiques avancées"
def test_collision_error_no_secret_leak(
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Vérifie qu'une sentinelle dans un ID dupliqué n'apparaît pas dans l'erreur.
:assert: PronoteSyncError est levée et la sentinelle n'apparaît pas dans l'erreur.
"""
# Use a secret pattern that the redaction system will actually catch
# The _ISOLATED_SECRET_PATTERN looks for things like secret=value or secret: value
sentinel_value = "SENTINELLE_M6_SECRET"
# Create an ID that contains a pattern like "secret=value" which will be redacted
secret_pattern = f"secret={sentinel_value}"
# Créer un fichier JSON avec 2 entrées ayant le même ID explicite contenant la sentinelle
data = {
"version": 1,
"lessons": [
{
"id": f"theoretical-{secret_pattern}-1",
"week": "all",
"day_of_week": 0,
"start_time": "08:00",
"end_time": "09:00",
"subject": "Maths",
"teachers": [],
"rooms": [],
},
{
"id": f"theoretical-{secret_pattern}-1",
"week": "all",
"day_of_week": 1,
"start_time": "09:00",
"end_time": "10:00",
"subject": "Français",
"teachers": [],
"rooms": [],
},
],
}
file_path = tmp_path / "collision_test.json"
file_path.write_text(json.dumps(data), encoding="utf-8")
with caplog.at_level("ERROR"):
with pytest.raises(PronoteSyncError) as exc_info:
JsonTheoreticalAgendaProvider(
file_path=str(file_path),
parity_service=None,
holiday_calendar=None,
)
# Check that the sentinel value does not appear in the error message
assert sentinel_value not in str(exc_info.value)
# Check that the sentinel value does not appear in the logs
assert sentinel_value not in caplog.text
# Check that the secret pattern was redacted (should contain REDACTED)
assert "REDACTED" in str(exc_info.value)
# Check that the sentinel does not appear in the cause
if exc_info.value.__cause__ is not None:
assert sentinel_value not in repr(exc_info.value.__cause__)

View File

@@ -0,0 +1,189 @@
"""Tests unitaires pour l'équivalence des UIDs entre les sources iCal et pronotepy.
Ce module vérifie que les fonctions de normalisation et de génération
d'UIDs produisent les mêmes résultats pour les mêmes données, garantissant
qu'aucun doublon ou suppression artificielle ne se produit lors du
changement de source.
"""
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from icalendar import Calendar
from pronote_sync.utils.uid import generate_deterministic_uid, normalize_pronote_uid
# --- Test 1: normalize_pronote_uid strips temporal suffix ---
def test_normalize_pronote_uid_strips_temporal_suffix() -> None:
"""Vérifie que normalize_pronote_uid produit un UID canonique stable.
Un UID brut Pronote avec suffixe temporel comme
``L-12345-20260905T120000Z-Index-Education`` doit être normalisé en
``L-12345``.
:return: None
"""
raw_uid = "L-12345-20260905T120000Z-Index-Education"
normalized = normalize_pronote_uid(raw_uid)
assert normalized == "L-12345"
def test_normalize_pronote_uid_strips_education_suffix_only() -> None:
"""Vérifie que normalize_pronote_uid retire aussi le suffixe -Index-Education seul.
:return: None
"""
raw_uid = "L-12345-Index-Education"
normalized = normalize_pronote_uid(raw_uid)
assert normalized == "L-12345"
def test_normalize_pronote_uid_idempotent() -> None:
"""Vérifie que normalize_pronote_uid est idempotente.
:return: None
"""
raw_uid = "L-12345"
normalized = normalize_pronote_uid(raw_uid)
assert normalized == "L-12345"
# --- Test 2: generate_deterministic_uid is stable ---
def test_generate_deterministic_uid_is_stable() -> None:
"""Vérifie que generate_deterministic_uid produit le même UID pour les mêmes entrées.
:return: None
"""
start = datetime(2026, 9, 5, 8, 0, 0)
end = datetime(2026, 9, 5, 9, 0, 0)
subject = "Mathématiques"
teachers = ["M. Dupont"]
rooms = ["204"]
group = "Classe entière"
uid1 = generate_deterministic_uid(
start=start, end=end, subject=subject, teachers=teachers, rooms=rooms, group=group
)
uid2 = generate_deterministic_uid(
start=start, end=end, subject=subject, teachers=teachers, rooms=rooms, group=group
)
assert uid1 == uid2
assert len(uid1) == 12
assert all(c in "0123456789abcdef" for c in uid1)
def test_generate_deterministic_uid_ignores_order() -> None:
"""Vérifie que l'ordre des enseignants/salles n'affecte pas le résultat.
:return: None
"""
start = datetime(2026, 9, 5, 8, 0, 0)
end = datetime(2026, 9, 5, 9, 0, 0)
subject = "Mathématiques"
uid1 = generate_deterministic_uid(
start=start,
end=end,
subject=subject,
teachers=["M. Dupont", "Mme Martin"],
rooms=["204", "205"],
group=None,
)
uid2 = generate_deterministic_uid(
start=start,
end=end,
subject=subject,
teachers=["Mme Martin", "M. Dupont"],
rooms=["205", "204"],
group=None,
)
assert uid1 == uid2
# --- Test 3: iCal and pronotepy produce same UID with raw UID ---
def test_ical_and_pronotepy_produce_same_uid_with_raw_uid() -> None:
"""Vérifie que iCal et pronotepy produisent le même UID avec un UID brut.
Comme les deux sources utilisent normalize_pronote_uid, le résultat doit
être identique.
:return: None
"""
raw_uid = "Edt_12345@index-education.net-20260905T120000Z-Index-Education"
# Simulation du chemin iCal
ical_normalized = normalize_pronote_uid(raw_uid)
# Simulation du chemin pronotepy
pronotepy_normalized = normalize_pronote_uid(raw_uid)
assert ical_normalized == pronotepy_normalized
assert ical_normalized == "Edt_12345@index-education.net"
# --- Test 4: iCal and pronotepy produce same UID with fallback ---
def test_ical_and_pronotepy_produce_same_uid_with_fallback() -> None:
"""Vérifie que iCal et pronotepy produisent le même UID déterministe en l'absence d'UID brut.
:return: None
"""
start = datetime(2026, 9, 5, 8, 0, 0)
end = datetime(2026, 9, 5, 9, 0, 0)
subject = "Mathématiques"
teachers = ["M. Dupont"]
rooms = ["204"]
group = "Classe entière"
# Simulation du chemin iCal (parse_ical utilise generate_deterministic_uid)
ical_uid = generate_deterministic_uid(
start=start, end=end, subject=subject, teachers=teachers, rooms=rooms, group=group
)
# Simulation du chemin pronotepy (get_lessons utilise generate_deterministic_uid)
pronotepy_uid = generate_deterministic_uid(
start=start, end=end, subject=subject, teachers=teachers, rooms=rooms, group=group
)
assert ical_uid == pronotepy_uid
# --- Test 5: UID from iCal fixture is normalized ---
def test_uid_from_ical_fixture_is_normalized() -> None:
"""Vérifie que les UIDs extraits du fixture pronote-4e.ics sont normalisés correctement.
:return: None
"""
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
with open(fixture_path, "rb") as f:
cal = Calendar.from_ical(f.read())
normalized_uids = []
for component in cal.walk("VEVENT"):
raw_uid = str(component.get("UID"))
normalized = normalize_pronote_uid(raw_uid)
normalized_uids.append(normalized)
# Vérifie que tous les UIDs sont normalisés (plus courts que les UIDs bruts)
assert len(normalized_uids) >= 2
# Vérifie que les UIDs normalisés ne contiennent pas de suffixes temporels
for uid in normalized_uids:
assert "-Index-Education" not in uid
assert "-2026" not in uid # Pas de suffixe temporel
# Vérifie que le premier UID est bien normalisé
assert "Edt_12345@index-education.net" in normalized_uids