Compare commits

..

12 Commits

Author SHA1 Message Date
bb1f90bf5f test(M4): couverture fallback.py 81→98% (11 tests ciblés)
- _fetch_agenda_ical sans ical_url → ValueError
- _agenda_sources AUTO pronotepy seul, _homework_sources modes
  explicites + AUTO + aucune source configurée
- fetch_agenda/fetch_homework : repli échoue → PipelineCriticalError,
  repli retourne [] → warning d'ambiguité
- fetch_informations : log redact_exception + re-raise sans secrets
- Couverture pronote_sync/sources/pronote/ : 95% (≥90% requis)

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

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

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

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

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-06 15:23:34 +02:00
e0cd073f9a test(M4): tests unitaires iCal, client pronotepy et repli
- test_ical.py : 20 tests (fetch file://, parsing VEVENT, détection
  cours annulé, collect_homeworks avec déduplication, normalisation,
  génération d'ID déterministe, redaction des erreurs)
- test_pronote_client.py : 9 tests (Protocol, messages mockés,
  informations, agenda fallback, credentials manquants, sécurité
  mot de passe)
- test_fallback.py : 16 tests (modes ICAL/PRONOTEPY/AUTO, repli,
  PipelineCriticalError si les deux sources échouent, redaction
  des erreurs)
- pre-commit : ajout de responses et pytest-mock au hook mypy
- Total : 42 nouveaux tests M4 (122 tests au total sur le projet)

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

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

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

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

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 13:20:24 +02:00
292aef4f5b fix(M3): immuabilité et validation des modèles — correction des écarts FIXME_M3
Immuabilité :
- Collections des modèles frozen : list → tuple (Lesson, TheoreticalLesson,
  SchoolEvent, AgendaDiff, Homework, XmppMessage, ExternalInfo)
- Gel de HomeworkBlock et ExternalInfo (sous-modèles imbriqués dans contrats)
- Valeurs par défaut immuables : Field(default=(), ...)

Validation de domaine :
- TheoreticalLesson.day_of_week : Field(ge=0, le=6)
- Lesson/TheoreticalLesson : start < end (validateur de modèle)
- SchoolEvent : from_date <= to_date
- AgendaChange : cohérence type/payload (ADDED/MODIFIED → lesson,
  REMOVED → theoretical_lesson)
- CalDAVSyncResult : compteurs Field(ge=0) sur added/updated/removed

Tests : 66 tests (construction/sérialisation, immuabilité, invariants)
couvrant le gel, la mutabilité légitime, et les cas invalides par invariant.

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 12:36:27 +02:00
2f6184df0b fix(M2): configuration, secrets et rédaction — correction des écarts FIXME_M2
Sécurité :
- ical_url : str → SecretStr | None (masquage dans str/repr/model_dump_json)
- En-têtes Authorization/Proxy-Authorization : masquage complet de la valeur
- _URL_PATTERN : insensible à la casse (HTTPS:// reconnu)
- redact_url() : masquage du nom d'utilisateur (userinfo complet)

Configuration :
- Sous-configs : Field(default_factory=...) pour rechargement à chaque appel
- Suppression du singleton settings (injection de dépendances)
- .env.example : ajout AI_PROVIDER et CALDAV_CALENDAR_PATH
- Guide : BLOG_RSS_ENABLED → BLOG_ENABLED, AI_MODEL=None par défaut
- .secrets.baseline : ligne décalée 5112 → 5117 (faux positif audité)
- pre-commit : ajout de pytest aux additional_dependencies du hook mypy

Tests : 15 tests (config + redaction) couvrant rechargement, masquage
SecretStr, en-têtes Authorization, URL auth intégrée et casse variable.

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-06 12:35:55 +02:00
a0ac75cec6 fix(M1): correction des écarts d'audit FIXME_M1
- requires-python : >=3.13 → >=3.13.5 (alignement sur le contrat TODO.md)
- test de fumée : ajout de tests/unit/test_smoke.py (pytest sort code 0)
- .gitignore : ajout de la règle *.tmp.ics pour les artefacts iCal temporaires
- .secrets.baseline : audit du faux positif GUIDE_DEV_PYTHON.md:5112
  (is_secret: false, is_verified: true → VERIFIED_FALSE)

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/verifier <verifier@agents.invalid>
2026-09-06 11:59:50 +02:00
aaca78c55d docs+fix: renforcement de l'architecture agentique et corrections de sécurité
AGENTS.md :
- Rôles d'agents renforcés : @coder ne valide pas, @debugger ne code pas,
  @verifier ne modifie pas, etc.
- Table « Séparation des rôles » : tâche → agent responsable → ne pas confier à
- Workflow étape 5 : délégation explicite à @verifier pour la validation

GUIDE_DEV_PYTHON.md :
- 7 notes « Décision d'implémentation » ajoutées aux sections concernées :
  3.1.1 (XMPP_RECIPIENT→XMPP_TO, vars ajoutées), 3.1.2 (SYNC_PAST_DAYS→AppSettings),
  3.2 (Pydantic v2 style, defaults corrigés), 4.2.1 (redact_exception module function),
  4.2.2 (getLevelNamesMapping), 5.1.5 (normalize_pronote_uid, usedforsecurity=False),
  6 (ConfigDict, StrEnum, alias _date, external_info Optional)

Sécurité (audit @security-auditor, corrections @coder, validation @verifier) :
- RedactingFormatter : redaction APRÈS formatage (corrige TypeError %s + fuite traceback)
- redact_url : masquage des credentials dans userinfo URL (HTTP Basic Auth)
- redact_secrets : patterns étendus (api_key, access_token, authorization, auth)
- redact_secrets : support JSON-style « key: value » avec guillemets

Validations (@verifier) :
- ruff check : PASS | mypy strict : PASS | bandit : PASS (0 issue)
- %s formatting : OK (password=REDACTED, pas de TypeError)
- Traceback redaction : OK (icalsecurise=REDACTED)
- URL userinfo : OK (user:REDACTED@host)
- JSON-style redaction : OK ({"token": "REDACTED"})
- Régression red-to-green : OK (historical HEAD reproduction)

Co-authored-by: OpenCode/orchestrator <opencode-orchestrator@agents.invalid>
2026-09-05 23:52:02 +02:00
32 changed files with 5430 additions and 595 deletions

View File

@@ -1,5 +1,7 @@
# --- Pronote --- # --- Pronote ---
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024 PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
PRONOTE_URL=https://college.ent/pronote/eleve.html
PRONOTE_ACCOUNT_TYPE=parent
PRONOTE_USERNAME=parent.dupont PRONOTE_USERNAME=parent.dupont
PRONOTE_PASSWORD=your_secure_password PRONOTE_PASSWORD=your_secure_password
PRONOTE_ENT=monbureaunumerique PRONOTE_ENT=monbureaunumerique
@@ -13,6 +15,7 @@ PRONOTE_MESSAGES_SOURCE=pronotepy
CALDAV_URL=https://caldav.example.com/calendars/user/pronote/ 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/
# Fenêtre de synchronisation (jours) # Fenêtre de synchronisation (jours)
SYNC_PAST_DAYS=7 SYNC_PAST_DAYS=7
@@ -34,9 +37,10 @@ XMPP_TIMEOUT=30
# --- IA (optionnelle) --- # --- IA (optionnelle) ---
AI_ENABLED=true AI_ENABLED=true
AI_PROVIDER=openai
AI_BASE_URL=https://api.openai.com/v1 AI_BASE_URL=https://api.openai.com/v1
AI_API_KEY=your_ai_api_key AI_API_KEY=your_ai_api_key
AI_MODEL=gpt-4o-mini # AI_MODEL=gpt-4o-mini # exemple recommandé, non activé par défaut
# --- Blog --- # --- Blog ---
BLOG_ENABLED=false BLOG_ENABLED=false

5
.gitignore vendored
View File

@@ -51,3 +51,8 @@ Thumbs.db
# --- Logs --- # --- Logs ---
*.log *.log
# --- Artefacts iCal temporaires ---
# Ignorer les exports iCal temporaires (peuvent contenir des jetons ou données scolaires).
# Les fixtures iCal anonymisées (M13) restent suivies car elles ne correspondent pas au suffixe *.tmp.ics.
*.tmp.ics

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"] additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.0", "types-requests>=2.31.0", "icalendar>=5.0.0", "pronotepy>=2.15.0", "responses>=0.25.0", "pytest-mock>=3.10.0"]
types: [python] types: [python]
pass_filenames: true pass_filenames: true

View File

@@ -90,6 +90,10 @@
{ {
"path": "detect_secrets.filters.allowlist.is_line_allowlisted" "path": "detect_secrets.filters.allowlist.is_line_allowlisted"
}, },
{
"path": "detect_secrets.filters.common.is_baseline_file",
"filename": ".secrets.baseline"
},
{ {
"path": "detect_secrets.filters.common.is_ignored_due_to_verification_policies", "path": "detect_secrets.filters.common.is_ignored_due_to_verification_policies",
"min_level": 2 "min_level": 2
@@ -135,10 +139,11 @@
"type": "Hex High Entropy String", "type": "Hex High Entropy String",
"filename": "GUIDE_DEV_PYTHON.md", "filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": false, "is_verified": true,
"line_number": 5073 "line_number": 5124,
"is_secret": false
} }
] ]
}, },
"generated_at": "2026-09-05T17:53:24Z" "generated_at": "2026-09-06T13:36:29Z"
} }

View File

@@ -189,22 +189,38 @@ Cette section s'applique uniquement lorsque le travail est exécuté avec le sys
Les rôles d'agents disponibles pour ce projet sont les suivants : Les rôles d'agents disponibles pour ce projet sont les suivants :
- `@architect` : Arbitrages d'architecture et choix techniques structurants pour le pipeline `pronote-sync`. - `@architect` : Arbitrages d'architecture et choix techniques structurants. **Ne produit pas de code.**
- `@coder` : Opérations de développement et changements de code dans le projet. - `@coder` : Écrit et modifie du code, de la configuration et des scripts. **Ne valide pas** (ruff, mypy, pytest) — c'est le rôle de `@verifier`. **Ne diagnostique pas** — c'est le rôle de `@debugger`.
- `@debugger` : Reproduction d'un symptôme et établissement de sa cause profonde (ex. : échec de synchronisation, repli iCal/pronotepy). - `@debugger` : Reproduit un symptôme et établit sa cause profonde. **Ne modifie pas le code.**
- `@explorer` : Exploration du dépôt en lecture seule et fourniture de contexte factuel. - `@explorer` : Explore le dépôt en lecture seule. **Ne modifie rien, n'exécute pas de commandes.**
- `@orchestrator` : Compréhension globale du projet, définition des jalons, coordination et garantie du résultat. - `@orchestrator` : Compréhension globale, définition des jalons, coordination et garantie du résultat. **N'écrit pas de code.**
- `@planner` : Transformation d'une demande complexe en unités exécutables avec frontières et dépendances claires. - `@planner` : Transforme une demande complexe en unités exécutables. **Ne dirige aucun technicien.**
- `@reviewer` : Revues indépendantes de correction, régression, contrats et maintenabilité. - `@reviewer` : Revues indépendantes de correction, régression, contrats et maintenabilité. **Ne modifie pas le code.**
- `@security-auditor` : Audit indépendant d'une surface de sécurité désignée (ex. : gestion des secrets, masquage des données). - `@security-auditor` : Audit indépendant d'une surface de sécurité. **Ne modifie pas le code.**
- `@tech-writer` : Rédaction et maintenance de documentation technique exacte et vérifiable. - `@tech-writer` : Rédige et maintient la documentation. **N'écrit pas de code applicatif.**
- `@test-engineer` : Conception, écriture et exécution de tests ciblés (unitaires, intégration, mocks). - `@test-engineer` : Conçoit, écrit et exécute des tests ciblés. **N'écrit pas de code de production.**
- `@ui-designer` : conception et implémentation d'interfaces Web et terminal. - `@ui-designer` : Conçoit et implémente les interfaces Web et terminal.
- `@verifier` : Vérification indépendante du comportement livré, des régressions et du respect des conventions (idempotence, mode dégradé). - `@verifier` : Vérifie indépendamment le comportement livré, les régressions et le respect des conventions (ruff, mypy, pytest, bandit, idempotence, mode dégradé). **Ne modifie pas le code.**
- `@web-explorer` : Recherche et extraction de sources Web vérifiables (ex. : documentation Pronote, CalDAV, XMPP). - `@web-explorer` : Recherche et extrait des sources Web vérifiables. **Ne modifie pas le dépôt.**
> **Note** : Ne pas utiliser `@coder` pour les tâches de documentation (`@tech-writer`) ni pour les tests (`@test-engineer`). > **Note** : Ne pas utiliser `@coder` pour les tâches de documentation (`@tech-writer`) ni pour les tests (`@test-engineer`).
### Séparation des rôles
| Type de tâche | Agent responsable | Ne pas confier à |
|---|---|---|
| Écrire/modifier du code | `@coder` | `@verifier`, `@explorer` |
| Valider (ruff, mypy, pytest, bandit) | `@verifier` | `@coder` |
| Diagnostiquer un bug | `@debugger` | `@coder` |
| Écrire un test | `@test-engineer` | `@coder` |
| Rédiger de la documentation | `@tech-writer` | `@coder` |
| Explorer le dépôt (lecture) | `@explorer` | `@coder`, `@verifier` |
| Arbitrage technique structurant | `@architect` | `@coder`, `@planner` |
| Revue de code | `@reviewer` | `@coder`, `@verifier` |
| Audit de sécurité | `@security-auditor` | `@coder`, `@verifier` |
| Recherche web | `@web-explorer` | `@explorer` |
| Découpage de travail complexe | `@planner` | `@coder` |
--- ---
## 10. Workflow de modification ## 10. Workflow de modification
@@ -213,7 +229,7 @@ Les rôles d'agents disponibles pour ce projet sont les suivants :
2. Préserver les changements existants de l'utilisateur. 2. Préserver les changements existants de l'utilisateur.
3. Pour une correction, reproduire d'abord le défaut avec un test automatisé lorsque c'est raisonnable. 3. Pour une correction, reproduire d'abord le défaut avec un test automatisé lorsque c'est raisonnable.
4. Faire une modification étroite et cohérente, en respectant les conventions du projet (idempotence, mode dégradé, repli iCal/pronotepy). 4. Faire une modification étroite et cohérente, en respectant les conventions du projet (idempotence, mode dégradé, repli iCal/pronotepy).
5. Vérifier le comportement nominal et les cas d'erreur, notamment : 5. Faire vérifier le comportement par `@verifier` (ruff, mypy, pytest, bandit) et les cas d'erreur, notamment :
- Succès de la synchronisation Pronote → CalDAV/XMPP. - Succès de la synchronisation Pronote → CalDAV/XMPP.
- Repli vers iCal en cas d'échec de `pronotepy`. - Repli vers iCal en cas d'échec de `pronotepy`.
- Gestion des erreurs explicites. - Gestion des erreurs explicites.

File diff suppressed because it is too large Load Diff

14
TODO.md
View File

@@ -73,13 +73,13 @@ Définir tous les modèles de domaine, immuables pour les contrats, mutables pou
Récupérer et normaliser l'agenda, les devoirs et les messages Pronote, avec repli entre iCal et pronotepy. Récupérer et normaliser l'agenda, les devoirs et les messages Pronote, avec repli entre iCal et pronotepy.
- [ ] Créer `sources/pronote/ical.py` : `fetch_ical(url)` (HTTP via `requests`, erreurs redactées) et parsing iCal → `Lesson`/`Homework`/`SchoolEvent` (`icalendar`). - [x] Créer `sources/pronote/ical.py` : `fetch_ical(url)` (HTTP via `requests`, erreurs redactées) et parsing iCal → `Lesson`/`Homework`/`SchoolEvent` (`icalendar`).
- [ ] Extraire les blocs de devoirs (`HomeworkBlock`) depuis `DESCRIPTION` et dédupliquer les devoirs (clé normalisée par date). - [x] Extraire les blocs de devoirs (`HomeworkBlock`) depuis `DESCRIPTION` et dédupliquer les devoirs (clé normalisée par date).
- [ ] Détecter les statuts (`CANCELLED`/`MOVED`) via `CATEGORIES` et `STATUS:CANCELLED`. - [x] Détecter les statuts (`CANCELLED`/`MOVED`) via `CATEGORIES` et `STATUS:CANCELLED`.
- [ ] Créer `sources/pronote/client.py` : client `pronotepy` (messages, informations, discussions, sondages, et devoirs en repli) avec masquage des erreurs. - [x] Créer `sources/pronote/client.py` : client `pronotepy` (messages, informations, discussions, sondages, et devoirs en repli) avec masquage des erreurs.
- [ ] Créer `sources/pronote/fallback.py` : sélection de source selon `PRONOTE_*_SOURCE` (auto/ical/pronotepy) et `PronoteFetcher` unifiant `fetch_agenda`/`fetch_homework`/`fetch_messages`. - [x] Créer `sources/pronote/fallback.py` : sélection de source selon `PRONOTE_*_SOURCE` (auto/ical/pronotepy) et `PronoteFetcher` unifiant `fetch_agenda`/`fetch_homework`/`fetch_messages`.
- [ ] Implémenter le repli : iCal échoue → pronotepy ; pronotepy échoue → iCal ; les deux échouent → `PipelineCriticalError`. - [x] Implémenter le repli unique : source primaire + un seul repli (l'autre source si configurée) ; les deux échouent → `PipelineCriticalError`.
- [ ] Normaliser les UID via `utils/uid.normalize_uid` pour la stabilité des événements. - [x] Normaliser les UID via `utils/uid.normalize_pronote_uid` pour la stabilité des événements.
### Critères d'acceptation ### Critères d'acceptation
- `fetch_ical` parse `tests/fixtures/pronote-4e.ics` en leçons/devoirs/événements corrects (cours annulé détecté). - `fetch_ical` parse `tests/fixtures/pronote-4e.ics` en leçons/devoirs/événements corrects (cours annulé détecté).

View File

@@ -10,7 +10,7 @@ from __future__ import annotations
from typing import Literal from typing import Literal
from pydantic import SecretStr from pydantic import Field, SecretStr, field_serializer
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -23,14 +23,28 @@ class PronoteSettings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="PRONOTE_") model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="PRONOTE_")
ical_url: str | None = None ical_url: SecretStr | None = None
username: str | None = None username: str | None = None
password: SecretStr | None = None password: SecretStr | None = None
ent: str | None = None ent: str | None = None
pronote_url: str | None = None
account_type: Literal["student", "parent"] = "parent"
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto" agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
homework_source: Literal["auto", "ical", "pronotepy"] = "auto" homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
messages_source: Literal["pronotepy"] = "pronotepy" messages_source: Literal["pronotepy"] = "pronotepy"
@field_serializer("ical_url")
def _serialize_ical_url(self, value: SecretStr | None) -> str | None:
"""Masque l'URL iCal lors de la sérialisation (repr, str, JSON).
:param value: Valeur du champ ``ical_url``.
:return: ``"**********"`` si la valeur est définie, ``None`` sinon.
:rtype: str | None
"""
if value is None:
return None
return "**********"
class CalDAVSettings(BaseSettings): class CalDAVSettings(BaseSettings):
"""Paramètres d'accès au serveur CalDAV de destination. """Paramètres d'accès au serveur CalDAV de destination.
@@ -121,12 +135,9 @@ class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore") model_config = SettingsConfigDict(env_file=".env", extra="ignore")
pronote: PronoteSettings = PronoteSettings() pronote: PronoteSettings = Field(default_factory=PronoteSettings)
caldav: CalDAVSettings = CalDAVSettings() caldav: CalDAVSettings = Field(default_factory=CalDAVSettings)
xmpp: XmppSettings = XmppSettings() xmpp: XmppSettings = Field(default_factory=XmppSettings)
ai: AISettings = AISettings() ai: AISettings = Field(default_factory=AISettings)
blog: BlogSettings = BlogSettings() blog: BlogSettings = Field(default_factory=BlogSettings)
app: AppSettings = AppSettings() app: AppSettings = Field(default_factory=AppSettings)
settings = Settings()

33
pronote_sync/errors.py Normal file
View File

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

View File

@@ -8,7 +8,7 @@ from datetime import time as _time
from enum import StrEnum from enum import StrEnum
from typing import Literal from typing import Literal
from pydantic import BaseModel, ConfigDict, Field from pydantic import BaseModel, ConfigDict, Field, model_validator
class Status(StrEnum): class Status(StrEnum):
@@ -30,6 +30,8 @@ class LessonStatus(StrEnum):
class HomeworkBlock(BaseModel): class HomeworkBlock(BaseModel):
"""Représente un bloc de devoir extrait de la description d'un cours.""" """Représente un bloc de devoir extrait de la description d'un cours."""
model_config = ConfigDict(frozen=True)
kind: Literal["due", "assigned"] = Field( kind: Literal["due", "assigned"] = Field(
..., description="Type de bloc (échéance ou attribution)" ..., description="Type de bloc (échéance ou attribution)"
) )
@@ -47,16 +49,28 @@ class Lesson(BaseModel):
start: datetime = Field(..., description="Date/heure de début") start: datetime = Field(..., description="Date/heure de début")
end: datetime = Field(..., description="Date/heure de fin") end: datetime = Field(..., description="Date/heure de fin")
subject: str = Field(..., description="Matière") subject: str = Field(..., description="Matière")
teachers: list[str] = Field(default_factory=list, description="Liste des professeurs") teachers: tuple[str, ...] = Field(default=(), description="Liste des professeurs")
rooms: list[str] = Field(default_factory=list, description="Liste des salles") rooms: tuple[str, ...] = Field(default=(), description="Liste des salles")
group: str | None = Field(None, description="Groupe (ex: Classe entière)") group: str | None = Field(None, description="Groupe (ex: Classe entière)")
status: LessonStatus = Field(default=LessonStatus.NORMAL, description="Statut du cours") status: LessonStatus = Field(default=LessonStatus.NORMAL, description="Statut du cours")
content: str | None = Field(None, description="Contenu pédagogique") content: str | None = Field(None, description="Contenu pédagogique")
homework_blocks: list[HomeworkBlock] = Field( homework_blocks: tuple[HomeworkBlock, ...] = Field(
default_factory=list, default=(),
description="Blocs de devoirs extraits de la description", description="Blocs de devoirs extraits de la description",
) )
@model_validator(mode="after")
def _validate_time_order(self) -> Lesson:
"""Valide que l'heure de fin est postérieure à l'heure de début.
:return: L'instance validée.
:rtype: Lesson
:raises ValueError: Si ``end`` n'est pas postérieur à ``start``.
"""
if self.end <= self.start:
raise ValueError("end doit être postérieur à start")
return self
class SchoolEventKind(StrEnum): class SchoolEventKind(StrEnum):
"""Type d'événement scolaire.""" """Type d'événement scolaire."""
@@ -75,6 +89,18 @@ class SchoolEvent(BaseModel):
from_date: _date = Field(..., description="Date de début (inclusive)") from_date: _date = Field(..., description="Date de début (inclusive)")
to_date: _date = Field(..., description="Date de fin (exclusive)") to_date: _date = Field(..., description="Date de fin (exclusive)")
@model_validator(mode="after")
def _validate_date_order(self) -> SchoolEvent:
"""Valide que la date de fin est postérieure ou égale à la date de début.
:return: L'instance validée.
:rtype: SchoolEvent
:raises ValueError: Si ``to_date`` est antérieur à ``from_date``.
"""
if self.to_date < self.from_date:
raise ValueError("to_date doit être postérieur ou égal à from_date")
return self
class TheoreticalLesson(BaseModel): class TheoreticalLesson(BaseModel):
"""Représente un cours dans l'agenda théorique.""" """Représente un cours dans l'agenda théorique."""
@@ -82,9 +108,23 @@ class TheoreticalLesson(BaseModel):
model_config = ConfigDict(frozen=True) model_config = ConfigDict(frozen=True)
id: str = Field(..., description="Identifiant unique") id: str = Field(..., description="Identifiant unique")
day_of_week: int = Field(..., description="Jour de la semaine (0=lundi, 6=dimanche)") day_of_week: int = Field(
..., ge=0, le=6, description="Jour de la semaine (0=lundi, 6=dimanche)"
)
start_time: _time = Field(..., description="Heure de début") start_time: _time = Field(..., description="Heure de début")
end_time: _time = Field(..., description="Heure de fin") end_time: _time = Field(..., description="Heure de fin")
subject: str = Field(..., description="Matière") subject: str = Field(..., description="Matière")
teachers: list[str] = Field(default_factory=list, description="Liste des professeurs") teachers: tuple[str, ...] = Field(default=(), description="Liste des professeurs")
rooms: list[str] = Field(default_factory=list, description="Liste des salles") rooms: tuple[str, ...] = Field(default=(), description="Liste des salles")
@model_validator(mode="after")
def _validate_time_order(self) -> TheoreticalLesson:
"""Valide que l'heure de fin est postérieure à l'heure de début.
:return: L'instance validée.
:rtype: TheoreticalLesson
:raises ValueError: Si ``end_time`` n'est pas postérieur à ``start_time``.
"""
if self.end_time <= self.start_time:
raise ValueError("end_time doit être postérieur à start_time")
return self

View File

@@ -26,17 +26,22 @@ class BlogArticle(BaseModel):
class ExternalInfo(BaseModel): class ExternalInfo(BaseModel):
"""Agrège les informations externes (blog, messages Pronote) pour le message XMPP.""" """Agrège les informations externes (blog, messages Pronote) pour le message XMPP.
blog_articles: list[BlogArticle] = Field( Modèle figé (``frozen``) : les instances sont immuables après création.
default_factory=list, """
model_config = ConfigDict(frozen=True)
blog_articles: tuple[BlogArticle, ...] = Field(
default=(),
description="Liste des nouveaux articles du blog", description="Liste des nouveaux articles du blog",
) )
pronote_messages: list[Message] = Field( pronote_messages: tuple[Message, ...] = Field(
default_factory=list, default=(),
description="Liste des messages Pronote", description="Liste des messages Pronote",
) )
other_info: list[str] = Field( other_info: tuple[str, ...] = Field(
default_factory=list, default=(),
description="Autres informations (extensible)", description="Autres informations (extensible)",
) )

View File

@@ -5,7 +5,7 @@ from __future__ import annotations
from datetime import date from datetime import date
from enum import StrEnum from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field from pydantic import BaseModel, ConfigDict, Field, model_validator
from pronote_sync.models.agenda import Lesson, TheoreticalLesson from pronote_sync.models.agenda import Lesson, TheoreticalLesson
@@ -30,6 +30,22 @@ class AgendaChange(BaseModel):
) )
details: str = Field(default="", description="Détails du changement") details: str = Field(default="", description="Détails du changement")
@model_validator(mode="after")
def _validate_payload_consistency(self) -> AgendaChange:
"""Valide la cohérence entre le type de changement et le payload.
:return: L'instance validée.
:rtype: AgendaChange
:raises ValueError: Si le payload ne correspond pas au type de changement.
"""
if self.type in (AgendaChangeType.ADDED, AgendaChangeType.MODIFIED):
if self.lesson is None:
raise ValueError(f"lesson est requis pour le type {self.type!r}")
if self.type == AgendaChangeType.REMOVED:
if self.theoretical_lesson is None:
raise ValueError(f"theoretical_lesson est requis pour le type {self.type!r}")
return self
class AgendaDiff(BaseModel): class AgendaDiff(BaseModel):
"""Représente les différences entre l'agenda réel et l'agenda théorique.""" """Représente les différences entre l'agenda réel et l'agenda théorique."""
@@ -37,4 +53,4 @@ class AgendaDiff(BaseModel):
model_config = ConfigDict(frozen=True) model_config = ConfigDict(frozen=True)
target_date: date = Field(..., description="Date cible de la comparaison") target_date: date = Field(..., description="Date cible de la comparaison")
changes: list[AgendaChange] = Field(default_factory=list, description="Liste des changements") changes: tuple[AgendaChange, ...] = Field(default=(), description="Liste des changements")

View File

@@ -14,7 +14,7 @@ class Homework(BaseModel):
id: str = Field(..., description="ID stable (hachage)") id: str = Field(..., description="ID stable (hachage)")
subject: str = Field(..., description="Matière") subject: str = Field(..., description="Matière")
teachers: list[str] = Field(default_factory=list, description="Liste des professeurs") teachers: tuple[str, ...] = Field(default=(), description="Liste des professeurs")
assigned_on: date | None = Field(None, description="Date de distribution") assigned_on: date | None = Field(None, description="Date de distribution")
due_on: date = Field(..., description="Date d'échéance") due_on: date = Field(..., description="Date d'échéance")
text: str = Field(..., description="Texte du devoir (brut)") text: str = Field(..., description="Texte du devoir (brut)")

View File

@@ -36,7 +36,7 @@ class CalDAVSyncResult(BaseModel):
"""Résultat d'une synchronisation CalDAV (mutable).""" """Résultat d'une synchronisation CalDAV (mutable)."""
status: CalDAVSyncStatus = Field(..., description="Statut global") status: CalDAVSyncStatus = Field(..., description="Statut global")
added: int = Field(0, description="Nombre d'événements ajoutés") added: int = Field(0, ge=0, description="Nombre d'événements ajoutés")
updated: int = Field(0, description="Nombre d'événements mis à jour") updated: int = Field(0, ge=0, description="Nombre d'événements mis à jour")
removed: int = Field(0, description="Nombre d'événements supprimés") removed: int = Field(0, ge=0, description="Nombre d'événements supprimés")
errors: list[str] = Field(default_factory=list, description="Liste des erreurs") errors: list[str] = Field(default_factory=list, description="Liste des erreurs")

View File

@@ -19,13 +19,11 @@ class XmppMessage(BaseModel):
target_date: date = Field(..., description="Date cible") target_date: date = Field(..., description="Date cible")
synthesis: str | None = Field(None, description="Synthèse IA (optionnelle, 3-5 phrases)") synthesis: str | None = Field(None, description="Synthèse IA (optionnelle, 3-5 phrases)")
homeworks: list[Homework] = Field(default_factory=list, description="Liste brute des devoirs") homeworks: tuple[Homework, ...] = Field(default=(), description="Liste brute des devoirs")
changes: list[AgendaChange] = Field( changes: tuple[AgendaChange, ...] = Field(
default_factory=list, description="Liste des changements d'agenda" default=(), description="Liste des changements d'agenda"
)
messages: list[Message] = Field(
default_factory=list, description="Liste des messages/informations"
) )
messages: tuple[Message, ...] = Field(default=(), description="Liste des messages/informations")
external_info: ExternalInfo | None = Field( external_info: ExternalInfo | None = Field(
None, description="Informations externes (blog, messages Pronote)" None, description="Informations externes (blog, messages Pronote)"
) )

View File

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

View File

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

View File

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

View File

@@ -22,26 +22,15 @@ class RedactingFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str: def format(self, record: logging.LogRecord) -> str:
"""Formate un enregistrement de log en masquant les secrets. """Formate un enregistrement de log en masquant les secrets.
Le message et chaque argument textuel de l'enregistrement sont rédigés La rédaction est appliquée à la chaîne finale (message, arguments et
avant le formatage final effectué par :class:`logging.Formatter`. traceback inclus) produite par :class:`logging.Formatter`.
:param record: Enregistrement de log à formater. :param record: Enregistrement de log à formater.
:return: Message formaté, avec les secrets remplacés par ``REDACTED``. :return: Message formaté, avec les secrets remplacés par ``REDACTED``.
:rtype: str :rtype: str
""" """
record.msg = redact_secrets(str(record.msg)) formatted = super().format(record)
args = record.args return redact_secrets(formatted)
if args:
if isinstance(args, tuple):
record.args = tuple(
redact_secrets(arg) if isinstance(arg, str) else arg for arg in args
)
else:
record.args = {
key: redact_secrets(value) if isinstance(value, str) else value
for key, value in args.items()
}
return super().format(record)
def setup_logging(level: str = "INFO") -> None: def setup_logging(level: str = "INFO") -> None:

View File

@@ -10,10 +10,30 @@ from __future__ import annotations
import re import re
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
_SENSITIVE_QUERY_KEYS = frozenset({"icalsecurise", "token", "key", "password", "secret"}) _SENSITIVE_QUERY_KEYS = frozenset(
_URL_PATTERN = re.compile(r"https?://[^\s]+") {
"icalsecurise",
"token",
"key",
"password",
"secret",
"api_key",
"apikey",
"access_token",
"auth",
"authorization",
}
)
_URL_PATTERN = re.compile(r"https?://[^\s]+", re.IGNORECASE)
_AUTH_HEADER_PATTERN = re.compile(
r"((?:Proxy-)?Authorization)\s*[:=]\s*\S[^\r\n]*",
re.IGNORECASE,
)
_ISOLATED_SECRET_PATTERN = re.compile( _ISOLATED_SECRET_PATTERN = re.compile(
r"\b(icalsecurise|token|password|secret|key)\s*=\s*[^\s&]+", r"\b(icalsecurise|access_token|api_key|apikey|authorization|token|password|secret|key|auth)"
r"(\s*['\"]?\s*[:=]\s*)"
r"(['\"]?)"
r"([^\s&'\"]+)",
re.IGNORECASE, re.IGNORECASE,
) )
_REDACTED = "REDACTED" _REDACTED = "REDACTED"
@@ -21,15 +41,28 @@ _REDACTED_URL = "REDACTED_URL"
def redact_url(url: str) -> str: def redact_url(url: str) -> str:
"""Masque les paramètres sensibles dans une URL. """Masque les identifiants et les paramètres sensibles d'une URL.
:param url: URL pouvant contenir des paramètres sensibles (ex: ``icalsecurise``). Les informations d'authentification du netloc (``utilisateur:motdepasse@hôte``)
:return: URL avec les paramètres sensibles remplacés par ``REDACTED``, sont entièrement masquées (utilisateur et mot de passe), ainsi que les
paramètres sensibles de la requête (ex: ``icalsecurise``).
:param url: URL pouvant contenir des informations sensibles (ex: ``icalsecurise``).
:return: URL avec les éléments sensibles remplacés par ``REDACTED``,
ou ``REDACTED_URL`` si le traitement échoue. ou ``REDACTED_URL`` si le traitement échoue.
:rtype: str :rtype: str
""" """
try: try:
parts = urlsplit(url) parts = urlsplit(url)
if parts.username is not None or parts.password is not None:
# Netloc sûr : REDACTED@hôte:port. L'utilisateur et le mot de
# passe sont entièrement masqués.
host = parts.hostname or ""
if parts.port is not None:
netloc = f"{_REDACTED}@{host}:{parts.port}"
else:
netloc = f"{_REDACTED}@{host}"
parts = parts._replace(netloc=netloc)
query: list[tuple[str, str]] = parse_qsl(parts.query, keep_blank_values=True) query: list[tuple[str, str]] = parse_qsl(parts.query, keep_blank_values=True)
redacted_query = [ redacted_query = [
(key, _REDACTED if key.lower() in _SENSITIVE_QUERY_KEYS else value) (key, _REDACTED if key.lower() in _SENSITIVE_QUERY_KEYS else value)
@@ -43,16 +76,19 @@ def redact_url(url: str) -> str:
def redact_secrets(text: str) -> str: def redact_secrets(text: str) -> str:
"""Masque les secrets présents dans un texte arbitraire. """Masque les secrets présents dans un texte arbitraire.
Les URLs sont d'abord traitées par :func:`redact_url`, puis les affectations Les URLs sont d'abord traitées par :func:`redact_url`, puis les en-têtes
isolées de type ``cle=valeur`` (ex: ``icalsecurise=XXX``) sont masquées, d'authentification (``Authorization``, ``Proxy-Authorization``) et les
sans distinction de casse. affectations isolées de type ``cle=valeur`` ou ``cle:valeur``
(ex: ``icalsecurise=XXX``, ``"token": "XXX"``) sont masquées, sans
distinction de casse.
:param text: Texte pouvant contenir des URLs ou des secrets en clair. :param text: Texte pouvant contenir des URLs ou des secrets en clair.
:return: Texte avec les secrets remplacés par ``REDACTED``. :return: Texte avec les secrets remplacés par ``REDACTED``.
:rtype: str :rtype: str
""" """
redacted = _URL_PATTERN.sub(lambda match: redact_url(match.group(0)), text) redacted = _URL_PATTERN.sub(lambda match: redact_url(match.group(0)), text)
return _ISOLATED_SECRET_PATTERN.sub(r"\1=REDACTED", redacted) redacted = _AUTH_HEADER_PATTERN.sub(r"\1: REDACTED", redacted)
return _ISOLATED_SECRET_PATTERN.sub(r"\1\2\3REDACTED", redacted)
def redact_exception(exc: Exception) -> str: def redact_exception(exc: Exception) -> str:

View File

@@ -7,7 +7,7 @@ name = "pronote-sync"
version = "0.1.0" version = "0.1.0"
description = "Synchronisation Pronote → CalDAV + XMPP" description = "Synchronisation Pronote → CalDAV + XMPP"
license = {text = "MIT"} license = {text = "MIT"}
requires-python = ">=3.13" requires-python = ">=3.13.5"
authors = [ authors = [
{name = "Votre Nom", email = "votre@email.com"} {name = "Votre Nom", email = "votre@email.com"}
] ]

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

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

0
tests/unit/__init__.py Normal file
View File

119
tests/unit/test_config.py Normal file
View File

@@ -0,0 +1,119 @@
"""Tests unitaires pour la configuration et le chargement des secrets.
Ce module vérifie que la configuration est chargée correctement depuis les
variables d'environnement, que les secrets sont masqués dans les représentations
textuelles et sérialisées, et que le rechargement fonctionne comme attendu.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from pydantic import SecretStr
from pronote_sync.config.env import load_settings
from pronote_sync.config.settings import PronoteSettings, Settings
if TYPE_CHECKING:
from _pytest.monkeypatch import MonkeyPatch
def test_env_prefix_loading(monkeypatch: MonkeyPatch) -> None:
"""Vérifie que les variables d'environnement avec préfixe sont chargées correctement.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("PRONOTE_USERNAME", "testuser")
settings = load_settings()
assert settings.pronote.username == "testuser"
def test_env_reload_via_load_settings(monkeypatch: MonkeyPatch) -> None:
"""Vérifie que le rechargement de la configuration lit les nouvelles valeurs.
Ce test garantit que les champs configurés avec ``default_factory`` créent
des instances fraîches à chaque appel de ``load_settings()``, permettant ainsi
un rechargement dynamique des variables d'environnement.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
# Premier chargement (valeur par défaut None)
settings1 = load_settings()
assert settings1.pronote.username is None
# Modification de l'environnement
monkeypatch.setenv("PRONOTE_USERNAME", "newuser")
# Rechargement
settings2 = load_settings()
assert settings2.pronote.username == "newuser"
def test_secretstr_masking_ical_url() -> None:
"""Vérifie que l'URL iCal contenant un secret est masquée dans les représentations.
:return: None
"""
ical_url = "https://example.fr/ical?icalsecurise=SECRET_TOKEN" # pragma: allowlist secret
settings = Settings(pronote=PronoteSettings(ical_url=SecretStr(ical_url)))
# Vérification de la représentation string
str_repr = str(settings)
assert "SECRET_TOKEN" not in str_repr
assert "**********" in str_repr
# Vérification de la représentation repr
repr_repr = repr(settings)
assert "SECRET_TOKEN" not in repr_repr
assert "**********" in repr_repr
# Vérification de la sérialisation JSON
json_str = settings.model_dump_json()
assert "SECRET_TOKEN" not in json_str
assert "**********" in json_str
def test_secretstr_masking_password() -> None:
"""Vérifie que le mot de passe est masqué dans les représentations.
:return: None
"""
password = "mypass123" # pragma: allowlist secret
settings = Settings(pronote=PronoteSettings(password=SecretStr(password)))
# Vérification de la représentation string
str_repr = str(settings)
assert "mypass123" not in str_repr
assert "**********" in str_repr
# Vérification de la représentation repr
repr_repr = repr(settings)
assert "mypass123" not in repr_repr
assert "**********" in repr_repr
# Vérification de la sérialisation JSON
json_str = settings.model_dump_json()
assert "mypass123" not in json_str
assert "**********" in json_str
def test_no_singleton_import() -> None:
"""Vérifie que l'import du singleton settings lève une ImportError.
Ce test garantit que le pattern singleton a été supprimé et que la
configuration doit être chargée explicitement via ``load_settings()``.
:return: None
"""
with pytest.raises(ImportError):
# Attempt to import a non-existent attribute to trigger ImportError
# as required by the spec
from pronote_sync.config.settings import ( # type: ignore[attr-defined] # noqa: F401
settings,
)
# Ensure trailing newline

987
tests/unit/test_fallback.py Normal file
View File

@@ -0,0 +1,987 @@
"""Tests unitaires pour la logique de repli iCal / pronotepy.
Ce module valide le comportement du module :mod:`pronote_sync.sources.pronote.fallback`
et de son implémentation :class:`PronoteFetcher`. Les tests couvrent :
- La sélection de la source d'agenda (``ical``, ``pronotepy``, ``auto``) et de devoirs,
- Le repli automatique iCal → pronotepy en mode ``auto``,
- La levée de :class:`PipelineCriticalError` lorsque toutes les sources échouent,
- Le masquage des secrets dans les messages d'erreur,
- Les méthodes toujours basées sur pronotepy (messages, informations).
Tous les appels réseau et les interactions avec pronotepy sont mockés.
"""
from __future__ import annotations
from datetime import date, datetime
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
import pytest
from pydantic import SecretStr
from pronote_sync.config.settings import PronoteSettings, Settings
from pronote_sync.errors import PipelineCriticalError
from pronote_sync.models.agenda import (
HomeworkBlock,
Lesson,
LessonStatus,
SchoolEvent,
SchoolEventKind,
)
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
from pronote_sync.sources.pronote.fallback import PronoteFetcher
if TYPE_CHECKING:
from typing import Protocol
class _MockPronoteClientProtocol(Protocol):
def get_messages(self) -> list[Message]: ...
def get_informations(self) -> list[Message]: ...
def get_lessons(self, start: date, end: date) -> list[Lesson]: ...
def get_homeworks(self, start: date, end: date) -> list[Homework]: ...
@pytest.fixture(name="mock_settings")
def fixture_mock_settings() -> Settings:
"""Fixture fournissant une configuration Settings adaptée aux tests.
:return: Instance de :class:`Settings` avec des valeurs sûres pour les tests.
:rtype: Settings
"""
return Settings(
pronote=PronoteSettings(
pronote_url="https://pronote.example.com",
ical_url=SecretStr("file:///fake/ical.ics"),
agenda_source="auto",
homework_source="auto",
username="testuser",
password=SecretStr("testpass"),
ent="ent",
),
app=Settings().app,
)
@pytest.fixture(name="mock_fetcher")
def fixture_mock_fetcher(mock_settings: Settings) -> PronoteFetcher:
"""Fixture fournissant une instance de :class:`PronoteFetcher` prête à l'emploi.
:param mock_settings: Configuration de test.
:return: Instance de :class:`PronoteFetcher` pour les tests.
:rtype: PronoteFetcher
"""
client: _MockPronoteClientProtocol = MagicMock()
return PronoteFetcher(settings=mock_settings, pronote_client=client)
def test_fetch_agenda_ical_mode(mock_fetcher: PronoteFetcher) -> None:
"""Test la récupération de l'agenda en mode source iCal.
On mock ``fetch_ical`` et ``parse_ical`` pour retourner des cours et événements.
On vérifie que le fetcher retourne bien ces données.
:param mock_fetcher: Fetcher de test.
:return: None
"""
start_dt = datetime(2025, 9, 1, 8, 0)
end_dt = datetime(2025, 9, 1, 9, 30)
lessons = [
Lesson(
id="l1",
start=start_dt,
end=end_dt,
subject="Maths",
teachers=("Dupont",),
rooms=("S1",),
group="2ndeA",
status=LessonStatus.NORMAL,
content=None,
)
]
events = [
SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Vacances",
from_date=date(2025, 9, 1),
to_date=date(2025, 9, 15),
)
]
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.return_value = "BEGIN:VCALENDAR\n..."
m_parse_ical.return_value = (lessons, [], events)
result_lessons, result_events = mock_fetcher.fetch_agenda()
assert result_lessons == lessons
assert result_events == events
m_fetch_ical.assert_called_once()
m_parse_ical.assert_called_once()
def test_fetch_agenda_pronotepy_mode(mock_fetcher: PronoteFetcher) -> None:
"""Test la récupération de l'agenda en mode source pronotepy.
On mock ``get_lessons`` du client pour retourner des cours.
On vérifie que le fetcher retourne ces cours (événements scolaires vides).
:param mock_fetcher: Fetcher de test.
:return: None
"""
start_dt = datetime(2025, 9, 1, 8, 0)
end_dt = datetime(2025, 9, 1, 9, 30)
lessons = [
Lesson(
id="l1",
start=start_dt,
end=end_dt,
subject="Physique",
teachers=("Martin",),
rooms=("Labo1",),
group="1ereB",
status=LessonStatus.NORMAL,
content=None,
)
]
client = MagicMock()
client.get_lessons.return_value = lessons
mock_fetcher._pronote_client = client
result_lessons, result_events = mock_fetcher.fetch_agenda()
assert result_lessons == lessons
assert result_events == []
client.get_lessons.assert_called_once()
def test_fetch_agenda_auto_ical_success(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode auto : succès de l'iCal, pronotepy non appelé.
On mock iCal pour réussir, et on vérifie que pronotepy n'est pas sollicité.
:param mock_fetcher: Fetcher de test.
:return: None
"""
start_dt = datetime(2025, 9, 1, 8, 0)
end_dt = datetime(2025, 9, 1, 9, 30)
lessons = [
Lesson(
id="l1",
start=start_dt,
end=end_dt,
subject="SVT",
teachers=("Durand",),
rooms=("S2",),
group="3emeC",
status=LessonStatus.NORMAL,
content=None,
)
]
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.return_value = "BEGIN:VCALENDAR\n..."
m_parse_ical.return_value = (lessons, [], [])
result_lessons, _ = mock_fetcher.fetch_agenda()
assert result_lessons == lessons
def test_fetch_agenda_auto_fallback_to_pronotepy(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode auto : échec iCal, repli sur pronotepy.
On mock iCal pour échouer, pronotepy pour réussir. On vérifie que pronotepy est appelé.
:param mock_fetcher: Fetcher de test.
:return: None
"""
start_dt = datetime(2025, 9, 1, 8, 0)
end_dt = datetime(2025, 9, 1, 9, 30)
lessons = [
Lesson(
id="l1",
start=start_dt,
end=end_dt,
subject="Histoire",
teachers=("Lefevre",),
rooms=("S3",),
group="2ndeD",
status=LessonStatus.NORMAL,
content=None,
)
]
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_lessons.return_value = lessons
mock_fetcher._pronote_client = client
result_lessons, _ = mock_fetcher.fetch_agenda()
assert result_lessons == lessons
client.get_lessons.assert_called_once()
def test_fetch_agenda_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode auto : échec des deux sources → PipelineCriticalError.
On mock iCal et pronotepy pour échouer. On vérifie la levée de l'erreur critique.
:param mock_fetcher: Fetcher de test.
:return: None
"""
# Disable pronotepy so fallback is None
mock_fetcher._settings.pronote.pronote_url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "la source ical a échoué" in str(exc_info.value)
def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode ical : échec → PipelineCriticalError masquée.
On mock iCal pour échouer et désactive pronotepy pour que le repli soit None.
On vérifie que l'erreur brute est masquée dans la levée.
:param mock_fetcher: Fetcher de test.
:return: None
"""
# Override settings to use ical mode explicitly and disable fallback
mock_fetcher._settings.pronote.agenda_source = "ical"
mock_fetcher._settings.pronote.pronote_url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.side_effect = OSError(
"Impossible de lire le fichier iCal file:///fake/ical.ics : iCal unreachable"
)
m_parse_ical.side_effect = OSError("iCal parse error")
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "Impossible de récupérer l'agenda : la source ical a échoué" in str(exc_info.value)
# Vérifie que le message ne contient pas de secret
assert "file:///fake/ical.ics" not in str(exc_info.value)
def test_fetch_agenda_pronotepy_mode_failure(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode pronotepy : échec → PipelineCriticalError masquée.
On mock pronotepy pour échouer et désactive iCal pour que le repli soit None.
On vérifie que l'erreur est masquée dans la levée.
:param mock_fetcher: Fetcher de test.
:return: None
"""
# Override settings to use pronotepy mode explicitly and disable fallback
mock_fetcher._settings.pronote.agenda_source = "pronotepy"
mock_fetcher._settings.pronote.ical_url = None
client = MagicMock()
client.get_lessons.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "Impossible de récupérer l'agenda : la source pronotepy a échoué" in str(exc_info.value)
def test_fetch_homework_ical_mode(mock_fetcher: PronoteFetcher) -> None:
"""Test la récupération des devoirs en mode source iCal.
On mock iCal pour retourner des cours avec blocs de devoirs, et on cible une date.
On vérifie que les devoirs sont correctement collectés.
:param mock_fetcher: Fetcher de test.
:return: None
"""
target_date = date(2025, 9, 5)
start_dt = datetime(2025, 9, 1, 8, 0)
end_dt = datetime(2025, 9, 1, 9, 30)
lessons = [
Lesson(
id="l1",
start=start_dt,
end=end_dt,
subject="Maths",
teachers=("Dupont",),
rooms=("S1",),
group="2ndeA",
status=LessonStatus.NORMAL,
content=None,
homework_blocks=(
HomeworkBlock(
kind="due",
date=target_date,
text="Devoir sur les fonctions",
html="<p>Devoir sur les fonctions</p>",
),
),
)
]
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
):
m_fetch_ical.return_value = "BEGIN:VCALENDAR\n..."
m_parse_ical.return_value = (lessons, [], [])
m_collect.return_value = [
Homework(
id="hw1",
subject="Maths",
teachers=(),
assigned_on=None,
due_on=target_date,
text="Devoir sur les fonctions",
html="<p>Devoir sur les fonctions</p>",
)
]
result = mock_fetcher.fetch_homework(target_date)
assert len(result) == 1
assert result[0].subject == "Maths"
assert result[0].due_on == target_date
m_collect.assert_called_once_with(lessons, target_date)
def test_fetch_homework_auto_fallback(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode auto des devoirs : échec iCal, repli pronotepy.
On mock iCal pour échouer et pronotepy pour réussir. On vérifie que pronotepy est utilisé.
:param mock_fetcher: Fetcher de test.
:return: None
"""
target_date = date(2025, 9, 10)
homeworks = [
Homework(
id="hw1",
subject="Physique",
teachers=(),
assigned_on=None,
due_on=target_date,
text="TP à préparer",
html="TP à préparer",
)
]
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_homeworks.return_value = homeworks
mock_fetcher._pronote_client = client
m_collect.return_value = homeworks
result = mock_fetcher.fetch_homework(target_date)
assert result == homeworks
client.get_homeworks.assert_called_once()
def test_fetch_homework_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode auto des devoirs : échec des deux sources → PipelineCriticalError.
On mock iCal et pronotepy pour échouer. On vérifie la levée de l'erreur critique.
:param mock_fetcher: Fetcher de test.
:return: None
"""
target_date = date(2025, 9, 10)
# Disable pronotepy so fallback is None
mock_fetcher._settings.pronote.pronote_url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_homework(target_date)
assert "la source ical a échoué" in str(exc_info.value)
def test_fetch_messages(mock_fetcher: PronoteFetcher) -> None:
"""Test la récupération des messages (toujours via pronotepy).
On mock ``get_messages`` du client pour retourner des messages typés.
On vérifie que le fetcher retourne ces messages.
:param mock_fetcher: Fetcher de test.
:return: None
"""
messages = [
Message(
id="m1",
type=MessageType.DISCUSSION,
title="Devoir de maths",
content="À faire pour demain",
author="M. Dupont",
date=datetime(2025, 9, 1, 10, 0),
read=False,
)
]
client = MagicMock()
client.get_messages.return_value = messages
client.get_informations.return_value = []
mock_fetcher._pronote_client = client
result = mock_fetcher.fetch_messages()
assert result == messages
def test_fetch_informations(mock_fetcher: PronoteFetcher) -> None:
"""Test la récupération des informations (toujours via pronotepy).
On mock ``get_informations`` du client pour retourner des informations.
On vérifie que le fetcher retourne ces informations.
:param mock_fetcher: Fetcher de test.
:return: None
"""
infos = [
Message(
id="i1",
type=MessageType.INFORMATION,
title="Info rentrée",
content="Rappel des consignes",
author="CPE",
date=datetime(2025, 9, 1, 9, 0),
read=True,
)
]
client = MagicMock()
client.get_informations.return_value = infos
client.get_messages.return_value = []
mock_fetcher._pronote_client = client
result = mock_fetcher.fetch_informations()
assert result == infos
def test_no_secrets_in_error_messages(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que les messages d'erreur ne contiennent pas de secrets.
On simule une erreur iCal contenant un token ``icalsecurise`` et on vérifie que le log
masqué ne contient pas le token.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
"""
# Disable pronotepy so fallback is None to trigger PipelineCriticalError
mock_fetcher._settings.pronote.pronote_url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
error_msg = (
"Impossible de lire le fichier iCal file:///ical?icalsecurise=SECRET_TOKEN_123 : "
"[Errno 2] No such file or directory"
)
m_fetch_ical.side_effect = OSError(error_msg)
m_parse_ical.side_effect = OSError("parse error")
client = MagicMock()
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError):
mock_fetcher.fetch_agenda()
# Vérifie que le log contient la version masquée
assert "SECRET_TOKEN_123" not in caplog.text
assert "icalsecurise=REDACTED" in caplog.text or "icalsecurise" not in caplog.text
def test_fetch_agenda_ical_mode_fallback_to_pronotepy(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode ICAL : échec iCal, repli sur pronotepy.
On mock iCal pour échouer, pronotepy pour réussir. On vérifie que pronotepy est appelé.
:param mock_fetcher: Fetcher de test.
:return: None
"""
start_dt = datetime(2025, 9, 1, 8, 0)
end_dt = datetime(2025, 9, 1, 9, 30)
lessons = [
Lesson(
id="l1",
start=start_dt,
end=end_dt,
subject="Maths",
teachers=("Dupont",),
rooms=("S1",),
group="2ndeA",
status=LessonStatus.NORMAL,
content=None,
)
]
# Override settings to use ical mode explicitly
mock_fetcher._settings.pronote.agenda_source = "ical"
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_lessons.return_value = lessons
mock_fetcher._pronote_client = client
result_lessons, result_events = mock_fetcher.fetch_agenda()
assert result_lessons == lessons
assert result_events == []
client.get_lessons.assert_called_once()
def test_fetch_agenda_pronotepy_mode_fallback_to_ical(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode PRONOTEPY : échec pronotepy, repli sur iCal.
On mock pronotepy pour échouer, iCal pour réussir. On vérifie que iCal est appelé.
:param mock_fetcher: Fetcher de test.
:return: None
"""
start_dt = datetime(2025, 9, 1, 8, 0)
end_dt = datetime(2025, 9, 1, 9, 30)
lessons = [
Lesson(
id="l1",
start=start_dt,
end=end_dt,
subject="SVT",
teachers=("Durand",),
rooms=("S2",),
group="3emeC",
status=LessonStatus.NORMAL,
content=None,
)
]
# Override settings to use pronotepy mode explicitly
mock_fetcher._settings.pronote.agenda_source = "pronotepy"
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
client = MagicMock()
client.get_lessons.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
m_fetch_ical.return_value = "BEGIN:VCALENDAR\n..."
m_parse_ical.return_value = (lessons, [], [])
result_lessons, result_events = mock_fetcher.fetch_agenda()
assert result_lessons == lessons
assert result_events == []
m_fetch_ical.assert_called_once()
m_parse_ical.assert_called_once()
def test_fetch_agenda_no_source_configured_raises(mock_fetcher: PronoteFetcher) -> None:
"""Test le mode AUTO : aucune source configurée → PipelineCriticalError.
On désactive les deux sources. On vérifie que l'erreur critique est levée.
:param mock_fetcher: Fetcher de test.
:return: None
"""
# Disable both sources
mock_fetcher._settings.pronote.ical_url = None
mock_fetcher._settings.pronote.pronote_url = None
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
def test_fetch_homework_filters_by_target_date(mock_fetcher: PronoteFetcher) -> None:
"""Test que les devoirs sont filtrés par date cible en mode pronotepy.
On mock pronotepy pour retourner des devoirs avec différentes dates d'échéance.
On vérifie que seul le devoir correspondant à la date cible est retourné.
:param mock_fetcher: Fetcher de test.
:return: None
"""
target_date = date(2025, 9, 10)
other_date = date(2025, 9, 11)
homeworks = [
Homework(
id="hw1",
subject="Maths",
teachers=(),
assigned_on=None,
due_on=target_date,
text="Devoir pour aujourd'hui",
html="<p>Devoir pour aujourd'hui</p>",
),
Homework(
id="hw2",
subject="Physique",
teachers=(),
assigned_on=None,
due_on=other_date,
text="Devoir pour demain",
html="<p>Devoir pour demain</p>",
),
]
client = MagicMock()
client.get_homeworks.return_value = homeworks
mock_fetcher._pronote_client = client
result = mock_fetcher.fetch_homework(target_date)
assert len(result) == 1
assert result[0].id == "hw1"
assert result[0].due_on == target_date
client.get_homeworks.assert_called_once()
def test_fetch_messages_logs_and_re_raises(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que fetch_messages journalise et relance les exceptions.
On mock get_messages pour lever une exception contenant une URL secrète.
On vérifie que l'exception est relancée et que le log masqué ne contient pas l'URL.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
"""
client = MagicMock()
error_msg = "Erreur Pronote : impossible de récupérer les messages https://pronote.example.com/messages?token=SECRET_TOKEN_456"
client.get_messages.side_effect = OSError(error_msg)
mock_fetcher._pronote_client = client
with pytest.raises(OSError) as exc_info:
mock_fetcher.fetch_messages()
assert exc_info.value is client.get_messages.side_effect
assert "SECRET_TOKEN_456" not in caplog.text
assert (
"pronote.example.com/messages?token=REDACTED" in caplog.text
or "pronote.example.com/messages" in caplog.text
)
def test_fetch_agenda_ical_url_none_raises_value_error(mock_fetcher: PronoteFetcher) -> None:
"""Test que _fetch_agenda_ical lève ValueError si ical_url est None.
On désactive ical_url et on appelle _fetch_agenda_ical(). On vérifie que
l'erreur ValueError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
"""
mock_fetcher._settings.pronote.ical_url = None
with pytest.raises(ValueError, match="PRONOTE_ICAL_URL est requis pour la source iCal"):
mock_fetcher._fetch_agenda_ical()
def test_agenda_sources_auto_only_pronotepy_configured(mock_fetcher: PronoteFetcher) -> None:
"""Test _agenda_sources en mode AUTO avec seulement pronotepy configuré.
On désactive ical_url mais on garde pronotepy configuré. On vérifie que la
source primaire est pronotepy et qu'il n'y a pas de repli.
:param mock_fetcher: Fetcher de test.
:return: None
"""
mock_fetcher._settings.pronote.agenda_source = "auto"
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._agenda_sources()
assert primary == "pronotepy"
assert fallback is None
def test_fetch_agenda_fallback_both_fail_raises_pipeline_critical_error(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test que fetch_agenda lève PipelineCriticalError si ICAL échoue et pronotepy aussi.
On mock ICAL pour échouer, on garde pronotepy configuré mais on mock son échec.
On vérifie que l'erreur PipelineCriticalError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
"""
mock_fetcher._settings.pronote.agenda_source = "ical"
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_lessons.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
assert "les sources ical et pronotepy ont échoué" in str(exc_info.value)
def test_fetch_agenda_fallback_returns_empty_logs_warning(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que fetch_agenda retourne ([], []) et journalise un avertissement si le repli retourne vide.
On mock ICAL pour échouer, pronotepy configuré et retourne vide. On vérifie le retour et le log.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
"""
mock_fetcher._settings.pronote.agenda_source = "ical"
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
patch(
"pronote_sync.sources.pronote.fallback.PronoteFetcher._fetch_agenda_pronotepy"
) as m_fetch_pronotepy,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
mock_fetcher._pronote_client = client
m_fetch_pronotepy.return_value = ([], []) # Empty result from fallback
lessons, events = mock_fetcher.fetch_agenda()
assert lessons == []
assert events == []
assert "a retourné un résultat vide après l'échec de ical" in caplog.text
def test_homework_sources_explicit_ical_mode(mock_fetcher: PronoteFetcher) -> None:
"""Test _homework_sources en mode ICAL.
On vérifie que la source primaire est ical et que le repli est pronotepy si configuré,
ou None sinon.
:param mock_fetcher: Fetcher de test.
:return: None
"""
mock_fetcher._settings.pronote.homework_source = "ical"
# With pronotepy configured
primary, fallback = mock_fetcher._homework_sources()
assert primary == "ical"
assert fallback == "pronotepy"
# Without pronotepy configured
mock_fetcher._settings.pronote.pronote_url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "ical"
assert fallback is None
def test_homework_sources_explicit_pronotepy_mode(mock_fetcher: PronoteFetcher) -> None:
"""Test _homework_sources en mode PRONOTEPY.
On vérifie que la source primaire est pronotepy et que le repli est ical si configuré,
ou None sinon.
:param mock_fetcher: Fetcher de test.
:return: None
"""
mock_fetcher._settings.pronote.homework_source = "pronotepy"
# With ical configured
primary, fallback = mock_fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback == "ical"
# Without ical configured
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback is None
def test_homework_sources_auto_only_pronotepy_configured(mock_fetcher: PronoteFetcher) -> None:
"""Test _homework_sources en mode AUTO avec seulement pronotepy configuré.
On désactive ical_url mais on garde pronotepy configuré. On vérifie que la
source primaire est pronotepy et qu'il n'y a pas de repli.
:param mock_fetcher: Fetcher de test.
:return: None
"""
mock_fetcher._settings.pronote.homework_source = "auto"
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback is None
def test_homework_sources_auto_no_source_configured_raises(mock_fetcher: PronoteFetcher) -> None:
"""Test _homework_sources en mode AUTO avec aucune source configurée.
On désactive les deux sources. On vérifie que l'erreur PipelineCriticalError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
"""
mock_fetcher._settings.pronote.homework_source = "auto"
mock_fetcher._settings.pronote.ical_url = None
mock_fetcher._settings.pronote.pronote_url = None
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher._homework_sources()
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test que fetch_homework lève PipelineCriticalError si ICAL échoue et pronotepy aussi.
On mock ICAL pour échouer, on garde pronotepy configuré mais on mock son échec.
On vérifie que l'erreur PipelineCriticalError est levée.
:param mock_fetcher: Fetcher de test.
:return: None
"""
target_date = date(2025, 9, 10)
mock_fetcher._settings.pronote.homework_source = "ical"
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_homeworks.side_effect = OSError("Pronote API error")
mock_fetcher._pronote_client = client
m_collect.side_effect = OSError("collect error")
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_homework(target_date)
assert "les sources ical et pronotepy ont échoué" in str(exc_info.value)
def test_fetch_homework_fallback_returns_empty_logs_warning(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que fetch_homework retourne [] et journalise un avertissement si le repli retourne vide.
On mock ICAL pour échouer, pronotepy configuré et retourne vide. On vérifie le retour et le log.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
"""
target_date = date(2025, 9, 10)
mock_fetcher._settings.pronote.homework_source = "ical"
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
mock_fetcher._pronote_client = client
m_collect.return_value = [] # Empty result
result = mock_fetcher.fetch_homework(target_date)
assert result == []
assert "a retourné un résultat vide après l'échec de ical" in caplog.text
def test_fetch_informations_logs_and_re_raises_secret(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que fetch_informations journalise et relance les exceptions avec secret masqué.
On mock get_informations pour lever une exception contenant une URL secrète.
On vérifie que l'exception est relancée et que le log ne contient pas le secret.
:param mock_fetcher: Fetcher de test.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
"""
client = MagicMock()
error_msg = "Erreur Pronote : impossible de récupérer les informations https://pronote.example.com/infos?token=SECRET_TOKEN_789"
client.get_informations.side_effect = OSError(error_msg)
mock_fetcher._pronote_client = client
with pytest.raises(OSError) as exc_info:
mock_fetcher.fetch_informations()
assert exc_info.value is client.get_informations.side_effect
assert "SECRET_TOKEN_789" not in caplog.text
assert (
"pronote.example.com/infos?token=REDACTED" in caplog.text
or "pronote.example.com/infos" in caplog.text
)
# Ensure trailing newline

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

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

View File

@@ -0,0 +1,373 @@
"""Tests de construction et sérialisation JSON des modèles Pydantic.
Ce module vérifie que chaque modèle figé peut être sérialisé en JSON puis désérialisé
sans perte de données, garantissant la stabilité des contrats d'échange.
Les tests utilisent des données minimales valides extraites des définitions de modèles.
"""
from __future__ import annotations
from datetime import date, datetime, time
from typing import Any
import pytest
from pydantic import BaseModel
from pronote_sync.models.agenda import (
HomeworkBlock,
Lesson,
LessonStatus,
SchoolEvent,
SchoolEventKind,
TheoreticalLesson,
)
from pronote_sync.models.blog import BlogArticle, ExternalInfo
from pronote_sync.models.diff import AgendaChange, AgendaChangeType, AgendaDiff
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
from pronote_sync.models.xmpp import XmppMessage
@pytest.mark.parametrize(
"model_class, kwargs",
[
(
Lesson,
{
"id": "lesson-001",
"start": datetime(2024, 9, 6, 8, 0, 0),
"end": datetime(2024, 9, 6, 9, 30, 0),
"subject": "Mathématiques",
"teachers": ("M. Dupont",),
"rooms": ("Salle 101",),
"group": "Classe A",
"status": LessonStatus.NORMAL,
"content": "Algèbre linéaire",
"homework_blocks": (
HomeworkBlock(kind="due", date=date(2024, 9, 13), text="Exercice 42"),
),
},
),
(
TheoreticalLesson,
{
"id": "theo-lesson-001",
"day_of_week": 0, # Lundi
"start_time": time(8, 0, 0),
"end_time": time(9, 30, 0),
"subject": "Mathématiques",
"teachers": ("M. Dupont",),
"rooms": ("Salle 101",),
},
),
(
SchoolEvent,
{
"kind": SchoolEventKind.HOLIDAY,
"label": "Vacances de la Toussaint",
"from_date": date(2024, 10, 19),
"to_date": date(2024, 11, 3),
},
),
(
Homework,
{
"id": "hw-001",
"subject": "Mathématiques",
"teachers": ("M. Dupont",),
"assigned_on": date(2024, 9, 1),
"due_on": date(2024, 9, 15),
"text": "Exercice 42 à rendre",
"html": "<p>Exercice 42 à rendre</p>",
},
),
(
AgendaChange,
{
"type": AgendaChangeType.ADDED,
"lesson": Lesson(
id="lesson-002",
start=datetime(2024, 9, 7, 10, 0, 0),
end=datetime(2024, 9, 7, 11, 30, 0),
subject="Physique",
teachers=("M. Martin",),
rooms=("Labo 205",),
group=None,
content=None,
),
"theoretical_lesson": None,
"details": "Cours ajouté exceptionnellement",
},
),
(
AgendaDiff,
{
"target_date": date(2024, 9, 6),
"changes": (
AgendaChange(
type=AgendaChangeType.ADDED,
lesson=Lesson(
id="lesson-003",
start=datetime(2024, 9, 8, 14, 0, 0),
end=datetime(2024, 9, 8, 15, 30, 0),
subject="Histoire",
teachers=("Mme Bernard",),
rooms=("Salle 201",),
group=None,
content=None,
),
theoretical_lesson=None,
),
),
},
),
(
XmppMessage,
{
"target_date": date(2024, 9, 6),
"synthesis": "Aucun changement majeur aujourd'hui.",
"homeworks": (
Homework(
id="hw-002",
subject="Physique",
teachers=("M. Martin",),
assigned_on=None,
due_on=date(2024, 9, 7),
text="TP à préparer",
),
),
"changes": (
AgendaChange(
type=AgendaChangeType.MODIFIED,
lesson=Lesson(
id="lesson-004",
start=datetime(2024, 9, 6, 16, 0, 0),
end=datetime(2024, 9, 6, 17, 30, 0),
subject="SVT",
teachers=("M. Lefèvre",),
rooms=("Salle 302",),
group=None,
content=None,
),
theoretical_lesson=None,
),
),
"messages": (
Message(
id="msg-001",
type=MessageType.INFORMATION,
title="Rappel",
content="Réunion parents-professeurs demain",
author="Secrétariat",
date=datetime(2024, 9, 5, 10, 0, 0),
read=False,
),
),
"external_info": ExternalInfo(blog_articles=(), pronote_messages=(), other_info=()),
},
),
(
BlogArticle,
{
"id": "article-001",
"title": "Sortie scolaire à Paris",
"url": "https://college.fr/blog/sortie-paris",
"published_at": datetime(2024, 9, 5, 14, 30, 0),
"updated_at": datetime(2024, 9, 5, 16, 45, 0),
"category": "Actualités",
"author": "M. Dubois",
"content_html": "<p>Sortie prévue le 15 octobre</p>",
"content_text": "Sortie prévue le 15 octobre",
},
),
(
ExternalInfo,
{
"blog_articles": (
BlogArticle(
id="article-002",
title="Nouveau règlement intérieur",
url="https://college.fr/blog/reglement",
published_at=datetime(2024, 9, 1, 9, 0, 0),
updated_at=None,
category=None,
author=None,
content_html="<p>Mise à jour du règlement</p>",
content_text="Mise à jour du règlement",
),
),
"pronote_messages": (
Message(
id="msg-002",
type=MessageType.DISCUSSION,
title="Devoirs de maths",
content="Voir Pronote pour les détails",
author="Prof Maths",
date=datetime(2024, 9, 4, 8, 0, 0),
read=False,
),
),
"other_info": ("Info complémentaire 1", "Info complémentaire 2"),
},
),
(
Message,
{
"id": "msg-003",
"type": MessageType.SURVEY,
"title": "Enquête de satisfaction",
"content": "Merci de répondre au questionnaire",
"author": "Direction",
"date": datetime(2024, 9, 5, 11, 0, 0),
"read": False,
},
),
],
)
def test_model_json_roundtrip(model_class: type[BaseModel], kwargs: dict[str, Any]) -> None:
"""Vérifie qu'un modèle peut être sérialisé en JSON puis désérialisé sans perte.
:param model_class: Classe du modèle à tester.
:type model_class: type[BaseModel]
:param kwargs: Arguments de constructeur valides.
:type kwargs: dict[str, Any]
"""
# Construction de l'instance
instance = model_class(**kwargs)
# Sérialisation JSON
json_str = instance.model_dump_json()
# Désérialisation depuis JSON
roundtrip_instance = model_class.model_validate_json(json_str)
# Vérification des champs clés (comparaison par valeur)
for field_name in kwargs:
if field_name in (
"start",
"end",
"start_time",
"end_time",
"published_at",
"updated_at",
"date",
"due_on",
"assigned_on",
"generated_at",
):
# Pour les dates/heures, comparer les objets directement
assert getattr(instance, field_name) == getattr(roundtrip_instance, field_name), (
f"Le champ {field_name} ne correspond pas après round-trip"
)
else:
# Pour les autres champs, comparer directement
assert getattr(instance, field_name) == getattr(roundtrip_instance, field_name), (
f"Le champ {field_name} ne correspond pas après round-trip"
)
def test_lesson_status_enum_values() -> None:
"""Vérifie que les valeurs de l'enum LessonStatus sont correctes.
:raises ValidationError: Si une valeur invalide est utilisée.
"""
# Test des valeurs valides
for status in [LessonStatus.NORMAL, LessonStatus.CANCELLED, LessonStatus.MOVED]:
assert isinstance(status.value, str)
Lesson(
id="test",
start=datetime(2024, 9, 6, 8, 0),
end=datetime(2024, 9, 6, 9, 0),
subject="Test",
status=status,
group=None,
content=None,
)
def test_school_event_kind_enum_values() -> None:
"""Vérifie que les valeurs de l'enum SchoolEventKind sont correctes.
:raises ValidationError: Si une valeur invalide est utilisée.
"""
# Test des valeurs valides
for kind in [SchoolEventKind.HOLIDAY, SchoolEventKind.PUBLIC_HOLIDAY]:
assert isinstance(kind.value, str)
SchoolEvent(
kind=kind,
label="Test",
from_date=date(2024, 9, 1),
to_date=date(2024, 9, 2),
)
def test_message_type_enum_values() -> None:
"""Vérifie que les valeurs de l'enum MessageType sont correctes.
:raises ValidationError: Si une valeur invalide est utilisée.
"""
# Test des valeurs valides
for msg_type in [MessageType.DISCUSSION, MessageType.INFORMATION, MessageType.SURVEY]:
assert isinstance(msg_type.value, str)
Message(
id="test",
type=msg_type,
title="Test",
content="Test",
author="Test",
date=datetime(2024, 9, 1, 8, 0),
read=False,
)
def test_agenda_change_type_enum_values() -> None:
"""Vérifie que les valeurs de l'enum AgendaChangeType sont correctes.
:raises ValidationError: Si une valeur invalide est utilisée.
"""
# Test des valeurs valides
for change_type in [
AgendaChangeType.ADDED,
AgendaChangeType.REMOVED,
AgendaChangeType.MODIFIED,
]:
assert isinstance(change_type.value, str)
if change_type == AgendaChangeType.ADDED:
AgendaChange(
type=change_type,
lesson=Lesson(
id="test",
start=datetime(2024, 9, 6, 8, 0),
end=datetime(2024, 9, 6, 9, 0),
subject="Test",
group=None,
content=None,
),
theoretical_lesson=None,
)
elif change_type == AgendaChangeType.REMOVED:
AgendaChange(
type=change_type,
lesson=None,
theoretical_lesson=TheoreticalLesson(
id="test",
day_of_week=0,
start_time=time(8, 0, 0),
end_time=time(9, 0, 0),
subject="Test",
),
)
else:
AgendaChange(
type=change_type,
lesson=Lesson(
id="test",
start=datetime(2024, 9, 6, 8, 0),
end=datetime(2024, 9, 6, 9, 0),
subject="Test",
group=None,
content=None,
),
theoretical_lesson=None,
)

View File

@@ -0,0 +1,409 @@
"""Tests d'immuabilité des modèles Pydantic figés.
Ce module vérifie que les modèles marqués comme figés (frozen=True) ne peuvent pas
être modifiés après construction, et que leurs collections sont bien des tuples.
"""
from __future__ import annotations
from datetime import date, datetime
from typing import Any
import pytest
from pydantic import BaseModel, ValidationError
from pronote_sync.models.agenda import HomeworkBlock, Lesson, LessonStatus
from pronote_sync.models.blog import BlogArticle, ExternalInfo
from pronote_sync.models.diff import AgendaChange, AgendaChangeType
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
from pronote_sync.models.xmpp import XmppMessage
@pytest.mark.parametrize(
"model_class, kwargs",
[
(
Lesson,
{
"id": "lesson-immutability",
"start": datetime(2024, 9, 6, 8, 0, 0),
"end": datetime(2024, 9, 6, 9, 30, 0),
"subject": "Mathématiques",
"teachers": ("M. Dupont", "Mme Martin"),
"rooms": ("Salle 101",),
"status": LessonStatus.NORMAL,
},
),
(
Homework,
{
"id": "hw-immutability",
"subject": "Physique",
"teachers": ("M. Martin",),
"due_on": date(2024, 9, 15),
"text": "Exercice à rendre",
},
),
(
AgendaChange,
{
"type": AgendaChangeType.ADDED,
"lesson": Lesson(
id="lesson-ref",
start=datetime(2024, 9, 6, 10, 0, 0),
end=datetime(2024, 9, 6, 11, 30, 0),
subject="Histoire",
teachers=("Mme Bernard",),
rooms=("Salle 201",),
group=None,
content=None,
),
},
),
(
XmppMessage,
{
"target_date": date(2024, 9, 6),
"synthesis": "Test immuabilité",
},
),
(
HomeworkBlock,
{
"kind": "due",
"date": date(2024, 9, 15),
"text": "Devoir à faire",
},
),
(
ExternalInfo,
{
"blog_articles": (
BlogArticle(
id="article-1",
title="Article 1",
url="https://example.com/1",
published_at=datetime(2024, 9, 1, 8, 0, 0),
updated_at=None,
category=None,
author=None,
content_html="<p>Contenu</p>",
content_text="Contenu",
),
),
"pronote_messages": (
Message(
id="msg-1",
type=MessageType.INFORMATION,
title="Message 1",
content="Message",
author="Auteur",
date=datetime(2024, 9, 1, 8, 0, 0),
read=False,
),
),
},
),
],
)
def test_frozen_model_rejects_field_assignment(
model_class: type[BaseModel], kwargs: dict[str, Any]
) -> None:
"""Vérifie qu'un modèle figé rejette toute tentative d'assignation de champ.
:param model_class: Classe du modèle figé.
:type model_class: type[BaseModel]
:param kwargs: Arguments de constructeur valides.
:type kwargs: dict[str, Any]
"""
instance: Any = model_class(**kwargs)
# Tentative d'assignation directe sur un champ
with pytest.raises(ValidationError) as exc_info:
if model_class == Lesson:
instance.subject = "Nouveau sujet"
elif model_class == Homework:
instance.subject = "Nouveau sujet"
elif model_class == AgendaChange:
instance.details = "Nouveaux détails"
elif model_class == XmppMessage:
instance.synthesis = "Nouvelle synthèse"
elif model_class == HomeworkBlock:
instance.text = "Nouveau texte"
elif model_class == ExternalInfo:
instance.other_info = ("Nouvelle info",)
error_msg = str(exc_info.value).lower()
assert "instance is frozen" in error_msg or "immutable" in error_msg
@pytest.mark.parametrize(
"model_class, kwargs, field_name, expected_type",
[
(
Lesson,
{
"id": "lesson-tuple",
"start": datetime(2024, 9, 6, 8, 0),
"end": datetime(2024, 9, 6, 9, 0),
"subject": "Test",
},
"teachers",
tuple,
),
(
Lesson,
{
"id": "lesson-tuple2",
"start": datetime(2024, 9, 6, 8, 0),
"end": datetime(2024, 9, 6, 9, 0),
"subject": "Test",
},
"rooms",
tuple,
),
(
Lesson,
{
"id": "lesson-tuple3",
"start": datetime(2024, 9, 6, 8, 0),
"end": datetime(2024, 9, 6, 9, 0),
"subject": "Test",
"homework_blocks": (),
},
"homework_blocks",
tuple,
),
(
Homework,
{
"id": "hw-tuple",
"subject": "Test",
"teachers": ("Prof 1",),
"due_on": date(2024, 9, 15),
"text": "Devoir",
},
"teachers",
tuple,
),
(
AgendaChange,
{
"type": AgendaChangeType.ADDED,
"lesson": Lesson(
id="ref",
start=datetime(2024, 9, 6, 8, 0),
end=datetime(2024, 9, 6, 9, 0),
subject="Test",
group=None,
content=None,
),
},
"details",
str,
),
(XmppMessage, {"target_date": date(2024, 9, 6)}, "homeworks", tuple),
(XmppMessage, {"target_date": date(2024, 9, 6)}, "changes", tuple),
(XmppMessage, {"target_date": date(2024, 9, 6)}, "messages", tuple),
(
ExternalInfo,
{"blog_articles": (), "pronote_messages": (), "other_info": ()},
"blog_articles",
tuple,
),
(
ExternalInfo,
{"blog_articles": (), "pronote_messages": (), "other_info": ()},
"pronote_messages",
tuple,
),
(
ExternalInfo,
{"blog_articles": (), "pronote_messages": (), "other_info": ()},
"other_info",
tuple,
),
],
)
def test_collection_fields_are_tuples(
model_class: type[BaseModel],
kwargs: dict[str, Any],
field_name: str,
expected_type: type[Any],
) -> None:
"""Vérifie que les champs de collection sont bien des tuples et non des listes.
:param model_class: Classe du modèle.
:type model_class: type[BaseModel]
:param kwargs: Arguments de constructeur valides.
:type kwargs: dict[str, Any]
:param field_name: Nom du champ de collection à vérifier.
:type field_name: str
:param expected_type: Type attendu (tuple).
:type expected_type: type
"""
instance = model_class(**kwargs)
field_value = getattr(instance, field_name)
assert isinstance(field_value, expected_type), (
f"Le champ {field_name} devrait être de type {expected_type.__name__}, "
f"got {type(field_value).__name__}"
)
@pytest.mark.parametrize(
"model_class, kwargs, field_name",
[
(
Lesson,
{
"id": "lesson-no-append",
"start": datetime(2024, 9, 6, 8, 0),
"end": datetime(2024, 9, 6, 9, 0),
"subject": "Test",
},
"teachers",
),
(
Lesson,
{
"id": "lesson-no-append2",
"start": datetime(2024, 9, 6, 8, 0),
"end": datetime(2024, 9, 6, 9, 0),
"subject": "Test",
},
"rooms",
),
(
Homework,
{
"id": "hw-no-append",
"subject": "Test",
"teachers": ("Prof 1",),
"due_on": date(2024, 9, 15),
"text": "Devoir",
},
"teachers",
),
(
ExternalInfo,
{"blog_articles": (), "pronote_messages": (), "other_info": ()},
"blog_articles",
),
(
ExternalInfo,
{"blog_articles": (), "pronote_messages": (), "other_info": ()},
"pronote_messages",
),
(
ExternalInfo,
{"blog_articles": (), "pronote_messages": (), "other_info": ()},
"other_info",
),
],
)
def test_collection_fields_reject_append(
model_class: type[BaseModel], kwargs: dict[str, Any], field_name: str
) -> None:
"""Vérifie que les champs de collection (tuples) rejettent la méthode append.
:param model_class: Classe du modèle.
:type model_class: type[BaseModel]
:param kwargs: Arguments de constructeur valides.
:type kwargs: dict[str, Any]
:param field_name: Nom du champ de collection.
:type field_name: str
"""
instance = model_class(**kwargs)
with pytest.raises(AttributeError) as exc_info:
collection = getattr(instance, field_name)
collection.append("new_item")
assert "'tuple' object has no attribute 'append'" in str(exc_info.value)
@pytest.mark.parametrize(
"model_class, kwargs, field_name, new_value",
[
(CalDAVSyncResult, {"status": CalDAVSyncStatus.SUCCESS}, "added", 5),
(CalDAVSyncResult, {"status": CalDAVSyncStatus.SUCCESS}, "updated", 3),
(CalDAVSyncResult, {"status": CalDAVSyncStatus.SUCCESS}, "removed", 2),
(
CalDAVSyncResult,
{"status": CalDAVSyncStatus.FAILED},
"errors",
["Erreur 1", "Erreur 2"],
),
],
)
def test_mutable_models_accept_field_assignment(
model_class: type[BaseModel],
kwargs: dict[str, Any],
field_name: str,
new_value: Any,
) -> None:
"""Vérifie que les modèles mutables (non figés) acceptent l'assignation de champs.
:param model_class: Classe du modèle mutable.
:type model_class: type[BaseModel]
:param kwargs: Arguments de constructeur valides.
:type kwargs: dict[str, Any]
:param field_name: Nom du champ à assigner.
:type field_name: str
:param new_value: Nouvelle valeur à assigner.
:type new_value: Any
"""
instance = model_class(**kwargs)
# Assignation du champ
setattr(instance, field_name, new_value)
# Vérification que la valeur a bien été assignée
assert getattr(instance, field_name) == new_value, (
f"Le champ {field_name} n'a pas été correctement assigné"
)
@pytest.mark.parametrize(
"model_class, kwargs",
[
(
PronoteData,
{
"target_date": date(2024, 9, 6),
"generated_at": datetime(2024, 9, 6, 8, 0, 0),
},
),
(CalDAVSyncResult, {"status": CalDAVSyncStatus.SUCCESS}),
],
)
def test_mutable_models_reject_frozen_assignment(
model_class: type[BaseModel], kwargs: dict[str, Any]
) -> None:
"""Vérifie que les modèles mutables peuvent être modifiés mais pas rendus figés.
Note: Les modèles mutables comme PronoteData et CalDAVSyncResult n'ont pas
ConfigDict(frozen=True), donc ils peuvent être modifiés.
:param model_class: Classe du modèle mutable.
:type model_class: type[BaseModel]
:param kwargs: Arguments de constructeur valides.
:type kwargs: dict[str, Any]
"""
from pronote_sync.models.pronote import PronoteData
instance: Any = model_class(**kwargs)
# Assignation normale doit fonctionner
if model_class == PronoteData:
instance.lessons.append("dummy")
else:
instance.added = 10
# Vérification que l'instance n'est pas figée
assert not hasattr(instance.model_config, "frozen") or not instance.model_config.get(
"frozen", False
)

View File

@@ -0,0 +1,287 @@
"""Tests d'invariants et de validation des modèles Pydantic.
Ce module vérifie que les validateurs de modèle rejettent les données invalides
conformément aux spécifications (ex: ordre des dates, plages de valeurs).
"""
from __future__ import annotations
from datetime import date, datetime, time
import pytest
from pydantic import ValidationError
from pronote_sync.models.agenda import (
Lesson,
SchoolEvent,
SchoolEventKind,
TheoreticalLesson,
)
from pronote_sync.models.diff import AgendaChange, AgendaChangeType
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
class TestTheoreticalLessonDayOfWeek:
"""Tests de validation du champ day_of_week pour TheoreticalLesson."""
def test_day_of_week_7_invalid(self) -> None:
"""Vérifie que day_of_week=7 lève une ValidationError."""
with pytest.raises(ValidationError):
TheoreticalLesson(
id="test",
day_of_week=7,
start_time=time(8, 0, 0),
end_time=time(9, 30, 0),
subject="Mathématiques",
)
def test_day_of_week_minus_1_invalid(self) -> None:
"""Vérifie que day_of_week=-1 lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
TheoreticalLesson(
id="test",
day_of_week=-1,
start_time=time(8, 0, 0),
end_time=time(9, 30, 0),
subject="Mathématiques",
)
assert any(
"greater than or equal to 0" in str(error).lower() or "le 6" in str(error).lower()
for error in exc_info.value.errors()
)
def test_day_of_week_0_valid(self) -> None:
"""Vérifie que day_of_week=0 (lundi) est valide."""
instance = TheoreticalLesson(
id="test",
day_of_week=0,
start_time=time(8, 0, 0),
end_time=time(9, 30, 0),
subject="Mathématiques",
)
assert instance.day_of_week == 0
def test_day_of_week_6_valid(self) -> None:
"""Vérifie que day_of_week=6 (dimanche) est valide."""
instance = TheoreticalLesson(
id="test",
day_of_week=6,
start_time=time(8, 0, 0),
end_time=time(9, 30, 0),
subject="Mathématiques",
)
assert instance.day_of_week == 6
class TestLessonTimeOrder:
"""Tests de validation de l'ordre temporel pour Lesson."""
def test_lesson_end_before_start_invalid(self) -> None:
"""Vérifie que end <= start lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
Lesson(
id="test",
start=datetime(2024, 9, 6, 10, 0, 0),
end=datetime(2024, 9, 6, 9, 0, 0), # end avant start
subject="Mathématiques",
group=None,
content=None,
)
assert any(
"end doit être postérieur à start" in str(error) for error in exc_info.value.errors()
)
def test_lesson_end_after_start_valid(self) -> None:
"""Vérifie que end > start est valide."""
instance = Lesson(
id="test",
start=datetime(2024, 9, 6, 8, 0, 0),
end=datetime(2024, 9, 6, 9, 30, 0),
subject="Mathématiques",
group=None,
content=None,
)
assert instance.end > instance.start
class TestTheoreticalLessonTimeOrder:
"""Tests de validation de l'ordre temporel pour TheoreticalLesson."""
def test_theoretical_lesson_end_time_before_start_time_invalid(self) -> None:
"""Vérifie que end_time <= start_time lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
TheoreticalLesson(
id="test",
day_of_week=0,
start_time=time(9, 0, 0),
end_time=time(8, 0, 0), # end_time avant start_time
subject="Mathématiques",
)
assert any(
"end_time doit être postérieur à start_time" in str(error)
for error in exc_info.value.errors()
)
def test_theoretical_lesson_end_time_after_start_time_valid(self) -> None:
"""Vérifie que end_time > start_time est valide."""
instance = TheoreticalLesson(
id="test",
day_of_week=0,
start_time=time(8, 0, 0),
end_time=time(9, 30, 0),
subject="Mathématiques",
)
assert instance.end_time > instance.start_time
class TestSchoolEventDateOrder:
"""Tests de validation de l'ordre des dates pour SchoolEvent."""
def test_school_event_to_date_before_from_date_invalid(self) -> None:
"""Vérifie que to_date < from_date lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Test",
from_date=date(2024, 10, 20),
to_date=date(2024, 10, 19), # to_date avant from_date
)
assert any(
"to_date doit être postérieur ou égal à from_date" in str(error)
for error in exc_info.value.errors()
)
def test_school_event_to_date_equal_from_date_valid(self) -> None:
"""Vérifie que to_date == from_date est valide."""
instance = SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Test",
from_date=date(2024, 10, 20),
to_date=date(2024, 10, 20), # égal
)
assert instance.to_date >= instance.from_date
def test_school_event_to_date_after_from_date_valid(self) -> None:
"""Vérifie que to_date > from_date est valide."""
instance = SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Test",
from_date=date(2024, 10, 20),
to_date=date(2024, 10, 25),
)
assert instance.to_date >= instance.from_date
class TestAgendaChangeConsistency:
"""Tests de cohérence des changements d'agenda."""
def test_agenda_change_added_without_lesson_invalid(self) -> None:
"""Vérifie que type=ADDED avec lesson=None lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
AgendaChange(type=AgendaChangeType.ADDED, lesson=None, theoretical_lesson=None)
assert any(
"lesson est requis pour le type" in str(error) for error in exc_info.value.errors()
)
def test_agenda_change_removed_without_theoretical_lesson_invalid(self) -> None:
"""Vérifie que type=REMOVED avec theoretical_lesson=None lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
AgendaChange(
type=AgendaChangeType.REMOVED,
lesson=None,
theoretical_lesson=None,
)
assert any(
"theoretical_lesson est requis pour le type" in str(error)
for error in exc_info.value.errors()
)
def test_agenda_change_modified_without_lesson_invalid(self) -> None:
"""Vérifie que type=MODIFIED avec lesson=None lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
AgendaChange(type=AgendaChangeType.MODIFIED, lesson=None, theoretical_lesson=None)
assert any(
"lesson est requis pour le type" in str(error) for error in exc_info.value.errors()
)
def test_agenda_change_added_with_lesson_valid(self) -> None:
"""Vérifie que type=ADDED avec lesson=<valide> est valide."""
lesson = Lesson(
id="lesson-valid",
start=datetime(2024, 9, 6, 8, 0, 0),
end=datetime(2024, 9, 6, 9, 30, 0),
subject="Mathématiques",
group=None,
content=None,
)
instance = AgendaChange(
type=AgendaChangeType.ADDED,
lesson=lesson,
theoretical_lesson=None,
)
assert instance.type == AgendaChangeType.ADDED
assert instance.lesson is not None
def test_agenda_change_removed_with_theoretical_lesson_valid(self) -> None:
"""Vérifie que type=REMOVED avec theoretical_lesson=<valide> est valide."""
theoretical_lesson = TheoreticalLesson(
id="theo-lesson-valid",
day_of_week=0,
start_time=time(8, 0, 0),
end_time=time(9, 30, 0),
subject="Mathématiques",
)
instance = AgendaChange(
type=AgendaChangeType.REMOVED,
lesson=None,
theoretical_lesson=theoretical_lesson,
)
assert instance.type == AgendaChangeType.REMOVED
assert instance.theoretical_lesson is not None
def test_agenda_change_modified_with_lesson_valid(self) -> None:
"""Vérifie que type=MODIFIED avec lesson=<valide> est valide."""
lesson = Lesson(
id="lesson-valid-mod",
start=datetime(2024, 9, 6, 10, 0, 0),
end=datetime(2024, 9, 6, 11, 30, 0),
subject="Physique",
group=None,
content=None,
)
instance = AgendaChange(
type=AgendaChangeType.MODIFIED,
lesson=lesson,
theoretical_lesson=None,
)
assert instance.type == AgendaChangeType.MODIFIED
assert instance.lesson is not None
class TestCalDAVSyncResultInvariants:
"""Tests de validation des champs numériques de CalDAVSyncResult."""
def test_caldav_sync_result_added_negative_invalid(self) -> None:
"""Vérifie que added=-1 lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
CalDAVSyncResult(status=CalDAVSyncStatus.SUCCESS, added=-1, updated=0, removed=0)
assert any("greater than or equal to 0" in str(error) for error in exc_info.value.errors())
def test_caldav_sync_result_updated_negative_invalid(self) -> None:
"""Vérifie que updated=-1 lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
CalDAVSyncResult(status=CalDAVSyncStatus.SUCCESS, added=0, updated=-1, removed=0)
assert any("greater than or equal to 0" in str(error) for error in exc_info.value.errors())
def test_caldav_sync_result_removed_negative_invalid(self) -> None:
"""Vérifie que removed=-1 lève une ValidationError."""
with pytest.raises(ValidationError) as exc_info:
CalDAVSyncResult(status=CalDAVSyncStatus.SUCCESS, added=0, updated=0, removed=-1)
assert any("greater than or equal to 0" in str(error) for error in exc_info.value.errors())
def test_caldav_sync_result_zero_values_valid(self) -> None:
"""Vérifie que les valeurs à zéro sont valides."""
instance = CalDAVSyncResult(status=CalDAVSyncStatus.SUCCESS, added=0, updated=0, removed=0)
assert instance.added == 0
assert instance.updated == 0
assert instance.removed == 0

View File

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

View File

@@ -0,0 +1,128 @@
"""Tests unitaires pour les utilitaires de masquage des secrets.
Ce module vérifie que les fonctions de masquage des URLs, des en-têtes
et des secrets dans les textes fonctionnent correctement, sans fuite
d'informations sensibles.
"""
from __future__ import annotations
from pronote_sync.utils.redaction import redact_exception, redact_secrets, redact_url
def test_redact_url_with_sensitive_query_param() -> None:
"""Vérifie que les paramètres sensibles dans une URL sont masqués.
:return: None
"""
url = "https://example.fr/ical?icalsecurise=TOKEN123" # pragma: allowlist secret
redacted = redact_url(url)
assert "TOKEN123" not in redacted
assert "REDACTED" in redacted
def test_redact_url_with_embedded_auth() -> None:
"""Vérifie que les informations d'authentification intégrées dans une URL sont masquées.
:return: None
"""
url = "https://user:pass@host/path" # pragma: allowlist secret
redacted = redact_url(url)
assert "user" not in redacted
assert "pass" not in redacted
assert "REDACTED" in redacted
def test_redact_url_with_uppercase_scheme() -> None:
"""Vérifie que les URLs avec un schéma en majuscules sont traitées correctement.
:return: None
"""
url = "HTTPS://user:pass@host/path" # pragma: allowlist secret
redacted = redact_url(url)
assert "user" not in redacted
assert "pass" not in redacted
assert "REDACTED" in redacted
def test_redact_secrets_authorization_bearer() -> None:
"""Vérifie que les en-têtes Authorization: Bearer sont masqués.
:return: None
"""
text = "Authorization: Bearer mytoken123" # pragma: allowlist secret
redacted = redact_secrets(text)
assert "mytoken123" not in redacted
assert "REDACTED" in redacted
def test_redact_secrets_proxy_authorization_basic() -> None:
"""Vérifie que les en-têtes Proxy-Authorization: Basic sont masqués.
:return: None
"""
text = "Proxy-Authorization: Basic dXNlcjpwYXNz" # pragma: allowlist secret
redacted = redact_secrets(text)
assert "dXNlcjpwYXNz" not in redacted
assert "REDACTED" in redacted
def test_redact_secrets_lowercase_authorization() -> None:
"""Vérifie que les en-têtes authorization en minuscules sont masqués.
:return: None
"""
text = "authorization: Bearer tok" # pragma: allowlist secret
redacted = redact_secrets(text)
assert "tok" not in redacted
assert "REDACTED" in redacted
def test_redact_secrets_text_key_value() -> None:
"""Vérifie que les affectations clé=valeur dans du texte sont masquées.
:return: None
"""
text = "password=mysecret" # pragma: allowlist secret
redacted = redact_secrets(text)
assert "mysecret" not in redacted
assert "REDACTED" in redacted
def test_redact_secrets_json_key_value() -> None:
"""Vérifie que les secrets dans du JSON sont masqués.
:return: None
"""
text = '{"api_key": "myapikey"}' # pragma: allowlist secret
redacted = redact_secrets(text)
assert "myapikey" not in redacted
assert "REDACTED" in redacted
def test_redact_exception() -> None:
"""Vérifie que les exceptions contenant des secrets sont masquées.
:return: None
"""
exc = Exception("Authorization: Bearer leaked_token") # pragma: allowlist secret
redacted_msg = redact_exception(exc)
assert "leaked_token" not in redacted_msg
assert "REDACTED" in redacted_msg
def test_redact_url_preserves_host_and_path() -> None:
"""Vérifie que l'hôte et le chemin d'une URL sont préservés après masquage.
:return: None
"""
url = "https://user:pass@example.fr/path?icalsecurise=tok" # pragma: allowlist secret
redacted = redact_url(url)
assert "example.fr" in redacted
assert "/path" in redacted
assert "user" not in redacted
assert "pass" not in redacted
assert "tok" not in redacted
# Ensure trailing newline

16
tests/unit/test_smoke.py Normal file
View File

@@ -0,0 +1,16 @@
"""Test de fumée : vérifie que le package est importable.
Ce test garantit que ``pytest`` collecte au moins un test et retourne 0,
satisfaisant le critère d'acceptation de M1. Les tests détaillés relèvent
de M13.
"""
import pronote_sync
def test_package_importable() -> None:
"""Vérifie que le package ``pronote_sync`` est importable sans erreur.
:return: None
"""
assert pronote_sync is not None